adapter: linearize a read-then-write that matched no rows

The OCC path linearizes its `as_of` before subscribing, but the subscribe
then follows Persist, and Persist runs ahead of the timestamp oracle. Group
commit appends to the txns shard first, which makes the write readable and
advances the table's upper, and calls `oracle.apply_write` only after.

A read-then-write whose selection consolidates to empty inside that window
concluded from state no oracle-timestamped read could reach yet, and
returned immediately because it had nothing to write. A
strict-serializable read issued after that response then still saw the
rows the response said were not there. Real time orders the read last and
no serial order explains the history.

Only the zero-row exits are exposed. A write is linearized for free,
because group commit answers the writer after applying the timestamp, and
the coordinator's path takes its read timestamp from the oracle and so
cannot observe state the oracle has not reached.

`OccOutcome` now separates the two cases rather than encoding "nothing to
write" as a `Committed` with no timestamp. `NoRowsMatched` carries the
timestamp emptiness was concluded at, and the caller waits for the oracle
to reach it before responding. The timestamp is absent only for a
selection that reads no persisted state, which has nothing to linearize
against. Waiting is what the existing `ensure_read_linearized` does, and
it is preferable to applying the write timestamp ourselves: forcing the
oracle forward off another writer's timestamp would stall subsequent reads
of unrelated collections until they caught up.

The `group_commit_before_apply_write` failpoint holds the window open, and
is the same one a second `environmentd` opens on its own with no ordering
against local Persist visibility.

Two mzcompose tests come with this.
`workflow_test_occ_zero_row_write_linearization` is the regression test.
It parks the winning DELETE inside the window, confirms from a
serializable read that the delete is visible in Persist and from a
strict-serializable read that the oracle cannot serve it yet, and only
then runs the UPDATE that has to report zero rows. Attempts where the
subscribe reported progress before the append, told apart by the write
conflict count, prove nothing and are retried.

`workflow_mixed_mode_read_then_write` is a guard rather than a regression
test. The dyncfg is sampled once per process, so a rolling restart can
leave one instance sequencing on the coordinator behind in-process write
locks and another under OCC. Those locks never excluded a second process,
so this is not new, but what keeps the two from corrupting a row is the
catalog fence, and nothing states that. The test asserts the fence keeps
the commit windows disjoint, so it goes red if that stops holding.

Author: aljoscha

ci/nightly/pipeline.template.yml

@@ -2395,6 +2395,17 @@ steps:
     agents:
       queue: hetzner-aarch64-4cpu-8gb
 
+  - id: occ-mixed-mode-read-then-write
+    label: "Read-then-write sequencing split across two environmentd processes"
+    depends_on: build-aarch64
+    timeout_in_minutes: 30
+    plugins:
+      - ./ci/plugins/mzcompose:
+          composition: txn-wal-fencing
+          run: mixed-mode-read-then-write
+    agents:
+      queue: hetzner-aarch64-4cpu-8gb
+
   - group: "Copy"
     key: copy
     steps:

src/adapter/src/coord/appends.rs

