SS-350: sql-server: seed offset_committed from resume LSN on startup (#37680)

The progress operator left offset_committed at 0 until the first resume
upper arrived, so ingestion lag read as the full upstream LSN during the
initial snapshot. Initialize it from the resumption LSN, or max upstream
LSN, if snapshotting. Initialize offset_known concurrently to further
reduce the gap where these 2 could result in bogus lag calculation.

Author: martykulma

src/storage/src/source/sql_server.rs

@@ -32,7 +32,7 @@ use timely::dataflow::operators::Concat;
 use timely::dataflow::operators::core::Partition;
 use timely::dataflow::operators::vec::{Map, ToStream};
 use timely::dataflow::{Scope, StreamVec};
-use timely::progress::Antichain;
+use timely::progress::{Antichain, Timestamp};
 
 use crate::healthcheck::{HealthStatusMessage, HealthStatusUpdate, StatusNamespace};
 use crate::source::RawSourceCreationConfig;
@@ -58,6 +58,20 @@ struct SourceOutputInfo {
     initial_lsn: Lsn,
 }
 
+impl SourceOutputInfo {
+    /// The [`Lsn`] this output resumes reading from, or the provided fallback [`Lsn`].
+    ///
+    /// Panics if `resume_upper` is empty, which would mean the output has no
+    /// resumption point at all.
+    fn resume_lsn_or(&self, fallback: Lsn) -> Lsn {
+        match self.resume_upper.as_option() {
+            Some(lsn) if *lsn != Lsn::minimum() => *lsn,
+            Some(_) => fallback,
+            None => panic!("resume_upper has at least one value"),
+        }
+    }
+}
+
 #[derive(Debug, Clone, thiserror::Error)]
 pub enum ReplicationError {
     #[error(transparent)]

src/storage/src/source/sql_server/progress.rs

@@ -85,6 +85,11 @@ pub(crate) fn render<'scope>(
                 }
                 return Ok(());
             }
+
+            // Retrieve the latest upstream LSN eagerly to ensure the lag calculation
+            // (offset_known - offset_committed) is non-negative. Statistics represents these as
+            // uint8, which would cause the calculation to underflow for the brief period between
+            // setting offset_committed here and offset_known further below.
             let conn_config = connection
                 .resolve_config(
                     &config.config.connection_context.secrets_reader,
@@ -93,6 +98,32 @@ pub(crate) fn render<'scope>(
                 )
                 .await?;
             let mut client = mz_sql_server_util::Client::connect(conn_config).await?;
+            // increment here to match known_offset calculation below
+            let next_upstream_lsn: Lsn = get_max_lsn(&mut client).await?.increment();
+
+            // Seed `offset_committed` from the resumption LSN, or if not set, from the
+            // upstream's current max LSN. Otherwise, it stays at the default 0 until the
+            // initial snapshot durably commits and the first resume upper arrives, which
+            // for a large snapshot can be a long time. During that window the ingestion-lag
+            // calculation subtracts 0 from the (large) upstream LSN and reports an
+            // enormous, bogus lag.
+            //
+            // This defaults to upstream's current max offset instead of `initial_lsn` because
+            // `initial_lsn` can be ahead of the value returned by `sys.fn_cdc_get_max_lsn`. This
+            // is a very obscure edge case where a user has a CDC enabled table, creates a new one
+            // and configures a source for at least the second table immediately after, without
+            // performing any DML operations.
+            let mut max_committed_lsn = outputs
+                .values()
+                // resume_lsn_or will panic if info resume_upper is empty
+                .map(|info| info.resume_lsn_or(next_upstream_lsn))
+                .min()
+                .unwrap_or(next_upstream_lsn);
+
+            for stat in config.statistics.values() {
+                stat.set_offset_known(next_upstream_lsn.abbreviate());
+                stat.set_offset_committed(max_committed_lsn.abbreviate());
+            }
 
 
             // Terminate the progress probes if a restore has happened. Replication operator will
@@ -194,8 +225,14 @@ pub(crate) fn render<'scope>(
                                 }
                             }
                         }
+                        // Never regress below the seeded resumption LSN. During the initial
+                        // snapshot the resume upper sits at the minimum, which would otherwise
+                        // drag the committed offset back to 0 and reintroduce the bogus lag.
+                        if *committed_upper > max_committed_lsn {
+                            max_committed_lsn = *committed_upper;
+                        }
                         for stat in config.statistics.values() {
-                            stat.set_offset_committed(committed_upper.abbreviate());
+                            stat.set_offset_committed(max_committed_lsn.abbreviate());
                         }
                     }
                 };

src/storage/src/source/sql_server/replication.rs

