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.

The oracle is monotone and durable, so that matters. A write far ahead of the
clock is applied to the oracle, and every later write and strict-serializable
read on the timeline then blocks until the clock catches up. Restarting does
not help, since 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 never consult the oracle, so they pick a timestamp near the
clock and the acknowledged write stays invisible.

The frontend read-then-write path can produce such a target today. It writes
at the frontier its subscribe observed, and for a selection over a
materialized view with a `REFRESH` option 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 ceiling `check_runaway_write_ts` already
measures against, is refused before the append and surfaces as a statement
error. This is a backstop rather than a fix. The right answer is for the
caller to take its timestamp from the oracle and use the frontier only to
certify what it read. 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 degrades to that log line in production. Boot reports a catalog upper
that is already past the bound, the one channel that survives a restart and
was silent. It 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.

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

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

@@ -278,6 +278,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

@@ -1952,3 +1952,79 @@ fn test_rejected_in_non_writable_transaction() {
         .get::<_, i32>(0);
     assert_eq!(count, 0, "no rejected write may have landed");
 }
+
+/// 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");
+}