adapter: collect durable object hydration history

[aljoscha]

Stacked on #38346.

Motivation

Populates mz_internal.mz_object_hydration_history, which the preceding rung adds but leaves empty.

Design doc: 20260817_durable_object_hydration_history.md

Description

A single-flight sweep visits one user replica per interval, skipping replicas with introspection disabled and managed replicas whose rolled-up status is not Online. It installs an internal subscribe, aggregates the replica's worker rows per dataflow, anti-joins against history, and writes missing rows through timestamped OCC read-then-write. Retention retracts successive bounded batches on the catalog server even when collection from the user replica fails.

The main correctness boundaries are:

• Idempotence across environmentd processes. The subscribe reads its own target table. Concurrent collectors may compute the same candidate, but the oracle selects a write timestamp, the subscribe frontier certifies completeness below it, and a loser recomputes after observing the winner's row.
• Replica-scoped worker aggregation. Installation adds one row per worker with a nullable hydrated_at, and the replica owns and resets the collection as a unit. HAVING count(*) = count(hydrated_at) waits for every row visible at the OCC read timestamp, and max(hydrated_at) normally includes the persist sink's elected worker, whose stamp covers the snapshot write. Per-process logging clocks can place an ahead worker beyond the sampled timestamp. The collector explicitly accepts that race instead of depending on configured worker counts.
• One row per dataflow. The table stores the dataflow id under object_id; mz_object_global_ids is used only to filter to user indexes and materialized views. This preserves separate episodes when one catalog item has several live dataflows.
• Fleet-safe scheduling. Fires use a stable SHA-256 seed over the full environment id, so a fleet-wide dyncfg does not synchronize environments, including regions and ordinals in one organization, on the same absolute boundary.
• Bounded retention catch-up. The LIMIT is inside a derived table because top-level RowSetFinishing is not part of the OCC write selection. Each sweep retracts one bounded batch, so collection gets another turn while later sweeps drain the finite fixed-cutoff backlog without exceeding max_result_size.
• Background isolation. Background mutations do not take the process-wide OCC semaphore, because their replica-targeted subscribe may take arbitrarily long to hydrate. The sweep is single-flight, so this adds at most one concurrent read-then-write. Its task is coordinator-owned and abort-on-drop, while fallible coordinator calls handle concurrent shutdown. The entry point requires an actual system-table target and system-only dependencies, separating it from session DML.
• Operational visibility. Internal metrics report collection and retention outcomes, affected rows, complete sweep duration, and full retention batches without cluster, replica, or object labels. Repeated full batches indicate that retention may not be keeping up with the configured sweep interval.

The read-then-write path gains explicit background-caller contracts for replica targeting, system-table writes, system-read dependency validation, ownership, cancellation, exact-timestamp writes, and trivial top-level row-set finishing. Background subscribes are coordinator-owned, write no mz_subscriptions row, and count against an internal gauge. Dropped coordinator responses unwind as errors, including a background compute-client lookup racing shutdown.

Known limitations: a process whose logging clock is ahead can place its hydration row beyond the OCC read timestamp, so the durable finish can precede the latest worker's finish. Separately, the oracle chooses the write timestamp and the replica-targeted subscribe frontier certifies the read. If the replica's clock trails environmentd by more than an introspection interval, the frontier can certify each target only after the oracle has advanced past it. The resulting conflict loop ends at the sweep timeout or retry budget. It records nothing and logs a diagnostic naming the likely frontier.

Verification

• test/testdrive/hydration-status.td covers timestamp ordering, multiple replicas, idempotence, replica removal, and retention.
• test/restart/mzcompose.py verifies one pre-restart episode survives unchanged, one fresh episode is collected, neither is duplicated, and a materialized view on a two-worker replica records the all-worker maximum rather than worker 0.
• Unit tests cover replica rotation, all-worker completeness, filter placement, full-environment scheduling offsets, and forced migration policy.

Collection defaults off in production and is runtime configurable. The rollout plan is CI on merge, then staging, then production, one week apart.