@@ -589,6 +589,16 @@ impl GroupCommitter {
         let now: Timestamp = (self.now)().into();
         crate::coord::timeline::check_runaway_write_ts(&now, write_ts.timestamp);
 
+        // The append above is already readable in Persist and has advanced the
+        // table's upper, while no oracle-timestamped read can reach it until
+        // the line below. Anything concluding from a read that follows Persist
+        // rather than the oracle has to cope with this window, so a test can
+        // hold it open here. Every txns-shard write parks here while armed,
+        // including the keepalives that advance table uppers, so arm it with a
+        // bounded `sleep` rather than a `pause`. Used by
+        // workflow_test_occ_zero_row_write_linearization.
+        fail::fail_point!("group_commit_before_apply_write");
+
         self.oracle.apply_write(write_ts.timestamp).await;
 
         TxnsWriteAttempt::Applied

src/adapter/src/frontend_read_then_write.rs

@@ -171,10 +171,22 @@ impl FrontendWriteAttemptState {
 
 /// What the OCC loop produced.
 enum OccOutcome {
-    /// The write is durable at `write_ts`, or there was nothing to write.
+    /// The write is durable at `write_ts`.
     Committed {
         response: ExecuteResponse,
-        write_ts: Option<Timestamp>,
+        write_ts: Timestamp,
+    },
+    /// The selection was empty, so there was nothing to write.
+    ///
+    /// `observed_ts` is the timestamp emptiness was concluded at, and is
+    /// `None` only when the selection reads no persisted state. When it is
+    /// `Some`, the caller must linearize against it before responding: the
+    /// subscribe follows Persist, which runs ahead of the oracle, so the
+    /// emptiness can be concluded from state no oracle-timestamped read can
+    /// reach yet.
+    NoRowsMatched {
+        response: ExecuteResponse,
+        observed_ts: Option<Timestamp>,
     },
     /// Diffs from a selection that reads no persisted state. The subscribe ran
     /// to completion, so they are frontier-independent and the caller chooses
@@ -505,11 +517,27 @@ impl PeekClient {
         // waiting for our write to commit.
         let response = match result {
             Ok(OccOutcome::Committed { response, write_ts }) => {
-                if let Some(write_ts) = write_ts {
-                    session.apply_write(write_ts);
-                }
+                session.apply_write(write_ts);
                 Ok(response)
             }
+            Ok(OccOutcome::NoRowsMatched {
+                response,
+                observed_ts,
+            }) => {
+                // A write would have linearized this for us, because group
+                // commit advances the oracle before it answers. With nothing
+                // to write we have to do it ourselves, or we report an empty
+                // selection from state a later strict-serializable read cannot
+                // see yet, and that read finds the rows we said were not
+                // there.
+                match observed_ts {
+                    Some(observed_ts) => self
+                        .ensure_read_linearized(&timeline, observed_ts)
+                        .await
+                        .map(|()| response),
+                    None => Ok(response),
+                }
+            }
             Ok(OccOutcome::Blind { response, diffs }) => {
                 match self
                     .submit_blind_write(
@@ -840,8 +868,10 @@ impl PeekClient {
     /// are frontier-independent. Those are returned as [`OccOutcome::Blind`]
     /// for the caller to submit or buffer, and this never writes them.
     ///
-    /// Read linearization is the caller's responsibility: `as_of` must
-    /// already be linearized (oracle read_ts >= `as_of`) on entry. See
+    /// Read linearization is the caller's responsibility, on both ends.
+    /// `as_of` must already be linearized (oracle read_ts >= `as_of`) on
+    /// entry, and an [`OccOutcome::NoRowsMatched`] carrying an `observed_ts`
+    /// must be linearized against it before the response goes out. See
     /// `ensure_read_linearized` at the call site.
     ///
     /// Returns `(retry_count, result)` so the caller can record OCC retry
@@ -893,9 +923,9 @@ impl PeekClient {
                     // flatten to `Timestamp::MIN` for `consolidate_updates`.
                     state.consolidate(Timestamp::MIN);
                     if state.all_diffs.is_empty() {
-                        break Ok(OccOutcome::Committed {
+                        break Ok(OccOutcome::NoRowsMatched {
                             response: build_no_rows_response(&kind),
-                            write_ts: None,
+                            observed_ts: None,
                         });
                     }
                     let success_response = match self.build_success_response(
@@ -940,10 +970,10 @@ impl PeekClient {
                                     &table_desc,
                                 ) {
                                     ProcessResult::Continue { .. } => {}
-                                    ProcessResult::NoRowsMatched => {
-                                        break Some(Ok(OccOutcome::Committed {
+                                    ProcessResult::NoRowsMatched { observed_ts } => {
+                                        break Some(Ok(OccOutcome::NoRowsMatched {
                                             response: build_no_rows_response(&kind),
-                                            write_ts: None,
+                                            observed_ts: Some(observed_ts),
                                         }));
                                     }
                                     ProcessResult::Error(e) => {
@@ -1036,7 +1066,7 @@ impl PeekClient {
                             // fires off the cleanup message.
                             break Ok(OccOutcome::Committed {
                                 response: success_response,
-                                write_ts: Some(timestamp),
+                                write_ts: timestamp,
                             });
                         }
                         WriteOutcome::Failed(err) => break Err(err),
@@ -1085,10 +1115,10 @@ impl PeekClient {
                         }
                     }
                 }
-                ProcessResult::NoRowsMatched => {
-                    break Ok(OccOutcome::Committed {
+                ProcessResult::NoRowsMatched { observed_ts } => {
+                    break Ok(OccOutcome::NoRowsMatched {
                         response: build_no_rows_response(&kind),
-                        write_ts: None,
+                        observed_ts: Some(observed_ts),
                     });
                 }
                 ProcessResult::Error(e) => {
@@ -1314,8 +1344,13 @@ impl OccState {
 
 /// Result of processing a single subscribe message in the OCC loop.
 enum ProcessResult {
-    Continue { ready_to_write: bool },
-    NoRowsMatched,
+    Continue {
+        ready_to_write: bool,
+    },
+    /// The consolidated selection is empty as of `observed_ts`.
+    NoRowsMatched {
+        observed_ts: Timestamp,
+    },
     Error(AdapterError),
 }
 
@@ -1398,7 +1433,7 @@ fn process_message(
                     // `src/adapter/src/active_compute_sink.rs` for
                     // the emission order.
                     if ts > as_of && state.all_diffs.is_empty() {
-                        return ProcessResult::NoRowsMatched;
+                        return ProcessResult::NoRowsMatched { observed_ts: ts };
                     }
                 } else {
                     let Some(diff_datum) = datums.next() else {

test/cluster/mzcompose.py

@@ -4638,6 +4638,161 @@ def subscriber() -> None:
         ), "statement execution was ended twice; end-of-execution ownership handoff regressed"
 
 
+def workflow_test_occ_zero_row_write_linearization(c: Composition) -> None:
+    """A read-then-write that reports zero rows must not retire before the write
+    that emptied its selection is readable through the timestamp oracle.
+
+    The OCC path linearizes only its initial `as_of`, and its internal subscribe
+    follows Persist visibility, which runs ahead of the oracle: the group
+    committer appends before it applies the write timestamp. A DELETE or UPDATE
+    can therefore consolidate its selection to empty against state no
+    oracle-timestamped read can reach yet, report zero rows through
+    `NoRowsMatched`, and return. A strict-serializable read issued after that
+    response then still sees the row the response said was not there, and no
+    serial order explains that history.
+
+    The `group_commit_before_apply_write` failpoint holds the winning writer
+    inside that window, the same one a second `environmentd` process opens on
+    its own with no ordering against local Persist visibility.
+    """
+
+    # Every txns-shard write parks here while armed, including the keepalives
+    # that advance table uppers, so this has to be a bounded `sleep` and not a
+    # `pause`: a keepalive would take the `pause` first and the winning DELETE
+    # would never get to append. The window only has to outlast a peek and one
+    # subscribe dataflow installation.
+    failpoint = "group_commit_before_apply_write"
+    arm = f"SET failpoints = '{failpoint}=sleep(10000)'"
+    disarm = f"SET failpoints = '{failpoint}=off'"
+
+    def occ_writes() -> tuple[int, int]:
+        """Read-then-writes the OCC path sequenced, and how many of their write
+        attempts lost the race for their write timestamp."""
+        metric = "mz_occ_read_then_write_retry_count"
+        metrics = c.exec(
+            "materialized", "curl", "localhost:6878/metrics", capture=True
+        ).stdout
+        values = {
+            line.split()[0]: int(float(line.split()[1]))
+            for line in metrics.splitlines()
+            if line.startswith((f"{metric}_count ", f"{metric}_sum "))
+        }
+        return values[f"{metric}_count"], values[f"{metric}_sum"]
+
+    def count(cur: Cursor, key: int) -> int:
+        cur.execute(f"SELECT count(*) FROM t WHERE k = {key}".encode())
+        row = cur.fetchone()
+        assert row is not None
+        return int(row[0])
+
+    with c.override(
+        Materialized(
+            # Sampled once at startup, so this cannot be an `ALTER SYSTEM SET`.
+            additional_system_parameter_defaults={
+                "enable_adapter_frontend_occ_read_then_write": "true"
+            },
+        )
+    ):
+        c.up("materialized")
+        c.sql("CREATE TABLE t (k int, v int)")
+
+        # Ask the process rather than the catalog which path it takes: the UPDATE
+        # only reaches the histogram if the frontend sequenced it.
+        sequenced = occ_writes()[0]
+        c.sql("UPDATE t SET v = v + 1 WHERE k = 0")
+        assert occ_writes()[0] > sequenced, (
+            "the UPDATE did not go through the OCC path, so this would exercise the "
+            "coordinator's lock-based path instead"
+        )
+
+        # Connections are opened before the failpoint is armed: starting a session
+        # appends to `mz_sessions`, which parks like any other write.
+        with (
+            c.sql_cursor() as control,
+            c.sql_cursor() as probe,
+            c.sql_cursor() as winner_cur,
+            c.sql_cursor() as session,
+        ):
+            # A serializable read may pick a timestamp past the oracle's read
+            # timestamp, so it sees the winner's append while a strict-serializable
+            # read cannot.
+            probe.execute("SET transaction_isolation = 'serializable'")
+
+            # The UPDATE's subscribe can instead report progress at the table's
+            # pre-append upper, with the row still there, then submit a write,
+            # queue behind the parked committer, and conclude zero rows only once
+            # the oracle has caught up. That proves nothing either way, so such an
+            # attempt is retried; the write conflict count tells the two apart.
+            for attempt in range(1, 4):
+                key = attempt
+                c.sql(f"INSERT INTO t VALUES ({key}, 1)")
+                armed = Event()
+
+                def delete_winner(key: int = key) -> None:
+                    armed.wait()
+                    # Lands its append and advances t's upper, then the committer
+                    # parks before applying that timestamp to the oracle.
+                    winner_cur.execute(f"DELETE FROM t WHERE k = {key}".encode())
+
+                winner = PropagatingThread(target=delete_winner, name="winner")
+                winner.start()
+                control.execute(arm)
+                armed.set()
+
+                # The winner's append is visible in Persist from here on ...
+                deadline = time.time() + 120
+                while count(probe, key) > 0:
+                    assert (
+                        time.time() < deadline
+                    ), "the winning DELETE never became visible in Persist"
+                    time.sleep(0.1)
+                # ... and the oracle cannot serve reads at it yet, which is what
+                # puts us inside the window. This witness says nothing about the
+                # UPDATE, so it stays valid once the zero-row path waits.
+                before = count(session, key)
+
+                conflicts = occ_writes()[1]
+                started = time.time()
+                session.execute(f"UPDATE t SET v = v + 1 WHERE k = {key}".encode())
+                matched = session.rowcount
+                elapsed = time.time() - started
+                after = count(session, key)
+                conflicts = occ_writes()[1] - conflicts
+
+                control.execute(disarm)
+                # `off` does not interrupt a `sleep` under way, so this waits out
+                # the rest of the window.
+                winner.join(timeout=120)
+                assert not winner.is_alive(), "the winning DELETE never finished"
+
+                print(
+                    f"attempt {attempt}: UPDATE matched {matched} row(s) in "
+                    f"{elapsed:.1f}s with {conflicts} write conflict(s); "
+                    f"strict-serializable reads saw {before} row(s) before it and "
+                    f"{after} row(s) after"
+                )
+                assert matched == 0, (
+                    f"the UPDATE matched {matched} row(s) with the winner's delete "
+                    "already visible, so it never took the zero-row path"
+                )
+                if before == 0 or conflicts > 0:
+                    continue
+
+                assert after == 0, (
+                    "the UPDATE reported zero rows matched from state the oracle had "
+                    f"not applied yet, and a strict-serializable read after it saw "
+                    f"{after} row(s): the zero-row response was not linearized against "
+                    "the write that emptied the selection"
+                )
+                break
+            else:
+                raise AssertionError(
+                    "no attempt got the UPDATE to report zero rows from the newer "
+                    "state without first submitting a write, so the window was never "
+                    "observed"
+                )
+
+
 def workflow_test_refresh_mv_warmup(
     c: Composition, parser: WorkflowArgumentParser
 ) -> None: