adapter: sequence read-then-write from the session task with OCC

[aljoscha]

NOTE: The diff here is quite big: the largest part is actually tests, then there's the Coordinator side support machinery for frontend read-then-write, and then there's the actual new occ loop to review.

Motivation

Part 6 of 7, the feature itself, in a stack that moves DELETE, UPDATE and
INSERT ... SELECT off the coordinator onto the session task, using
optimistic concurrency control.

Design doc lands with this PR:
20260210_incremental_occ_read_then_write.md.

These statements run on the coordinator today, holding a write lock on the
target table across both the read and the write. Every such statement
serializes against every other one on that table, and the coordinator loop is
occupied for the duration.

Closes SQL-592

Description

The selection is read through an internal subscribe, which streams the
mutation's diffs directly rather than peeking every matched row and
recomputing them. The write is submitted at the timestamp those diffs were
observed at, and the group committer refuses it if another writer got there
first. A refusal is a retry with a fresh snapshot, not a lost update. The
retry budget is max_occ_retries.

Concurrency is bounded by a semaphore of max_concurrent_occ_writes permits,
acquired before the read holds, so that queued operations do not pin
compaction on their read dependencies while they wait. statement_timeout is
enforced in one place, the select! that owns the whole operation, because
every phase can block: permit acquisition, linearization against a
far-future as_of, and the retry loop itself.

Off by default, gated by enable_adapter_frontend_occ_read_then_write, read
once at startup and fixed for the life of the process. A mixed-mode window
would be unsound: the lock-based path excludes concurrent writers, the OCC
path detects them afterwards, and the two do not synchronize. The
coordinator's read-then-write path therefore rejects any statement that
reaches it while frontend sequencing is on, since that could only be a
routing bug.

Removes the dead_code allowances that parts 4 and 5 carried, in the same
commit that adds their callers.

Performance. Large mutations get faster because the subscribe streams
diffs. Small ones get slower because each installs a dataflow where the old
path used a fast-path peek. That trade is deliberate. Measured numbers,
including a cluster-memory step on ManySmallUpdates, are recorded in the
design doc.

The last commit closes a linearizability hole found in review. The path
linearizes its as_of before subscribing, but the subscribe then follows
Persist, which runs ahead of the oracle: group commit appends first and calls
oracle.apply_write after. A selection that consolidated to empty inside that
window reported zero rows from state no oracle-timestamped read could reach
yet, and a strict-serializable read issued after that response still saw the
rows. Only the zero-row exits are exposed, because a write is linearized for
free by group commit answering downstream of apply_write. OccOutcome now
separates "nothing to write" from "committed" and carries the timestamp
emptiness was concluded at, which the caller waits for before responding.

Verification

• Feature benchmark and parallel benchmark runs, with the accepted regressions
  and their causes written down in the design doc.
• Parallel workload with the oracle from part 1, which is what would catch a
  lost update.
• A statement-logging parity harness asserting the frontend and coordinator
  paths produce the same mz_statement_execution_history rows for the same
  statement.
• Tests for cancellation and timeout releasing the OCC permit, for a
  dependency dropped underneath a running mutation, for zero-row RETURNING,
  for max_result_size, and for mz_now() error parity with the coordinator.
• An adversarial pass over the OCC path, whose findings are pinned by tests in
  src/environmentd/tests/read_then_write.rs, covering contention, permit
  starvation, races with ALTER TABLE, and constraint enforcement.

The read-then-write tests live in a new read_then_write.rs rather than in
server.rs, and the DML statement-logging tests join the statement_logging.rs
that part 3 creates. Interleaving them into server.rs would have grown it to
9826 lines and made this diff churn a thousand lines of untouched tests.

[linear-code]

SQL-592

[def-]

3 issues from the QA LLM review, verified with 3 tests:

1. enable_adapter_frontend_occ_read_then_write is sampled once at process startup. During a 0dt upgrade another environmentd can come up with a different setting, which will be a problem. This seems like a potential blocker for 0dt-enabling this feature?

diff --git a/test/txn-wal-fencing/mzcompose.py b/test/txn-wal-fencing/mzcompose.py
index 3e2e254692..512301530c 100644
--- a/test/txn-wal-fencing/mzcompose.py
+++ b/test/txn-wal-fencing/mzcompose.py
@@ -14,6 +14,7 @@ purpose of exercising fencing.

 import argparse
 import random