This release will record completed hydration of indexes and materialized views in mz_internal.mz_object_hydration_history when collection is enabled.

Closes: SQL-644

[linear-code]

SQL-632

SQL-644

[aljoscha]

potentially brittle?

[aljoscha]

Removed the invariant instead of documenting it. ActiveSubscribe::introspection_session_uuid() is now the single place that decides whether a subscribe appears in mz_subscriptions, and it hands back the uuid the row needs. pack_subscribe_update takes that uuid as a parameter, so it can no longer be called without one, and both call sites became a match on Some/None rather than an internal check plus an expect.

[aljoscha]

let's do an exhaustive match please, so we don't sneak in bugs in the future

[aljoscha]

Done, exhaustive match on the owner with the background arm spelled out.

[aljoscha]

better to return an internal error or sth rather than panic

[aljoscha]

or maybe we can swing some generalization in the api shape? don't change if that would be too big a diff, though, please

[aljoscha]

Both of these are fixed by the reshape below, see the next comment. No panic left: the blind path now destructures WriteAttempt::Session with a let ... else that soft-logs and answers Indeterminate, which the caller already turns into an internal error.

[aljoscha]

Took the generalization, it stayed small. Command::AttemptWrite had a conn_id: Option<ConnectionId> next to a write_ts: Option<Timestamp>, where (None, None) was the invalid combination the expect was guarding. That is now one enum:

pub enum WriteAttempt {
    Session { conn_id: ConnectionId, write_ts: Option<Timestamp> },
    Background { write_ts: Timestamp },
}

Group commit picking the timestamp needs a connection to answer through, so that case is only reachable from Session and the invalid combination is now unrepresentable. Diff is about 60 lines across the command, its handler, and three call sites.

[aljoscha]

please don't document other places in the code here, just our facts and invariants

[aljoscha]

Trimmed to just the invariant: a background subscribe is always internal, since there is no session to attribute a row to. The cross-reference is gone, and with the change above nothing depends on that invariant to avoid a panic anymore.

[aljoscha]

"the" makes it sound like there is always one, but that's not the case, right?

[aljoscha]

Right, there often isn't one. Now reads "Hydration-history sweep, while one is in flight."

[aljoscha]

don't say "two" here, and document the requirements, if any, on the respective variant

[aljoscha]

Rewrote it. No "two", and the requirements now sit on the variant that carries them: Background documents that the caller builds the statement itself and must tolerate a log relation that is sealed empty, which is how a replica with introspection disabled presents one.

[aljoscha]

lose the "gates" here, just describe what it does please

[aljoscha]

Reworded to say what it does: dependency_policy decides which relations the selection may read, and both policies reject mz_now() in the transitive dependencies.

[aljoscha]

can't we somehow get that enforcement back? Also, please make more succinct

[aljoscha]

Got the enforcement back rather than documenting the obligation. The validator was doing two things at once: rejecting mz_now() in view bodies, and checking that every dependency is a readable user object. Only the second is a user-DML rule, so it now takes a DependencyPolicy:

• UserDml behaves as before.
• SystemReads still walks the whole transitive closure and still rejects mz_now() anywhere in it, but does not object to system relations.

So background callers get the mz_now() guarantee enforced instead of promised, and the doc on background_read_then_write is down to two lines pointing at RtwCaller::Background. The traversal cap is 100k, so walking the system closure is not a concern.

[aljoscha]

maybe soft log or error, or whatever we use here, and also a pending() after that, so we don't accidentally panic

[aljoscha]

Done, soft_panic_or_log! followed by pending(). Taking the process down is worse than one stuck session, and under soft asserts it still fails loudly in CI. The background arm stays silent since losing that race is expected during shutdown.

[aljoscha]

sweeps are replica targeted, right, but the retention sweep is not, yes?

[aljoscha]

Correct, and the comment was misleading. Collection visits one replica per cycle, retention is not replica-targeted at all, it runs on the catalog server so it keeps working when there are no user replicas. Said so in both comments.

[def-]

QA LLM Review

