adapter: collect durable replica hydration history - #38512
Conversation
Add the internal durable table for successful replica hydration episodes, including object counts and nullable memory and disk peaks. Protect the table from bootstrap resets and forced replacement migrations. Document episode boundaries, process-lifetime peak semantics, retention, and current failure-reporting limits.
Extend the hydration history sweep to record the latest completed replica episode after collecting per-object rows. Derive episode boundaries from connected object hydration intervals and wait for every configured process before snapshotting resource peaks. Use the maximum cgroup memory peak and filesystem or swap peak across processes, retain rows with the existing 30-day policy, and cover incremental hydration, restart durability, deduplication, and retention.
47d61fc to
03e5ca1
Compare
Replica hydration is replica-wide, so filtering through the catalog to indexes and materialized views omitted metric sinks and system indexes. An omitted export could also split one continuous hydration period into separate episodes.\n\nBuild episodes from every non-transient compute export instead. Transient exports remain excluded because the collector's own subscribe would otherwise prevent the completeness check from succeeding.
Describe the replica table in terms of hydration episodes throughout the catalog and ontology. Keep the user-facing section focused on episode boundaries, retention, and resource interpretation.\n\nReduce the design document to the replica-level additions and refer to the object history design for shared collection mechanics. Restore the original collection-setting description, which is sufficient for the shared collector.
Describe replica hydration episode boundaries and counts in terms of maintained compute exports, including the lifetime of referenced catalog objects. Align the companion designs with installed-to-hydrated intervals over all non-transient exports. Add a system-only testdrive assertion that compares the durable episode membership with the replica's visible non-transient exports.
Replica history includes system exports, so both quickstart.r1 and hydration_history.r1 record episodes and fresh identities when materialized restarts. Limit the assertion to the cluster under test.
| @@ -0,0 +1,195 @@ | |||
| # Durable Replica Hydration History | |||
There was a problem hiding this comment.
we need to cut this down quite a bit, we can rever to the other hystory design doc for reference, and really only describe here the bits we add
There was a problem hiding this comment.
Done in c453b5c. The replica design now references the durable object hydration history design for shared collector mechanics and is limited to replica episode boundaries, resource interpretation, and table shape.
|
|
||
| ## `mz_replica_hydration_history` | ||
|
|
||
| The `mz_replica_hydration_history` table samples successful replica hydration |
There was a problem hiding this comment.
I think we can call these replica hydration episodes
There was a problem hiding this comment.
Done in c453b5c. The relation description and field semantics now call these replica hydration episodes.
| before every preceding object has hydrated belongs to the same component. The | ||
| component finishes when all surviving object intervals have hydrated. | ||
|
|
||
| Collection is disabled by default. Setting |
There was a problem hiding this comment.
need to cut these operational concerns and also cut this doc section smaller overall
There was a problem hiding this comment.
Done in c453b5c. I removed rollout, introspection, migration, and collector-concurrency details. The section now keeps the user-facing episode boundaries, retention, sampling, and resource interpretation.
| independently to each process, so the table records the largest process peak | ||
| rather than a sum of peaks that may not have occurred at the same time. | ||
|
|
||
| Retention is eventual across concurrent `environmentd` processes. Collectors |
There was a problem hiding this comment.
also all these operational concerns, just report the simpler user facing facts: same as for the other history table we have in here already look at it's length and shape and contents for guidance here
There was a problem hiding this comment.
Done in c453b5c. This now follows the length and shape of the object hydration history section, with two short paragraphs and the relation fields.
| ); | ||
|
|
||
| /// How often to sweep replicas for completed object hydration episodes. | ||
| /// How often to sweep introspection-enabled replicas for completed hydration history. |
There was a problem hiding this comment.
no need to talk about the introspection-enabled detail here
There was a problem hiding this comment.
Done in c453b5c. I removed the introspection-enabled detail.
| "hydration_history_collection_interval", | ||
| Duration::ZERO, | ||
| "How often to record completed object hydration episodes. A zero duration disables collection.", | ||
| "How often to record completed object hydration and replica hydration samples. A zero duration disables collection. Replicas with compute introspection disabled are skipped.", |
There was a problem hiding this comment.
the original description was fine here please restore
There was a problem hiding this comment.
Done in c453b5c. The original setting description is restored.
| } | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
why'd we need this change?
There was a problem hiding this comment.
The active sweep owns a PeekClient that can cache an Arc to the batching timestamp oracle without keeping the coordinator command channel open. Dropping AbortOnDropHandle requests cancellation but is not a completion barrier. Build 132852 exposed the race when the oracle worker stopped first and panicked at batching_oracle.rs:115 because the sweep still held a sender. The abort and await in 34e3b19 releases the sweep oracle client and outstanding read timestamp while global_timelines still owns the oracle worker.
| self.0.inner.is_finished() | ||
| } | ||
|
|
||
| /// Aborts the task, then waits for it to release its owned resources. |
There was a problem hiding this comment.
why these changes?
There was a problem hiding this comment.
A plain JoinHandle::abort would lose the wrapper’s abort-on-drop protection. The wrapper’s Future also intentionally remains pending after cancellation, so the coordinator cannot use it as a join barrier. The consuming abort_and_wait API in 34e3b19 preserves abnormal abort-on-drop behavior and provides ordered normal teardown. Its unit test verifies that task-owned resources are dropped before the call returns.
| JOIN mz_internal.mz_object_global_ids AS ids ON ids.global_id = t.export_id | ||
| JOIN mz_catalog.mz_objects AS o ON o.id = ids.id | ||
| WHERE t.export_id LIKE 'u%' | ||
| AND o.type IN ('index', 'materialized-view') |
There was a problem hiding this comment.
do we need this join and check, what other types could there even be running on there, and we probably want all the running objects anyways
There was a problem hiding this comment.
Removed in 155226c. Replica episodes now use every export with export_id NOT LIKE t%, with no catalog join or object-type filter. This includes system and introspection exports as well as metric sinks. Transient exports stay excluded because the collector subscribe is transient and would otherwise wait on itself. dd64244 adds a system-only test that checks the durable episode count against every visible non-transient export. The object-level collector remains limited to user indexes and materialized views.
QA LLM Review1. MEDIUM -- Widening replica episodes to system exports makes
|
|
The restart test finding was real, but is already fixed in 6ebb99e. I’m leaving the docs as is. |
ggevay
left a comment
ggevay
left a comment
There was a problem hiding this comment.
Some comments from Claude for now. I'll try to take a deeper look tomorrow.
| 'hydrated'::text AS status | ||
| FROM episode AS e | ||
| CROSS JOIN resources AS r | ||
| WHERE (SELECT bool_and(hydrated) FROM objects) |
There was a problem hiding this comment.
The gate is stricter than the episode definition needs, and since only the latest completed episode is ever recorded, the strictness is lossy rather than just late. An export installed after the candidate's finished_at cannot belong to the candidate (its interval starts after every interval in the candidate ended, so by the definition it opens the next episode), yet it blocks the write. With dataflows created one after another, every sweep that lands during a hydration records nothing, and once that hydration ends the previous episode is no longer the latest and is never recorded.
The predicate the definition actually needs is
NOT EXISTS (
SELECT 1 FROM objects AS o
WHERE NOT o.hydrated AND o.installed_at <= e.finished_at
)An unhydrated export installed at or before the candidate's finish would extend it; one installed after it starts the next episode. The exposure to late-visible worker rows is the same either way (an entirely invisible export is absent from objects under both predicates).
Separately, and true under either predicate: one permanently unhydrated export (a sink whose broker is down, a materialized view over a stalled source, the mv_wmr_stuck fixture in hydration-status.td) suppresses every later replica episode on that replica, restart episodes included, because its open interval connects to everything installed after it. Worth a sentence in the design doc's limitations; "successful episodes only" does not say that one stuck export silences the table for the whole replica.
| Ok(Err(error)) => { | ||
| self.observe_mutation(step, "error"); | ||
| if step == "collection" && matches!(&error, AdapterError::ReadThenWriteContention) { | ||
| if step.ends_with("collection") |
There was a problem hiding this comment.
nit: step.ends_with("collection") ties the trailing-frontier hint to the spelling of four metric labels, and a renamed label moves a step in or out of this branch with no compile error. A small Step enum with label() and is_collection() would carry both.
| ) | ||
| } | ||
|
|
||
| /// A bounded batch of replica history rows that have aged out. |
There was a problem hiding this comment.
nit: finished_at is nullable by design (the PR body reserves it for failed and OOM-killed episodes), and a NULL there would make the row immortal here. The object variant's doc calls that out; the same sentence belongs here. Keying on started_at instead would break the shared-cutoff symmetry (a row with started_at < cutoff <= finished_at would be deleted and re-collected every sweep); coalesce(finished_at, started_at) < cutoff would keep it and is identical for every row collection writes today.
| ) | ||
| } | ||
|
|
||
| /// What one sweep needs to run its mutations against the history table. |
There was a problem hiding this comment.
nit: "against the history table" here, and "one bounded batch" on retain below, are now two tables and one batch per table.
| ); | ||
|
|
||
| /// How long to retain completed object hydration episodes. | ||
| /// How long to retain completed object and replica hydration episodes. |
There was a problem hiding this comment.
nit: HYDRATION_HISTORY_COLLECTION_INTERVAL above still says "object hydration episodes" in both its rustdoc and its description, while this one now names both tables.
| ), | ||
| ( | ||
| "object_count", | ||
| "The number of maintained compute dataflows in the hydration episode.", |
There was a problem hiding this comment.
nit: after 155226c this count includes the replica's introspection log indexes (the system-only episode in the testdrive is exactly that), which a reader of "maintained compute dataflows" will not expect on an idle replica; "compute dataflows, including introspection indexes" or similar would say it. Also, the mz_object_hydration_history section of mz_internal.md keeps its cross-process clock-skew sentence, but the new section has none, although the replica interval spans more processes.
| deadline = time.time() + 120 | ||
| replica_before = [] | ||
| while time.time() < deadline: | ||
| replica_before = replica_episodes() |
There was a problem hiding this comment.
nit: replica_before is captured at first sighting, and the two exact-count assertions after the restart (len(replica_fresh_ids) == 1, len(replica_settled_ids) == len(replica_before_ids) + 1) assume nothing else was recorded before the kill. That holds today only because the DDL lands in the same logging bucket as the log indexes' hydration, so they form one island; re-reading replica_before right before c.kill makes it independent of that timing.
3 participants
Motivation
Materialize records successful hydration for individual indexes and materialized views, but it does not retain a replica-level history. This makes it difficult to answer when a replica finished hydrating, how many maintained compute exports participated, or what process resource high-water marks were visible when the episode was recorded.
Description
This change adds
mz_internal.mz_replica_hydration_historyand extends the existing durable hydration-history sweep to populate it.A replica hydration episode is a connected set of overlapping export intervals. Each live non-transient compute export contributes an interval from its earliest worker installation to its latest worker hydration. The collector waits for every visible worker of every live non-transient export to hydrate. A gap between intervals means that the replica was fully hydrated before the next export was installed, so the next interval starts a new episode. Transient exports are excluded because the collection query creates one itself and would otherwise wait on its own hydration. The object-level collector remains limited to user indexes and materialized views.
Each row records replica and cluster IDs, episode boundaries, the number of compute exports in the episode, status, and the largest resource high-water marks reported by any replica process. Memory uses cgroup
memory_peak. Disk uses sampledstatvfsfs_used_peakwhen a scratch filesystem is present, with cgroupswap_peakas the fallback. Managed resource limits apply independently to each process, so the maximum across processes is relevant. Summing process peaks could combine values that did not occur simultaneously.Resource observations have process-lifetime scope through collection. They are associated with an episode but are not bounded by its timestamps. Work after hydration can raise them, and a later episode can retain an earlier process peak.
Collection is snapshot sampling rather than an event log. Each sweep records only the latest completed episode visible in its snapshot. Intermediate episodes can be missed, and exports retracted before collection leave no evidence. Only successful episodes are currently recorded. Process-local clock skew can merge intervals that did not overlap in real time.
The collector reuses the replica-targeted timestamped read-then-write path. Exact-timestamp OCC and an anti-join make concurrent collection idempotent. A monotonic history guard prevents a later snapshot from recording an episode that precedes or overlaps retained history. Retention is bounded and eventually consistent across concurrent collectors.
The table survives normal
environmentdand replica restarts and is protected from forced replacement migrations. Collection remains disabled by default throughhydration_history_collection_interval. Coordinator shutdown explicitly aborts and waits for an active sweep before releasing timestamp-oracle workers, which preserves dependency ordering for timestamp-oracle clients cached by the sweep.Tests
The hydration testdrive coverage exercises the production collector. It verifies that a system-only replica episode includes every visible non-transient export while excluding the collector's transient subscribe. It also checks single-export episode boundaries and compares persisted memory and disk values with independently published resource metrics.
The restart workflow verifies that retained rows survive
environmentdand replica restarts, a restart creates a fresh episode identity, and repeated sweeps do not duplicate an identity. Unit coverage checks episode query construction, latest-episode selection, process-count gating, replica rotation, task cancellation, and migration policy. Catalog expectations cover the relation, comments, OID, ontology metadata, and system catalog plans.Release note
This release will add
mz_internal.mz_replica_hydration_history, an internal durable history of successful replica hydration episodes and their observed process resource high-water marks.