@@ -391,14 +391,10 @@ pub(crate) fn render<'scope>(
             // Resumption point is the minimum LSN that has been observed per capture instance.
             let mut resume_lsns = BTreeMap::new();
             for src_info in outputs.values() {
-                let resume_lsn = match src_info.resume_upper.as_option() {
-                    Some(lsn) if *lsn != Lsn::minimum() => *lsn,
-                    // initial_lsn is the max lsn observed, but the resume lsn
-                    // is the next lsn that should be read.  After a snapshot, initial_lsn
-                    // has been read, so replication will start at the next available lsn.
-                    Some(_) => src_info.initial_lsn.increment(),
-                    None => panic!("resume_upper has at least one value"),
-                };
+                // initial_lsn is the max lsn observed, but the resume lsn
+                // is the next lsn that should be read.  After a snapshot, initial_lsn
+                // has been read, so replication will start at the next available lsn.
+                let resume_lsn = src_info.resume_lsn_or(src_info.initial_lsn.increment());
                 resume_lsns.entry(Arc::clone(&src_info.capture_instance))
                     .and_modify(|existing| *existing = std::cmp::min(*existing, resume_lsn))
                     .or_insert(resume_lsn);

test/sql-server-cdc/72-ingestion-lag-when-idle.td

@@ -0,0 +1,133 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+#
+# Regression test: a freshly created source must report a near-zero ingestion
+# lag, even while its upstream database is idle.
+#
+# The two offset statistics must be initialized from a common validated
+# upstream point. Leaving `offset_committed` at 0 after `offset_known` advances
+# produces an enormous lag. Publishing the full resumption LSN before the known
+# frontier catches up reverses the ordering and makes the same lag calculation
+# (`offset_known - offset_committed`) underflow.
+#
+# A newly enabled capture instance can start ahead of the database-wide CDC
+# maximum. Purification must use that higher start LSN as the source's resumption
+# point. The progress probe still observes the lower database maximum. We stop
+# this database's capture job before enabling the target table so the two LSNs
+# cannot converge and mask the invalid statistics.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server split-lines=false
+IF EXISTS (SELECT name FROM sys.databases WHERE name = N'lag_test')
+BEGIN
+    ALTER DATABASE lag_test SET SINGLE_USER WITH ROLLBACK IMMEDIATE;
+    DROP DATABASE lag_test;
+END;
+
+$ sql-server-execute name=sql-server
+CREATE DATABASE lag_test COLLATE Latin1_General_100_CI_AI_SC_UTF8;
+USE lag_test;
+EXEC sys.sp_cdc_enable_db;
+ALTER DATABASE lag_test SET ALLOW_SNAPSHOT_ISOLATION ON;
+
+$ sql-server-execute name=sql-server
+USE lag_test;
+CREATE TABLE lag_seed (f1 VARCHAR(20));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'lag_seed', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO lag_seed VALUES ('seed');
+
+# Establish a nonzero database maximum, then freeze it. Enabling `lag_t` after
+# this point gives its capture instance a later start LSN.
+$ sql-server-execute name=sql-server split-lines=false
+WAITFOR DELAY '00:00:20';
+
+$ sql-server-execute name=sql-server
+USE lag_test;
+EXEC sys.sp_cdc_stop_job @job_type = 'capture';
+CREATE TABLE lag_t (f1 VARCHAR(20));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'lag_t', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO lag_t VALUES ('one'), ('two'), ('three');
+
+# This is the upstream state the test depends on. If SQL Server changes its CDC
+# behavior, fail here rather than attributing a later result to Materialize.
+$ sql-server-execute name=sql-server split-lines=false
+IF NOT EXISTS (
+    SELECT 1
+    FROM lag_test.cdc.change_tables
+    WHERE capture_instance = 'dbo_lag_t'
+      AND start_lsn > lag_test.sys.fn_cdc_get_max_lsn()
+)
+    THROW 50000, 'expected capture start LSN above database maximum', 1;
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET storage_statistics_collection_interval = 1000;
+ALTER SYSTEM SET storage_statistics_interval = 2000;
+
+> CREATE SECRET IF NOT EXISTS mspass AS '${arg.default-sql-server-password}'
+
+> CREATE CONNECTION msconn TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE lag_test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET mspass
+  );
+
+> CREATE CLUSTER lag_cluster SIZE '${arg.default-replica-size}'
+> CREATE SOURCE lag_source IN CLUSTER lag_cluster FROM SQL SERVER CONNECTION msconn;
+> CREATE TABLE lag_t FROM SOURCE lag_source (REFERENCE dbo.lag_t);
+
+# We intentionally do not wait for the snapshot to become queryable. On an idle
+# database the source frontier only advances once SQL Server advances its max
+# LSN, which can take minutes. That frontier advance is also what would fix a
+# buggy `offset_committed`, so waiting for it would mask the bug. Instead we
+# assert on the statistics, which are populated within seconds regardless of the
+# frontier.
+
+# Ensure statistics for the source's replica exist before pinning the replica
+# id below, which does not retry on a missing row.
+> SELECT COUNT(*) > 0
+  FROM
+    mz_cluster_replicas cr,
+    mz_internal.mz_source_statistics u,
+    mz_sources s
+  WHERE
+    cr.id = u.replica_id AND s.name = 'lag_source' AND u.id = s.id
+true
+
+$ set-from-sql var=replica_id
+SELECT cr.id
+  FROM
+    mz_clusters c,
+    mz_cluster_replicas cr,
+    mz_internal.mz_source_statistics u,
+    mz_sources s
+  WHERE
+    c.name = 'lag_cluster' AND c.id = cr.cluster_id AND cr.id = u.replica_id
+    AND s.name = 'lag_source' AND u.id = s.id
+  ORDER BY cr.id
+  LIMIT 1
+
+# The reported committed offset must not exceed the first known upstream offset,
+# even though the actual resumption LSN does. The second expression is the
+# ingestion-lag calculation used by customers. Since both columns are `uint8`,
+# invalid ordering makes the subtraction itself fail instead of reporting lag.
+$ set-sql-timeout duration=60s
+
+> SELECT
+    u.offset_committed > 0,
+    u.offset_known >= u.offset_committed,
+    u.offset_known - u.offset_committed < 1000000
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics u ON s.id = u.id
+  WHERE s.name = 'lag_source' AND u.replica_id = '${replica_id}'
+true true true