1. MEDIUM -- max(installed_at) <= min(hydrated_at) permanently rejects episodes on multi-process replicas

src/adapter/src/coord/hydration_history.rs:315

The completeness guard compares installed_at against hydrated_at across
workers, but those timestamps are stamped from each clusterd process's own
wall clock. On a scale>1 replica, any inter-process clock skew larger than the
hydration duration makes max(installed_at) > min(hydrated_at), and since those
log values never change the episode is rejected on every later sweep too. Large
replicas therefore silently record nothing for fast-hydrating objects, and no
test configuration can catch it because they all run a single process.

Details

mz_compute_hydration_times_per_worker.installed_at / hydrated_at come from
DemuxHandler::event_timestamp (src/compute/src/logging/compute.rs:835),
which reads self.time: Instant elapsed added to a SystemTime::now() anchor
captured per worker at logging init
(src/compute/src/logging/initialize.rs:57). A scale=N replica is N separate
clusterd processes — ReplicaLocation::workers() is
allocation.workers * num_processes (src/controller/src/clusters.rs:306) — so
the worker rows in one group carry N independent anchors. Sizes scale=2
through scale=32 are real allocations (src/catalog/src/config.rs:203), and
their processes are separate pods, where tens to hundreds of milliseconds of
skew is ordinary.

Concretely, two processes whose anchors differ by 200ms, and an index that
hydrates in 50ms: the lagging process reports (I, I+50), the leading one
(I+200, I+250). max(installed_at) = I+200 > min(hydrated_at) = I+50, so the
HAVING drops the group, permanently. mz_hydration_statuses keeps reporting
the object hydrated while mz_object_hydration_history never gets a row, so the
omission is invisible.

Same root cause, secondary effect: when the guard does pass,
installed_at = min(...) (line 304) and finished_at = max(...) (line 306) are
read off different anchors, so the recorded interval is inflated by the skew —
the same class of wrong duration the guard exists to prevent, at a smaller
magnitude.

Neither new test can see this. test/testdrive/hydration-status.td uses
scale=1,workers=1 and test/restart/mzcompose.py uses scale=1,workers=2;
both are one process, and mzcompose runs every process on one host, so the
anchors always agree.

Direction for a fix: the property the guard wants is "these worker rows come
from one install", which cross-process wall-clock ordering only approximates.
Bounding the spread instead — max(installed_at) - min(installed_at) within the
introspection interval — separates "one broadcast install, clocks slightly off"
from "rows from either side of a restart" without requiring the processes'
clocks to agree with each other. A restart generation or epoch carried in the
log would remove the wall-clock dependency entirely, and would also let
installed_at/finished_at be taken from a single worker rather than mixed
across anchors.

[aljoscha]

Thanks, this one is valid and I have fixed it.

I verified the premise before changing anything. initialize_logging captures SystemTime::now() as the anchor per worker at logging init, and ReplicaLocation::workers() counts allocation.workers * num_processes, so on a scale>1 replica one group's worker rows genuinely carry independent anchors. Comparing max(installed_at) against min(hydrated_at) therefore measures cross-process skew as much as elapsed work, and since the log values never change the rejection is permanent. The invisibility is the worst part: the object reports hydrated everywhere else while the history stays empty, on exactly the large replicas this feature is for.

Fixed by taking your suggested direction. The guard now bounds the spread of the installation stamps against each other rather than comparing an installation to a completion:

HAVING count(*) = {workers}
   AND count(*) = count(t.hydrated_at)
   AND max(t.installed_at) - min(t.installed_at) <= INTERVAL '60 seconds'

That still separates "one broadcast install, clocks slightly off" from "rows from either side of a re-install", without requiring the processes' clocks to agree with each other. The tolerance errs generous on purpose: too tight silently records nothing, too loose records one episode whose duration looks like an outlier, and the second failure is the one you can see.

Worth adding for whoever picks this up: a re-install is per replica, not per process. Instance::rehydrate_replica removes and re-adds the whole replica with a bumped epoch, so every process gets fresh stamps together, which is why bounding the spread is sufficient here.

