adapter: refuse a write timestamp past the write timeline's bound

`GroupCommitter::commit_timestamped` documented an obligation it never
discharged. Its doc lists the wall-clock throttle among the things it skips,
"`target_timestamp` is the caller's to choose, and it must not run the write
timeline ahead of the clock", and then it committed whatever it was handed.

That is load-bearing, because the oracle is monotone and durable. A write
that lands far ahead of the clock is applied to the oracle, and from then on
every write and strict-serializable read on the timeline blocks until the
clock catches up. Restarting does not help. The oracle row is durable, and
the same timestamp also reaches the catalog shard's upper, which boot
re-applies to the oracle. Under `serializable` it is worse than a block:
those reads pick a timestamp near the clock and never consult the oracle, so
an acknowledged write stays invisible until the clock arrives.

The frontend read-then-write path can produce such a target today. It writes
at the frontier its subscribe observed, and a selection over a materialized
view with a `REFRESH` option settles until the next refresh, so that frontier
is legitimately hours or days out.

So enforce the bound where it is stated. A target above
`write_ts_upper_bound(now)`, the same ceiling `check_runaway_write_ts`
already measured against, is refused before the append, and the session
turns that into a statement error. This is a backstop rather than a fix: the
right answer is for the caller to pick a timestamp near the clock and use the
frontier only to certify what it read, which is a larger change to the OCC
loop. What this guarantees is that the worst outcome is one failed statement
instead of a stalled timeline.

Two smaller things in the same area. `check_runaway_write_ts` soft panics
instead of logging, so a runaway fails a test rather than leaving a line in a
log, and it degrades to that log line in production. Boot reports a catalog
upper that is already past the bound, which is the one channel that survives
a restart and was silent. Boot does not refuse to start, because the timeline
is stalled either way and a process that will not boot turns that into an
outage plus a crash loop.

The wall-clock bound on `write_ts` also holds in every oracle implementation
and is pinned by the shared conformance test, but was stated in none of them,
so the trait now carries it. Along with it: `peek_write_ts` does not advance
the timestamp, `apply_write` raises both timestamps rather than only bounding
future reads, and it is the door through which a timestamp the oracle did not
allocate enters, which is why the bound is a property of the
`EpochMilliseconds` timeline rather than of the oracle.

Tests: an integration test that drives the far-future target from a
materialized view with two refreshes, one a few seconds out so the view is
readable at all and one far away so the write target is deterministic, and
asserts the statement is refused, the oracle did not move, and the timeline
still takes writes and reads afterwards.

Author: aljoscha

src/adapter/src/coord.rs