+import threading
 import time
 from concurrent import futures
 from dataclasses import dataclass
@@ -96,6 +97,17 @@ SERVICES = [
     Materialized(name="mz_second"),
 ]

+# Selects how a process sequences DELETE/UPDATE/INSERT ... SELECT: `false` keeps
+# them on the Coordinator behind in-process write locks, `true` sequences them
+# from the session task under optimistic concurrency control. The value is
+# sampled once at process startup, so two processes in one environment can hold
+# different values.
+OCC_FLAG = "enable_adapter_frontend_occ_read_then_write"
+
+# Observed once per read-then-write that the OCC path sequenced, so the
+# histogram's sample count identifies which path a process took.
+OCC_METRIC = "mz_occ_read_then_write_retry_count_count"
+

 def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
     parser.add_argument(
@@ -281,3 +293,264 @@ def run_workload(c: Composition, workload: Workload, args: argparse.Namespace) -
                 ), f"Unexpected result {result}; commit: {commit}; target {target}"

         print("Verification complete.")
+
+
+# Connections driving increments.
+MIXED_MODE_CONCURRENCY = 16
+
+# The statement never reached a server, or a server refused it, so it wrote
+# nothing. Keeping these apart from the indeterminate ones below matters: a
+# fenced instance produces plenty, and each one widens the band the counter is
+# checked against.
+NOT_APPLIED = [
+    # Targeted a container that is not up, or one that died before the
+    # connection was established, or one that stopped responding mid-connect.
+    "running docker compose failed",
+    "Connection refused",
+    "Connection timed out",
+    "canceling statement due to statement timeout",
+    # How the OCC path reports sustained contention.
+    "read-then-write exceeded maximum retry attempts under contention",
+]
+
+# The connection went away with the statement in flight, so the increment may or
+# may not be durable. A fenced container still publishes its port, so the
+# connection is accepted and reset.
+INDETERMINATE = [
+    "server closed the connection unexpectedly",
+    "Connection reset by peer",
+]
+
+
+class Increment(Enum):
+    ACKED = 0
+    REJECTED = 1
+    UNKNOWN = 2
+
+
+def occ_sequenced_writes(c: Composition, service: str) -> int:
+    """How many read-then-writes `service` has sequenced through the OCC path."""
+    metrics = c.exec(
+        service, "curl", "--silent", "localhost:6878/metrics", capture=True
+    ).stdout
+    for line in metrics.splitlines():
+        if line.startswith(f"{OCC_METRIC} "):
+            return int(float(line.split()[1]))
+    # The histogram is registered unconditionally, so a missing line means the
+    # scrape itself did not land.
+    raise RuntimeError(f"{OCC_METRIC} not found in {service} metrics")
+
+
+def increment_counter(args: tuple[Composition, str, bool]) -> Increment:
+    """Increments the shared counter by one through a read-then-write.
+
+    `slow` stretches the read phase, so that the operation can straddle the
+    moment the second instance comes up. The sleep takes its duration from the
+    `delay` column rather than from a literal, so that it is not folded away at
+    plan time and instead runs while the selection is read.
+    """
+    c, mz_service, slow = args
+    sleep = " AND mz_unsafe.mz_sleep(delay) IS NULL" if slow else ""
+    try:
+        c.sql_cursor(service=mz_service).execute(
+            f"UPDATE counter SET v = v + 1 WHERE k = 1{sleep}".encode()
+        )
+    except Exception as e:
+        if any(msg in str(e) for msg in NOT_APPLIED):
+            return Increment.REJECTED
+        if any(msg in str(e) for msg in INDETERMINATE):
+            return Increment.UNKNOWN
+        raise RuntimeError(f"unexpected exception: {e}")
+    return Increment.ACKED
+
+
+def workflow_mixed_mode_read_then_write(
+    c: Composition, parser: WorkflowArgumentParser
+) -> None:
+    """Two instances in one environment sequencing read-then-write differently.
+
+    `OCC_FLAG` is sampled once per process, so a rolling restart or a newly added
+    serving process can leave one instance on the Coordinator's write-lock path
+    and another on the OCC path. The two do not synchronize: the lock path
+    excludes concurrent writers, the OCC path detects them afterwards from the
+    write timestamp. A blind write from the lock path landing on top of an OCC
+    write leaves the row with a negative copy of the stale value and two copies
+    of the new one.
+
+    Both orderings run, because they put the lock path on opposite sides of the
+    handover. Each one goes red on a lost update or a broken multiplicity, and
+    also on the precondition for either: the two instances committing at the same
+    time.
+    """
+    parser.add_argument(
+        "--azurite", action="store_true", help="Use Azurite as blob store instead of S3"
+    )
+    args = parser.parse_args()
+
+    # Every increment opens its own connection, so leave out the per-invocation
+    # Docker Compose echo.
+    c.silent = True
+
+    for first_occ, second_occ in [("false", "true"), ("true", "false")]:
+        print(
+            f"+++ Running with {OCC_FLAG} {first_occ} on 'mz_first', {second_occ} on 'mz_second' ..."
+        )
+        run_mixed_mode(c, args.azurite, first_occ, second_occ)
+
+
+def run_mixed_mode(
+    c: Composition, azurite: bool, first_occ: str, second_occ: str
+) -> None:
+    """Runs one ordering: 'mz_first' comes up first, then 'mz_second' displaces it."""
+    c.down(destroy_volumes=True)
+    c.up(c.metadata_store())
+
+    with c.override(
+        *[
+            Materialized(
+                name=mz_name,
+                external_metadata_store=True,
+                external_blob_store=True,
+                blob_store_is_azure=azurite,
+                sanity_restart=False,
+                support_external_clusterd=True,
+                additional_system_parameter_defaults={OCC_FLAG: occ},
+            )
+            for mz_name, occ in [("mz_first", first_occ), ("mz_second", second_occ)]
+        ]
+    ):
+        c.up("mz_first")
+
+        # Idempotent, because a retried connection re-runs the whole batch after
+        # the statements it already applied. `mz_sleep` blocks the timely worker
+        # it runs on, so `delay` stays short enough that the slow increments do
+        # not starve the replica.
+        c.sql(
+            """
+            CREATE TABLE IF NOT EXISTS counter (k int, v bigint, delay double precision);
+            DELETE FROM counter;
+            INSERT INTO counter VALUES (1, 0, 0.5);
+            """,
+            service="mz_first",
+        )
+
+        print("--- Confirming the instances disagree on how to sequence")
+        assert (
+            increment_counter((c, "mz_first", False)) == Increment.ACKED
+        ), "baseline increment on 'mz_first' did not commit"
+        occ_writes = occ_sequenced_writes(c, "mz_first")
+        assert (occ_writes > 0) == (first_occ == "true"), (
+            f"'mz_first' sequenced {occ_writes} read-then-writes through OCC, "
+            f"which does not match {OCC_FLAG}={first_occ}"
+        )
+
+        print("--- Driving increments across both instances")
+        stop = threading.Event()
+
+        def drive(worker: int) -> list[tuple[str, Increment, float]]:
+            # Worker 0 keeps a slow read-then-write in flight on 'mz_first' for
+            # the whole run, so that 'mz_second' comes up while an operation
+            # there sits between its read and its write. The workers aimed at
+            # 'mz_second' start before it can serve and take their rejections
+            # until it can.
+            mz_service = "mz_first" if worker % 2 == 0 else "mz_second"
+            op = (c, mz_service, worker == 0)
+            outcomes = []
+            while not stop.is_set():
+                issued = time.time()
+                outcomes.append(
+                    (mz_service, increment_counter(op), issued, time.time())
+                )
+            return outcomes
+
+        with futures.ThreadPoolExecutor(MIXED_MODE_CONCURRENCY) as executor:
+            drivers = [
+                executor.submit(drive, worker)
+                for worker in range(MIXED_MODE_CONCURRENCY)
+            ]
+            try:
+                time.sleep(2)
+                c.up("mz_second")
+                # The fence lands while 'mz_second' opens the catalog, well
+                # before it reports healthy, so keep driving past that point to
+                # cover the handover from both sides.
+                time.sleep(15)
+            finally:
+                stop.set()
+            outcomes = [outcome for driver in drivers for outcome in driver.result()]
+
+        # The baseline increment on 'mz_first' counts too.
+        acked = 1 + sum(
+            1 for _, outcome, _, _ in outcomes if outcome == Increment.ACKED
+        )
+        unknown = sum(
+            1 for _, outcome, _, _ in outcomes if outcome == Increment.UNKNOWN
+        )
+        print(f"acked: {acked}, unknown: {unknown}")
+
+        occ_writes = occ_sequenced_writes(c, "mz_second")
+        assert (occ_writes > 0) == (second_occ == "true"), (
+            f"'mz_second' sequenced {occ_writes} read-then-writes through OCC, "
+            f"which does not match {OCC_FLAG}={second_occ}"
+        )
+
+        # Without a commit from each side the run exercised one instance only.
+        acks = {
+            mz_service: [
+                (issued, at)
+                for service, outcome, issued, at in outcomes
+                if service == mz_service and outcome == Increment.ACKED
+            ]
+            for mz_service in ["mz_first", "mz_second"]
+        }
+        for mz_service, times in acks.items():
+            assert times, f"'{mz_service}' committed no increment"
+
+        # What keeps the two modes from corrupting the row today is that they
+        # never commit at the same time: both paths advance the catalog upper
+        # before every write, so an instance stops committing once the other has
+        # fenced it, and the fence lands before the other instance reads anything.
+        # Overlapping commit windows are the bug's precondition, because a
+        # lock-path write can then land on top of an OCC write and leave the row
+        # with a negative copy of the stale value and two copies of the new one.
+        # Sampling the flag once per process does not prevent that, so a red
+        # assertion here means the modes have to be fenced against each other
+        # rather than merely fixed per process.
+        # Comparing when 'mz_first' last *issued* a statement that went on to
+        # commit against when 'mz_second' first committed keeps a slow response
+        # from reading as an overlap: a statement issued after the other instance
+        # had already committed cannot have committed before it.
+        gap = min(at for _, at in acks["mz_second"]) - max(
+            issued for issued, _ in acks["mz_first"]
+        )
+        print(f"'mz_first' stopped committing {gap:.1f}s before 'mz_second' started")
+        assert gap > 0, (
+            f"'mz_first' committed a statement it issued {-gap:.1f}s after "
+            f"'mz_second' had committed, so both sequencing modes were live at once"
+        )
+
+        print("--- Verifying the counter")
+        cursor = c.sql_cursor(service="mz_second")
+
+        # A negative copy of the stale value has no rendering, so a broken
+        # multiplicity surfaces here as a missing row, an extra row, or a
+        # retraction error.
+        cursor.execute("SELECT v FROM counter WHERE k = 1")
+        rows = cursor.fetchall()
+        assert len(rows) == 1, f"counter holds {rows}, expected exactly one row"
+
+        # An acked increment is durable, one whose connection died may or may not
+        # be. Anything below `acked` is a lost update.
+        v = rows[0][0]
+        assert (
+            acked <= v <= acked + unknown
+        ), f"counter is {v}, expected between {acked} and {acked + unknown}"
+
+        # Checked last so that a lost update is reported as such rather than as a
+        # missing fence. The fence bounds the window in which the two modes
+        # overlap: both paths advance the catalog upper before every write, so an
+        # instance stops writing once the other one has fenced it.
+        log = c.invoke("logs", "mz_first", capture=True).stdout
+        assert (
+            "unable to advance catalog upper" in log or "fenced by envd" in log
+        ), "'mz_first' was never fenced, so the two instances never overlapped"

2. The OCC path linearizes only its initial as_of, before subscribing. Its internal subscribe then follows Persist visibility, which runs ahead of the oracle: the group committer appends to the txns shard, making the write readable and advancing the table's upper, and only then calls oracle.apply_write. A concurrent writer's delete landing in that gap makes the loser's subscribe consolidate its selection to empty at that newer timestamp, so process_message still sees the row. Real time orders the read last, so no serial order explains the history.

diff --git a/test/cluster/mzcompose.py b/test/cluster/mzcompose.py
index 68bd562e5a..c3fca643e9 100644
--- a/test/cluster/mzcompose.py
+++ b/test/cluster/mzcompose.py
@@ -4638,6 +4638,161 @@ def workflow_test_drop_index_during_subscribe_sequencing(c: Composition) -> 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:

3. Read-free bulk INSERTs in transactions are rejected:

  BEGIN;
  INSERT INTO t SELECT generate_series(1, 20000);   -- ERROR: cannot be run inside a transaction block
  COMMIT;

Test:

diff --git a/test/testdrive/insert-select.td b/test/testdrive/insert-select.td
index 11c5d327e1..553eec38f5 100644
--- a/test/testdrive/insert-select.td
+++ b/test/testdrive/insert-select.td
@@ -107,6 +107,25 @@ contains:cannot be run inside a transaction block

 > COMMIT

+# A source above the optimizer's constant-folding limit does not fold to a
+# literal like the small VALUES list above. It still reads no persisted
+# collection, so its diffs belong in the transaction's write ops.
+
+> CREATE TABLE big (i INT)
+
+> BEGIN
+> INSERT INTO big SELECT g FROM generate_series(1, 20000) AS g
+> ROLLBACK
+> SELECT count(*) FROM big
+0
+
+> BEGIN
+> INSERT INTO big VALUES (0)
+> INSERT INTO big SELECT g FROM generate_series(1, 20000) AS g
+> COMMIT
+> SELECT count(*) FROM big
+20001
+
 > CREATE MATERIALIZED VIEW v (a, b, c) AS SELECT 11, 12::real, 'f';

 > INSERT INTO t (i, f, t) SELECT a, b, c FROM v;

Source: https://github.com/MaterializeInc/qa-llm-review/blob/master/commit-bugs/done/analysis-pr-37923.md

[aljoscha]

Thanks, this was worth the trouble. Taking all three, with one correction.

2 is real and is now fixed in this PR, as its own commit so the diff is
reviewable here: 0a4cb9c. Your description of the mechanism is exactly right.
The one thing I'd add is why only the zero-row exits are exposed: a write is
linearized for free, because group commit answers the writer downstream of
oracle.apply_write, and the coordinator's path takes its read timestamp from
the oracle and so cannot observe state the oracle has not reached. Reading
ahead of the oracle is specific to this path's subscribe, and the zero-row exit
is the only place we return a result derived from that read without going
through a write.

The fix reuses ensure_read_linearized, which already blocks until the oracle
reaches a timestamp. OccOutcome now separates "nothing to write" from
"committed", rather than encoding the former as a Committed with no
timestamp, and the NoRowsMatched variant carries the timestamp emptiness was
concluded at. I chose waiting over calling apply_write ourselves: forcing the
global oracle forward off another writer's timestamp would stall subsequent
reads of unrelated, slower collections until they caught up.

I added the group_commit_before_apply_write failpoint your test needs and took
workflow_test_occ_zero_row_write_linearization as written. It runs
automatically, since test/cluster's default workflow enumerates all
workflows.

1 we're accepting, and I've taken your test as a guard. Two notes on the
reasoning. The write locks are per-process, so they never excluded a second
environmentd: two lock-path processes writing concurrently produce the same
lost update and negative multiplicity as the mixed case, so mixed mode is no
worse than what ships today. And OCC's conflict detection is shared state
(peek_write_ts and the txns-shard upper), not in-process, so two OCC
processes detect each other and retry. The only bad combination is the one
containing the old path.

I also checked the intra-process version, which would have been worse: with the
flag on, every DELETE/UPDATE/INSERT routes to OCC with no fallback into the
lock path, so the two never race inside one process.

Your test needed wiring up or it would never have run, since
txn-wal-fencing's default workflow only iterates WORKLOADS. It's now a
nightly step. Its value is that it pins the fence assumption, which is what the
safety argument actually rests on and which nothing stated before.

3 is fixed by the next PR in the stack, #37924, which buffers a read-free
read-then-write as a session write op instead of refusing it. It has
test_nonconstant_insert_in_transaction covering that exact statement,
deliberately above FOLD_CONSTANTS_LIMIT so it doesn't fold to a literal, with
both the rollback and the commit case. Are you OK with it landing there rather
than here? Splitting it back would mean moving the transaction gate across two
PRs.

One more correction while I'm here, unrelated to your findings but found
chasing them: this PR's description claimed statement logging sent end events
with try_send and dropped them on a full channel. That was backwards. Main
uses an unbounded channel and panics. This stack is what introduces the
try_send, deliberately. Fixed in #37920's description.