adapter: let the oracle choose the read-then-write write timestamp

The OCC loop wrote at the frontier its subscribe reported, which conflates two
jobs. A frontier certifies what the loop has a complete view of. Choosing the
timestamp to write at is a separate decision, and the two coincide only because
the target table is usually what pins the frontier.

They come apart when another input does. A materialized view with a `REFRESH`
option settles until its next refresh, so a selection over one reports a
frontier hours or days ahead of the clock while the target table's upper is
still near it. Writing there ratchets the timeline's oracle into the future,
and the oracle is monotone and durable, so every later write and
strict-serializable read blocks until the clock catches up. Under
`serializable` the write is simply invisible.

So the oracle chooses and the frontier certifies. The target `T` is one step
above the oracle's write timestamp, the smallest value `commit_timestamped`
accepts, and a conflict hands back the next eligible one. The loop writes at
`T` once the frontier reaches `T`, with every diff from strictly below `T` as
the payload and everything at or after it held back for a later target.

Three boundaries carry the correctness. A progress message at `F` certifies
completeness below `F`, so readiness is `F >= T`. The payload is `t < T`
strictly, because a diff at `T` is concurrent with the write. And `T > as_of`,
because the snapshot arrives at `as_of` and has to be in the payload, which
the pre-read linearization guarantees by leaving the oracle at or above
`as_of`.

Readiness gives `T <= F`, not equality, so the payload can exclude diffs the
subscribe already delivered in `[T, F)`. For a selection that reads the target
table those say the table moved past `T`, the compare-and-append refuses, and
the loop retries at the timestamp the refusal names. Persist arbitrates, not
the frontier. This path produces that window itself: the committer appends at
the target and applies it to the oracle only afterwards, so while one write
sits in between, a second statement's target is a step behind the table's
upper. Retries converge because both refusals name a strictly higher timestamp
and the loop waits for the frontier to certify each one.

A zero-row answer no longer waits on a frontier either. Emptiness that already
holds at `as_of` is reported against `as_of`, which the pre-read linearization
covered, so the common `UPDATE ... WHERE <no match>` returns without a group
commit. Emptiness that appears later is reported against the target it was
concluded at rather than a frontier that can be days out.

Tests: the far-future refresh-MV write commits near the clock instead of being
refused, and a `serializable` session reads back its own such write, which
pins the anomaly this closes. Unit tests cover the fold's boundaries, where
the off-by-ones live.

Author: aljoscha

doc/developer/design/20260210_incremental_occ_read_then_write.md

@@ -49,9 +49,9 @@ a subscribe that continually tracks the current state of the data.
 - Removing the in-process locks immediately. During rollout, the old lock-based
   path and the new OCC path coexist behind a feature flag. The locks can be
   removed once the OCC path is fully rolled out.
-- Mixed read/write transactions. A write that reads persisted state commits at
-  the frontier it observed, which it cannot postpone until COMMIT, so it runs
-  only as a single statement. A write that reads nothing does compose with
+- Mixed read/write transactions. A write that reads persisted state commits at a
+  timestamp the oracle handed out while the statement ran, which it cannot
+  postpone until COMMIT, so it runs only as a single statement. A write that reads nothing does compose with
   transactions: its diffs are frontier-independent, so they are buffered as
   session write ops and land when the transaction commits. That covers, for
   example, `INSERT INTO t SELECT generate_series(1, 20000)`, whose values are
@@ -65,12 +65,13 @@ subscribe-based OCC loop:
 1. Open a subscribe on the read expression (the `selection` from the
    `ReadThenWrite` plan), starting at the timestamp determined by the oracle
 2. Accumulate diffs from the subscribe
-3. When the subscribe frontier advances to T (meaning we have a consistent
-   snapshot), attempt to write the accumulated diffs at timestamp T
-4. If the write succeeds, done
-5. If the write fails because another writer already committed at timestamp T,
-   the subscribe will deliver the new state; go back to step 3 with the updated
-   diffs
+3. Take the write timestamp T from the timeline's oracle, one step above its
+   write timestamp, which is the smallest value the group committer accepts
+4. Once the subscribe frontier has advanced to T, so the accumulated diffs below
+   T are complete, attempt to write those diffs at T
+5. If the write succeeds, done
+6. If the write fails because another writer already took T, adopt the timestamp
+   the committer reports as next eligible and go back to step 4 with it
 
 This approach is correct by construction: the subscribe always reflects the
 committed state of the data, and the timestamped write mechanism ensures that
@@ -144,11 +145,12 @@ Session Task                         Coordinator
   |                                      |
   |   +-- OCC Loop ------------------+   |
   |   | receive diffs from subscribe |   |
-  |   | on frontier advance:         |   |
-  |   |   consolidate diffs          |   |
+  |   | target T from the oracle     |   |
+  |   | once frontier >= T:          |   |
+  |   |   consolidate diffs below T  |   |
   |   |   AttemptTimestampedWrite -> |-->|-- group_commit()
   |   |   <-- Success/Failed --------|<--|
-  |   |   if Failed: continue loop   |   |
+  |   |   if Failed: T = next, loop  |   |
   |   |   if Success: break          |   |
   |   +------------------------------+   |
   |                                      |
@@ -230,16 +232,20 @@ selection.
 
 ### The timestamped write ensures atomicity
 
-The write is submitted at the timestamp corresponding to the subscribe's
-frontier. The group commit machinery checks that this timestamp hasn't been
+The write is submitted at a timestamp taken from the timeline's oracle, once the
+subscribe's frontier has reached it so that the accumulated diffs below it are
+complete. The group commit machinery checks that this timestamp hasn't been
 passed by the oracle:
 
 - If the timestamp is still valid: the write is committed at exactly that
   timestamp, and the oracle is advanced past it. Any concurrent OCC loops that
   were targeting the same timestamp will fail and retry.
 - If the timestamp has already passed (another write committed first): the
-  write also fails. The OCC loop continues, the subscribe delivers the updates
-  from the intervening write, and the loop retries at the new frontier.
+  write also fails, and the reply names the next eligible timestamp. The OCC
+  loop adopts that as its new target, waits for the subscribe's frontier to
+  reach it, which folds the intervening writes' updates into the payload, and
+  retries. The reported timestamp is always strictly above the rejected one, so
+  the retries make progress.
 
 This ensures that the write is always based on the state of the data at exactly
 the write timestamp. There is no window for lost updates: either the write
@@ -324,13 +330,19 @@ that the next reader does not take them for bugs.
   When inputs are caught up the lock path's window is milliseconds wide and also
   needs a materially conflicting write plus a reader inside it, which is
   presumably why it went unnoticed.
-- **A lagging dependency blocks rather than waits.** This is the price of the
-  strengthening above. A selection dependency that persistently lags by more
-  than about one `default_timestamp_interval` makes every attempt conflict,
-  because the observed frontier is bounded by the lagging input while the write
-  timestamp keeps advancing with the oracle. The statement then burns retries
-  until `statement_timeout` instead of committing, where the lock path's peek
-  simply waited for the input to catch up.
+- **A lagging dependency delays rather than being read stale.** This is the
+  price of the strengthening above. The write timestamp comes from the oracle,
+  and the loop waits for the subscribe's frontier to certify it before
+  submitting, so a lagging selection dependency delays the statement by its lag.
+  A dependency that catches up commits normally. One that persistently lags by
+  more than about one `default_timestamp_interval` never lets an attempt land:
+  every wait ends with the oracle already past the target, the committer refuses
+  it and names a newer one, and the next wait is again bounded by the lagging
+  input. Each round costs one of `max_occ_retries`, but the rounds are paced by
+  frontier advances rather than spinning, so what ends the statement is
+  `statement_timeout`, which it runs out while holding an OCC permit and its
+  subscribe. The lock path's peek simply waited for the input to catch up and
+  then committed.
 - **Statement lifecycle events.** The frontend path records an
   `optimization-finished` event for a DML, the coordinator path does not,
   because it hands the read-then-write's inner peek a trivial logging context
@@ -357,11 +369,19 @@ that the next reader does not take them for bugs.
   limit on the coordinator path and succeed on the frontend path. We keep the
   frontend's accounting: it matches what the write actually appends, one entry
   with a large diff.
-- **The write-timeline throttle.** A timestamped write does not go through the
-  throttle that a blind write's group commit applies, because its timestamp
-  comes from an observed subscribe frontier rather than from the clock. See the
-  doc comment on `GroupCommitter::commit_timestamped` for the full list of what
-  that path skips and why.
+- **The write-timeline throttle.** A blind write's group commit sleeps in the
+  committer until the wall clock catches up with the oracle's write timestamp,
+  keeping the timeline from running ahead of the clock. A timestamped write
+  cannot be throttled that way: its timestamp is fixed before it reaches the
+  committer, so sleeping would only delay a write that already has to land at
+  that timestamp. The committer instead refuses a target above
+  `write_ts_upper_bound(now)` outright. That refusal is unreachable in normal
+  operation, since the target is one step above the oracle's write timestamp and
+  the oracle clamps itself to the clock. It fires only for a write timeline that
+  has already run away from the clock, which is an environment-level invariant
+  violation rather than something a statement can provoke. See the doc comment on
+  `GroupCommitter::commit_timestamped` for the full list of what that path skips
+  and why.
 - **Zero-row `INSERT ... RETURNING`.** Both paths report `INSERT 0 0` with no
   result set when no rows match, because the coordinator decides the response
   kind from the evaluated RETURNING rows and there are none. Postgres returns an
@@ -398,10 +418,10 @@ throughput (left) and latency (right). Key observations:
   a subscribe sees only progress from another table's write. They do still
   contend, in three ways: the concurrency semaphore is process-global across
   tables and clusters, the conflict predicate is the global oracle plus the
-  shared txns-shard upper, so two writers that observed the same frontier refuse
-  each other, and each timestamped write is its own committer round rather than
-  merging into a shared group commit. Every write benchmark is single-table, so
-  the cross-table case is unmeasured.
+  shared txns-shard upper, so two writers that took the same target timestamp
+  refuse each other, and each timestamped write is its own committer round rather
+  than merging into a shared group commit. Every write benchmark is single-table,
+  so the cross-table case is unmeasured.
 
 The chart above is from the PoC, which benchmarked `UPDATE t SET x = x + 1` over
 a larger table (the regime where OCC wins). It does not capture the small-write

src/adapter/src/coord/appends.rs

@@ -445,11 +445,11 @@ impl GroupCommitter {
     ///
     /// What [`Self::commit`] does that this skips, and why that is safe:
     ///
-    /// * The wall-clock throttle. `target_timestamp` is the caller's to choose, so
-    ///   instead of sleeping until the clock catches up we refuse a target above
-    ///   [`write_ts_upper_bound`] outright. Sleeping is the wrong answer for a caller
-    ///   whose target can be hours out, and committing there would advance the oracle
-    ///   with it.
+    /// * The wall-clock throttle. `target_timestamp` is the caller's to choose, and a
+    ///   target above [`write_ts_upper_bound`] is refused rather than slept off.
+    ///   Committing there would advance the oracle with it, and a caller that took its
+    ///   target from the oracle cannot exceed the bound unless the timeline has already
+    ///   run away, which sleeping would not resolve.
     /// * A [`GroupCommitPermit`]. The caller bounds how many of these are in
     ///   flight, and that is the backpressure for this path.
     /// * Merging queued commits. There is nothing to merge into: these diffs

src/adapter/src/error.rs

@@ -143,10 +143,13 @@ pub enum AdapterError {
     /// A frontend read-then-write's write timestamp ran further ahead of the wall
     /// clock than the write timeline may be advanced.
     ///
-    /// Nothing was appended. Committing there would advance the timeline's oracle
-    /// to that timestamp, and the oracle is monotone and durable, so every later
-    /// write and strict-serializable read would block until the wall clock caught
-    /// up, restarts included.
+    /// Nothing was appended. The timestamp comes from the timeline's oracle, one
+    /// step above its write timestamp, so hitting the bound means the oracle has
+    /// run away from the wall clock rather than that the statement asked for
+    /// anything unusual. Committing there would advance the oracle further, and
+    /// the oracle is monotone and durable, so every later write and
+    /// strict-serializable read would block until the wall clock caught up,
+    /// restarts included.
     ReadThenWriteTimestampTooFarAhead {
         target_timestamp: mz_repr::Timestamp,
         limit: mz_repr::Timestamp,
@@ -848,10 +851,9 @@ impl AdapterError {
                  committing. Retry the statement, or lower the write concurrency.".into()
             ),
             AdapterError::ReadThenWriteTimestampTooFarAhead { .. } => Some(
-                "The selection reads a collection whose contents are already settled far \
-                 into the future, for example a materialized view with a REFRESH option. \
-                 Read it into a table first, or select from it at a time it is still \
-                 changing.".into()
+                "This environment's write timestamp is far ahead of the wall clock. Writes \
+                 and strict-serializable reads cannot proceed until the clock catches up \
+                 with it. The statement cannot be rewritten to avoid this.".into()
             ),
             AdapterError::CollectionUnreadable { .. } => Some(
                 "This could be because the collection has recently been dropped.".into()
@@ -918,7 +920,12 @@ impl AdapterError {
             }
             AdapterError::ReadThenWriteContention => SqlState::T_R_SERIALIZATION_FAILURE,
             AdapterError::ReadThenWriteTimestampTooFarAhead { .. } => {
-                SqlState::FEATURE_NOT_SUPPORTED
+                // An invariant violation in the environment rather than a property of
+                // the statement: the write timeline has run away from the wall clock.
+                // The same condition trips `check_runaway_write_ts`'s soft panic on the
+                // next write the timeline applies, so a test build fails rather than
+                // returning this at all.
+                SqlState::INTERNAL_ERROR
             }
             AdapterError::CollectionUnreadable { .. } => SqlState::NO_DATA_FOUND,
             AdapterError::NoClusterReplicasAvailable { .. } => SqlState::FEATURE_NOT_SUPPORTED,