On your secondary point, installed_at = min(...) and finished_at = max(...) do still read off different anchors, so a recorded interval is inflated by the skew. I left that as a documented imprecision rather than fixing it, since it is bounded by the skew itself, which is small next to a hydration worth recording. I also recorded your epoch suggestion in the design doc as the principled fix: an install generation in the log would remove the wall-clock dependency for both the bound and the interval, and would let one worker's stamps be used instead of a mix. That is compute-side work, so I have not pulled it into this PR.

Also noted that neither test can see this, and I have not added one that can, since it needs a genuinely multi-process replica with skewed clocks. The unit test now at least pins that the query compares installations to each other and never min(t.hydrated_at), so the regression cannot come back silently.

[aljoscha]

pleaes remove this one, we'll remove the lock path so this will rot quickly

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

Removed the section, along with the lock-path wording in the design and implementation comments. The runtime target check remains as the boundary of the coordinator-owned system-maintenance API, without documenting the path being retired.

[aljoscha]

what is this test even doing, won't history collection fail quickly and loudly if something is not allowed there?

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

Removed this synthetic 99-line test. The hydration workflows exercise the permitted system source and table path directly, and the policy itself is now a small item-ID and item-type check rather than test scaffolding around a generic override.

[aljoscha]

what is this, aren't these basically all of them? I think what's important here is:

• we write to a table and use the oracle (given by this being a read-then-write)
• for system we allow reading from other sources, which yes, are not in the same linearizability domain, but we just accept that this is the case, no. What else could we do. So maybe record that invariant or fact

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

Simplified around the actual contract. User DML retains its same-time-domain restrictions. System maintenance accepts readable system catalog dependencies across time domains, while still rejecting secrets, connections, sinks, metric sinks, and indexes. Common temporal checking and transitive dependency traversal now happen once before the policy-specific predicate.

[aljoscha]

let's state the fact, it doesn't have to be lowered, it's just that at our stage we can't have one

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

Reworded and inlined the check. This stage emits raw diffs and therefore requires trivial top-level finishing. It no longer claims callers must lower finishing first.

[aljoscha]

not sure these tests add much value

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

Removed both helper-only tests and inlined the small runtime checks. The supported system-table path remains covered by the hydration workflows, while the deleted tests only exercised extracted predicates rather than execution behavior.

[aljoscha]

is this whole change "adapter: skip unready hydration replicas" worth the hassle? What does this buy us?

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

Kept the optimization but reduced it to one transient scan per sweep. The coordinator builds a set of installed hydration subscribes that have not delivered data, then admits replicas absent from that set or reported Online. Missing subscribes preserve behavior when introspection subscribes are disabled, and Online covers delayed invalidation after replacement data arrived.

The cost is O(S log U + R log U) per enabled sweep, with S subscribes, U unready replicas, and R candidates. There is no persistent state or per-replica subscribe scan. The standalone helper, bespoke test, long comment, and readiness design-doc discussion are gone.

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

Follow-up after narrowing the production scope: unmanaged replicas are test-only and never receive orchestrator status events, so distinguishing ready from unreachable unmanaged replicas did not justify coupling this collector to introspection-subscribe state. I removed the scan and helper entirely.

Managed replicas are now admitted only when their rolled-up status is Online. Unmanaged test replicas remain eligible unconditionally, so working ones can still record history. A deliberately unreachable test replica may consume the bounded mutation timeout, which is an accepted test-only tradeoff.

[def-]

QA LLM Review

1. MEDIUM -- hoisting entry.uses() out of the type match widens UserDml dependency traversal and rejects DML that reads a webhook table

src/adapter/src/coord/read_then_write.rs:318

The refactor moved ids_to_check from inside the Func | View | MaterializedView arm to an unconditional let ids_to_check = entry.uses();, so tables are no longer terminal in the traversal. A user table's resolved_ids include everything its DDL names, and for CREATE TABLE ... FROM WEBHOOK ... CHECK (WITH (SECRET s) ...) that includes the secret, which DependencyPolicy::UserDml rejects. Any INSERT ... SELECT, UPDATE or DELETE whose selection reads such a table (directly or through a view) now fails where it succeeded before.

Details

Before this commit, and on main (src/adapter/src/coord/read_then_write.rs:289), the worklist was only extended for Func | View | MaterializedView; a valid user table ended the walk. Now the walk continues into table.resolved_ids, which is what mz_sql::names::resolve collected from the CREATE TABLE statement. DependencyVisitor::visit_item_name records every ResolvedItemName::Item, and the webhook CHECK clause carries the secret as CreateWebhookSourceSecret::secret: T::ItemName (src/sql-parser/src/ast/defs/statement.rs:965) — the reason CHECK (WITH (SECRET non_existent_secret) ...) fails with unknown catalog item at name resolution (test/sqllogictest/webhook.slt:261). A webhook table is not a source export, so entry.source_export_details() is None (src/catalog/src/memory/objects.rs:2964) and the table itself stays valid; the walk then hits the secret and Source | Secret | Connection => false.

Repro:

CREATE SECRET s AS 'k';
CREATE TABLE wh FROM WEBHOOK BODY FORMAT TEXT
  CHECK (WITH (BODY, SECRET s) body = s);
CREATE TABLE dest (body text);
INSERT INTO dest SELECT body FROM wh;

The INSERT is planned as a ReadThenWritePlan (src/adapter/src/coord/sequencer/inner.rs:2734) and now retires with secret 'materialize.public.s' may not be used in this operation; the selection may refer to views and materialized views, but transitive dependencies must not include sources or source-export tables. Both entry points are affected: the lock path at inner.rs:2877 and validate_selection_dependencies at frontend_read_then_write.rs:444 pass DependencyPolicy::UserDml. A view over wh does not help, since the view's own traversal reaches wh and continues. CREATE TABLE ... FROM WEBHOOK is gated on enable_create_table_from_source, which defaults to true (src/sql/src/session/vars/definitions.rs:2271), and secret-based CHECK is the standard webhook authentication pattern, so this is a live combination. No existing test does DML over a webhook table, so CI stays green.

The widening is also unnecessary for the rest: Type entries only reference other types, and Func has empty references(), so the only reachable behavior delta is this one. Scoping the traversal back to relation-bodied items keeps the simplification and drops the regression:

let ids_to_check = match item_type {
    Func | View | MaterializedView => entry.uses(),
    _ => BTreeSet::new(),
};

Under DependencyPolicy::SystemReads the same hoist is harmless today — builtin tables and sources carry empty resolved_ids — but it is the same unintended change of scope.

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

Addressed in 0efe63a24898, with a narrower correction than the proposed traversal-only fix.

The reported behavior change was real, but webhook tables should not be accepted by user read-then-write. They are TableDataSource::DataSource tables whose updates arrive through storage, outside the table-write transaction and oracle domain. UserDml now requires every relation leaf to be a user TableDataSource::TableWrites table. Views and materialized views still traverse to their relation leaves. Table DDL references such as webhook validation secrets are no longer traversed as read dependencies.

Boundary coverage now rejects both an ordinary webhook table and a secret-authenticated webhook table as a source-backed table, while retaining successful read-then-write through regular tables, views, and materialized views. SystemReads remains the explicit cross-domain policy for coordinator-authored hydration work.

[aljoscha]

seems brittle to hardcode this list? Is there a canonical list or method or sth maybe?

[aljoscha]

Posted by AJ, Aljoscha's coding agent.

There was no existing method with the required distinction: uses() includes table DDL and lifecycle metadata, while is_relation() also includes relation leaves.

Added CatalogItem::query_dependencies in 044c0532d477. Functions and views return uses(). Materialized views return uses() without replacement_target, which is lifecycle metadata and not part of the query definition. Tables, sources, logs, and non-selectable items are leaves. The match is exhaustive, so a new catalog item variant must choose semantics at the catalog boundary. The read-then-write validator now calls this method and carries no item-type traversal list.