test/sql-server-cdc/73-ingestion-lag-during-connection-delay.td

@@ -0,0 +1,66 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Hold the first runtime connection open and verify that the ingestion-lag
+# calculation remains valid before `offset_known` is initialized.
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password}
+
+$ sql-server-execute name=sql-server
+USE test;
+CREATE TABLE t73 (f1 VARCHAR(20));
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't73', @role_name = 'SA', @supports_net_changes = 0;
+INSERT INTO t73 VALUES ('one');
+
+$ http-request method=POST url=http://toxiproxy:8474/proxies content-type=application/json
+{
+  "name": "sql-server",
+  "listen": "0.0.0.0:1433",
+  "upstream": "sql-server:1433",
+  "enabled": true
+}
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET storage_statistics_collection_interval = 1000;
+ALTER SYSTEM SET storage_statistics_interval = 2000;
+
+> CREATE CLUSTER storage REPLICAS ()
+> CREATE SECRET pass AS '${arg.default-sql-server-password}'
+> CREATE CONNECTION conn TO SQL SERVER (
+    HOST 'toxiproxy',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET pass
+  )
+> CREATE SOURCE s IN CLUSTER storage
+  FROM SQL SERVER CONNECTION conn
+  FOR TABLES (dbo.t73)
+
+# Stall the first connection past several statistics intervals.
+$ http-request method=POST url=http://toxiproxy:8474/proxies/sql-server/toxics content-type=application/json
+{
+  "name": "startup-delay",
+  "type": "latency",
+  "attributes": { "latency": 30000 }
+}
+
+> CREATE CLUSTER REPLICA storage.r1 SIZE = 'scale=1,workers=1'
+
+# The correct state remains zero, so there is no nonzero value to poll for.
+$ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration="5s"
+
+$ set-sql-timeout duration=10s
+
+> SELECT u.offset_known - u.offset_committed
+  FROM mz_sources s
+  JOIN mz_internal.mz_source_statistics u ON s.id = u.id
+  WHERE s.name = 's'
+0

test/sql-server-cdc/mzcompose.py

@@ -29,6 +29,7 @@
 from materialize.mzcompose.services.sql_server import SqlServer
 from materialize.mzcompose.services.test_certs import TestCerts
 from materialize.mzcompose.services.testdrive import Testdrive
+from materialize.mzcompose.services.toxiproxy import Toxiproxy
 
 TLS_CONF_PATH = MZ_ROOT / "test" / "sql-server-cdc" / "tls-mssconfig.conf"
 
@@ -41,6 +42,7 @@
     ),
     Testdrive(),
     TestCerts(),
+    Toxiproxy(),
     SqlServer(
         volumes_extra=[
             "secrets:/var/opt/mssql/certs",
@@ -122,10 +124,18 @@ def workflow_cdc(c: Composition, parser: WorkflowArgumentParser) -> None:
     c.rm("sql-server")
     c.kill("materialized")
     c.rm("materialized")
+    c.kill("toxiproxy")
+    c.rm("toxiproxy")
 
     # must start test-certs, otherwise the certificates needed by sql-server may not be available
     # in the secrets volume when it starts up
-    c.up("materialized", "test-certs", "sql-server", Service("testdrive", idle=True))
+    c.up(
+        "materialized",
+        "test-certs",
+        "sql-server",
+        "toxiproxy",
+        Service("testdrive", idle=True),
+    )
     seed = random.getrandbits(16)
 
     ssl_ca = c.exec(