@@ -4828,6 +4828,20 @@ pub fn serve(
             .expect("inserted above")
             .oracle;
 
+        // The catalog shard's upper is durable, so a write that once landed far ahead of the
+        // clock is re-applied to the oracle here on every boot and cannot be waited out. We
+        // report it rather than refusing to start: the timeline is stalled either way, and a
+        // process that will not boot turns that into a total outage plus a crash loop.
+        let boot_now: mz_repr::Timestamp = (now)().into();
+        if catalog_upper > timeline::write_ts_upper_bound(&boot_now) {
+            tracing::error!(
+                %catalog_upper, %boot_now,
+                "catalog upper is far ahead of the wall clock, so writes and \
+                strict-serializable reads on the EpochMilliseconds timeline will block \
+                until the clock catches up",
+            );
+        }
+
         let mut boot_ts = if read_only_controllers {
             let read_ts = epoch_millis_oracle.read_ts().await;
             std::cmp::max(read_ts, catalog_upper)

src/adapter/src/coord/appends.rs

@@ -64,6 +64,7 @@ use tokio::sync::{Notify, OwnedMutexGuard, OwnedSemaphorePermit, Semaphore, mpsc
 use tracing::{Instrument, Span, info, warn};
 
 use crate::catalog::{BuiltinTableUpdate, Catalog, CatalogUpperHandle};
+use crate::coord::timeline::write_ts_upper_bound;
 use crate::coord::{Coordinator, Message, PendingTxn, PlanValidity};
 use crate::metrics::Metrics;
 use crate::session::{EndTransactionAction, GroupCommitWriteLocks, Session, WriteLocks};
@@ -174,6 +175,12 @@ pub enum WriteResult {
         target_timestamp: Timestamp,
         next_eligible_timestamp: Timestamp,
     },
+    /// The requested timestamp ran further ahead of the wall clock than the write
+    /// timeline may be advanced, so the write was refused before it was attempted.
+    TimestampTooFarAhead {
+        target_timestamp: Timestamp,
+        limit: Timestamp,
+    },
     /// The write was canceled before it entered the committer.
     Canceled,
     /// The coordinator cannot accept writes.
@@ -438,8 +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,
-    ///   and it must not run the write timeline ahead of the clock.
+    /// * 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.
     /// * 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
@@ -466,6 +476,19 @@ impl GroupCommitter {
             return ControlFlow::Continue(());
         }
 
+        // The oracle is monotone and durable, so a write above the bound is not a delay we
+        // can wait out. It would strand the timeline past every restart until the wall
+        // clock caught up, and the write is applied to the oracle below.
+        let now: Timestamp = (self.now)().into();
+        let limit = write_ts_upper_bound(&now);
+        if target_timestamp > limit {
+            result.send(WriteResult::TimestampTooFarAhead {
+                target_timestamp,
+                limit,
+            });
+            return ControlFlow::Continue(());
+        }
+
         let write_ts = WriteTimestamp {
             timestamp: target_timestamp,
             advance_to: target_timestamp.step_forward(),

src/adapter/src/coord/timeline.rs

@@ -25,7 +25,7 @@ use mz_storage_types::sources::Timeline;
 use mz_timestamp_oracle::batching_oracle::BatchingTimestampOracle;
 use mz_timestamp_oracle::{self, TimestampOracle, TimestampOracleConfig, WriteTimestamp};
 use timely::progress::Timestamp as _;
-use tracing::{Instrument, debug, error, info};
+use tracing::{Instrument, debug, info};
 
 use crate::AdapterError;
 use crate::catalog::Catalog;
@@ -347,25 +347,35 @@ impl Coordinator {
     }
 }
 
-/// Convenience function for calculating the current upper bound that we want to
-/// prevent the global timestamp from exceeding.
-fn upper_bound(now: &mz_repr::Timestamp) -> mz_repr::Timestamp {
+/// The highest timestamp the `EpochMilliseconds` write timeline may be advanced to
+/// while the wall clock reads `now`.
+///
+/// A write above this is a runaway: the oracle is monotone and durable, so every later
+/// write and strict-serializable read on the timeline blocks until the wall clock catches
+/// up, across restarts. Group commit stays under it by allocating from the oracle, which
+/// clamps to the clock. A caller that chooses its own write timestamp has to be checked
+/// against it, see `GroupCommitter::commit_timestamped`.
+pub(crate) fn write_ts_upper_bound(now: &mz_repr::Timestamp) -> mz_repr::Timestamp {
     const TIMESTAMP_INTERVAL_MS: u64 = 5000;
     const TIMESTAMP_INTERVAL_UPPER_BOUND: u64 = 2;
 
     now.saturating_add(TIMESTAMP_INTERVAL_MS * TIMESTAMP_INTERVAL_UPPER_BOUND)
 }
 
-/// Logs an error when `timestamp` is further ahead of `now` than a local write timestamp
-/// should ever be, the signal that the `EpochMilliseconds` timeline has run away (e.g. after a
-/// wall-clock regression).
+/// Reports a write timestamp that is further ahead of `now` than
+/// [`write_ts_upper_bound`] allows, the signal that the `EpochMilliseconds` timeline has
+/// run away (e.g. after a wall-clock regression, or from a durably poisoned oracle).
+///
+/// This is a detector, not a guard: the timestamp has already been chosen, and every
+/// caller that can still refuse one checks the bound itself. It soft panics so that a
+/// runaway fails a test rather than only leaving a line in a log, and in production it
+/// degrades to that log line.
 pub(crate) fn check_runaway_write_ts(now: &mz_repr::Timestamp, timestamp: mz_repr::Timestamp) {
-    let upper_bound = upper_bound(now);
+    let upper_bound = write_ts_upper_bound(now);
     if timestamp > upper_bound {
-        error!(
-            %now,
-            "Setting local write timestamp to {timestamp}, which is more than \
-            the desired upper bound {upper_bound}."
+        mz_ore::soft_panic_or_log!(
+            "setting local write timestamp to {timestamp}, which is more than \
+            the desired upper bound {upper_bound} (now={now})"
         );
     }
 }

src/adapter/src/error.rs

@@ -140,6 +140,17 @@ pub enum AdapterError {
     /// The statement is retryable: every attempt was refused before anything
     /// was appended, so nothing it intended has been committed.
     ReadThenWriteContention,
+    /// 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.
+    ReadThenWriteTimestampTooFarAhead {
+        target_timestamp: mz_repr::Timestamp,
+        limit: mz_repr::Timestamp,
+    },
     CollectionUnreadable {
         id: String,
     },
@@ -836,6 +847,12 @@ impl AdapterError {
                 "Concurrent writes to the target table kept this statement from \
                  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()
+            ),
             AdapterError::CollectionUnreadable { .. } => Some(
                 "This could be because the collection has recently been dropped.".into()
             ),
@@ -900,6 +917,9 @@ impl AdapterError {
                 SqlState::T_R_SERIALIZATION_FAILURE
             }
             AdapterError::ReadThenWriteContention => SqlState::T_R_SERIALIZATION_FAILURE,
+            AdapterError::ReadThenWriteTimestampTooFarAhead { .. } => {
+                SqlState::FEATURE_NOT_SUPPORTED
+            }
             AdapterError::CollectionUnreadable { .. } => SqlState::NO_DATA_FOUND,
             AdapterError::NoClusterReplicasAvailable { .. } => SqlState::FEATURE_NOT_SUPPORTED,
             AdapterError::OperationProhibitsTransaction(_) => SqlState::ACTIVE_SQL_TRANSACTION,
@@ -1281,6 +1301,16 @@ impl fmt::Display for AdapterError {
                     "read-then-write exceeded maximum retry attempts under contention"
                 )
             }
+            AdapterError::ReadThenWriteTimestampTooFarAhead {
+                target_timestamp,
+                limit,
+            } => {
+                write!(
+                    f,
+                    "read-then-write would have to commit at {target_timestamp}, past the \
+                     highest timestamp the write timeline may be advanced to ({limit})"
+                )
+            }
             AdapterError::CollectionUnreadable { id } => {
                 write!(f, "collection '{id}' is not readable at any timestamp")
             }

src/adapter/src/frontend_read_then_write.rs

@@ -248,6 +248,13 @@ fn classify_write_result(
                 .requested_error()
                 .unwrap_or(AdapterError::Canceled),
         ),
+        WriteResult::TimestampTooFarAhead {
+            target_timestamp,
+            limit,
+        } => WriteOutcome::Failed(AdapterError::ReadThenWriteTimestampTooFarAhead {
+            target_timestamp,
+            limit,
+        }),
         WriteResult::ReadOnly => WriteOutcome::Failed(AdapterError::ReadOnly),
         WriteResult::TargetChanged => {
             // A concurrent DDL gave the table a new generation after we

src/environmentd/tests/read_then_write.rs

@@ -1721,3 +1721,79 @@ fn test_zero_row_write_does_not_wait_for_keepalive() {
          interval rather than a nudged group commit (total {elapsed:?})"
     );
 }
+
+/// A read-then-write whose write timestamp would run past what the write timeline
+/// may be advanced to has to be refused, not committed.
+///
+/// The OCC path derives its write timestamp from the frontier its subscribe observed,
+/// and a selection over a materialized view with a `REFRESH` option settles until the
+/// next refresh, so that frontier is legitimately hours or days ahead of the clock.
+/// Committing there advances the timeline's oracle with it, and the oracle is monotone
+/// and durable, so every later write and strict-serializable read on the timeline blocks
+/// until the clock catches up, restarts included. Under `serializable` it is worse than a
+/// block: reads pick a timestamp near the clock, so the acknowledged write stays
+/// invisible.
+///
+/// The MV here refreshes once a few seconds out and once far away. Past the near
+/// refresh it is readable, and its upper is then the far one, which is what makes the
+/// write target far future deterministically rather than by racing a refresh interval.
+#[mz_ore::test]
+#[allow(clippy::disallowed_methods)]
+fn test_far_future_write_timestamp_is_refused() {
+    let server = frontend_occ_harness()
+        .unsafe_mode()
+        .with_system_parameter_default("enable_refresh_every_mvs".to_string(), "true".to_string())
+        .start_blocking();
+    let mut client = server.connect(postgres::NoTls).unwrap();
+
+    client.batch_execute("CREATE TABLE src (a INT)").unwrap();
+    client
+        .batch_execute("INSERT INTO src VALUES (1), (2), (3)")
+        .unwrap();
+    client.batch_execute("CREATE TABLE dst (a INT)").unwrap();
+    client
+        .batch_execute(
+            "CREATE MATERIALIZED VIEW mv \
+             WITH (REFRESH AT mz_now()::text::int8 + 3000, REFRESH AT '3000-01-01') \
+             AS SELECT a FROM src",
+        )
+        .unwrap();
+
+    // Reaching the write at all means waiting for the near refresh: until then the MV
+    // holds no readable content and the read parks instead.
+    client
+        .batch_execute("SET statement_timeout = '60s'")
+        .unwrap();
+
+    let err = client
+        .execute("INSERT INTO dst SELECT a FROM mv", &[])
+        .expect_err("a write at a far-future timestamp must be refused");
+    let message = server_error_message(&err);
+    assert!(
+        message.contains("past the highest timestamp the write timeline may be advanced to"),
+        "unexpected error for a far-future write: {message}"
+    );
+
+    // The refusal has to happen before the append, so the oracle never learns the
+    // far-future timestamp. Checked before anything reads `dst`, because a read cannot be
+    // served once the oracle is out there.
+    let skew: i64 = client
+        .query_one(
+            "SELECT mz_now()::text::bigint - (extract(epoch FROM now()) * 1000)::bigint",
+            &[],
+        )
+        .unwrap()
+        .get(0);
+    assert!(
+        skew.abs() < 60_000,
+        "the oracle is {skew}ms from wall clock, so the refused write reached it anyway"
+    );
+
+    // The timeline is still usable, both for writes and for reads of the target.
+    client.batch_execute("INSERT INTO dst VALUES (4)").unwrap();
+    let rows = client
+        .query_one("SELECT count(*) FROM dst", &[])
+        .unwrap()
+        .get::<_, i64>(0);
+    assert_eq!(rows, 1, "the refused write must not have landed");
+}

src/timestamp-oracle/src/lib.rs

@@ -41,16 +41,29 @@ pub struct WriteTimestamp<T = mz_repr::Timestamp> {
 ///
 /// Specifically, all read timestamps will be greater or equal to all previously
 /// reported completed write timestamps, and strictly less than all subsequently
-/// emitted write timestamps.
+/// emitted write timestamps. The read timestamp never exceeds the write
+/// timestamp.
 #[async_trait]
 pub trait TimestampOracle<T>: std::fmt::Debug {
     /// Acquire a new timestamp for writing.
     ///
     /// This timestamp will be strictly greater than all prior values of
     /// `self.read_ts()` and `self.write_ts()`.
+    ///
+    /// An implementation built with a wall-clock [`NowFn`] must return
+    /// `max(previous_write_ts + 1, now())`, so it runs ahead of that clock only by the
+    /// number of allocations that landed within one of its ticks. An in-memory oracle
+    /// with no clock is exempt, it only has to keep the monotonicity above. Callers
+    /// depend on the bound to keep a write timeline from
+    /// running away: the oracle is monotone and durable, so a timestamp handed out far
+    /// ahead of the clock stalls every later write and linearized read on the timeline
+    /// until the clock catches up, restarts included. `timestamp_oracle_impl_test` pins
+    /// this, by asserting the returned value exactly rather than as a range.
     async fn write_ts(&self) -> WriteTimestamp<T>;
 
-    /// Peek the current write timestamp.
+    /// Peek the current write timestamp without advancing it.
+    ///
+    /// The value never decreases from one call to the next.
     async fn peek_write_ts(&self) -> T;
 
     /// Acquire a new timestamp for reading.
@@ -62,9 +75,18 @@ pub trait TimestampOracle<T>: std::fmt::Debug {
 
     /// Mark a write at `write_ts` completed.
     ///
-    /// All subsequent values of `self.read_ts()` will be greater or equal to
-    /// `write_ts`.
-    async fn apply_write(&self, lower_bound: T);
+    /// Neither timestamp is lowered and both end at or above `write_ts`, so all
+    /// subsequent values of `self.read_ts()` are greater or equal to `write_ts`. A
+    /// timestamp already above `write_ts` stays where it is, so this is not a way to
+    /// learn what the oracle will hand out next.
+    ///
+    /// NOTE: This is the door through which a timestamp that `write_ts` did not allocate
+    /// enters the oracle, so the wall-clock bound above holds only for a timeline whose
+    /// `apply_write` arguments are themselves clock- or oracle-derived. That is the case
+    /// for `EpochMilliseconds`. A source-driven timeline is deliberately ratcheted to its
+    /// inputs' frontiers instead, and is built with a clock pinned to the minimum
+    /// timestamp.
+    async fn apply_write(&self, write_ts: T);
 }
 
 /// A [`NowFn`] that is generic over the timestamp.