adapter: move arrangement size snapshotting off the coordinator loop

[leedqin]

Remove these sections if your commit already has a good description!

Motivation

mz_object_arrangement_size_history collection is currently disabled in
staging and production via the arrangement_size_history_collection_interval
LD flag: the snapshot ran its persist reads and row building inline on the
coordinator main loop and dominated the "Slow Coordinator Messages"
dashboards during canary verification
(https://materializeinc.slack.com/archives/CTESPM7FU/p1778508034797449).

This PR fixes CNS-42, plus two known issues with the same feature.

Fixes SQL-218
Fixes SQL-271

Description

Three commits, best reviewed individually:

1. Move snapshotting off the coordinator loop. Mirrors the
   storage_usage_fetch/storage_usage_update split: the snapshot handler
   only resolves catalog ids on the main loop, a spawned task does the
   persist snapshots and record prep, and a new ArrangementSizesWrite
   message carries the prepared records back for the cheap
   stamp-pack-append phase. The read timestamp is taken inside the task
   immediately before the reads; a failed snapshot skips the cycle and
   reschedules, as before. No change to the data produced.

2. Don't record stale sizes after a restart (SQL-218). The backing
   collections retain pre-restart rows until the new introspection
   subscribes deliver and the shards are reconciled, so post-restart
   snapshots recorded outdated sizes tagged hydration_complete = true.
   Each introspection subscribe now tracks when it first appended data in
   this process, and the snapshot only records rows for replicas whose
   sizes and hydration subscribes first delivered at least 10s ago. The
   margin covers the collection manager's batched flush and the oracle
   read timestamp trailing the wall clock. Observable change: a short
   history gap after restarts and replica reconnects, instead of poison
   rows that lived until the 7-day pruner.

3. Report sub-10 MiB arrangements (SQL-271). The subscribe's HAVING
   floor silently dropped small objects, contradicting the column docs.
   The floor is removed and the round-to-nearest-10-MiB quantization
   stays, so arrangements below 5 MiB report a size of 0 and every
   arranged object is present, with churn behavior unchanged. The history
   snapshot skips size-0 rows, since the floor had been doubling as
   history volume control. Observable change: mz_object_arrangement_sizes
   gains rows for small objects; the size column comment now describes
   the actual behavior.

Verification

• Record preparation is extracted into a pure function with unit tests
  covering the hydration flag, freshness filtering, zero-size skip, and
  malformed input.
• New restart workflow arrangement_sizes_stale_snapshot_after_restart
  (test/restart/mzcompose.py): five kill/restart rounds at a 500ms
  collection interval asserting no post-restart history row carries a
  pre-restart size. Without commit 2 it fails with 299 stale rows in the
  first round.
• Extended test/testdrive/arrangement-sizes.td: a sub-10 MiB index
  appears in the live collection with size 0 on both replicas and never
  lands in the history.
• Measured locally at a 10s interval: the snapshot handler's main-loop
  time (mz_slow_message_handling) drops from ~163ms, the full collection
  time by construction, to ~61µs over 31 cycles, while the collection work
  runs on the spawned task.

Rollout: after this ships in a release, re-enable
arrangement_size_history_collection_interval for a canary environment in
LaunchDarkly and confirm the "Slow Coordinator Messages" panel stays flat
before restoring the default.

[def-]

Needs a rebase at least. Test changes lgtm! I'm less familiar with the adapter code, so will leave that for the team! (I'll come in complaining with the QA review in ~1.5 h if it finds anything on top)

[def-]

MEDIUM - replica restarts can still poison arrangement-size history before the subscribe error is processed

The PR adds IntrospectionSubscribe::first_data_at and gates
mz_object_arrangement_size_history snapshots on
fresh_introspection_replicas so rows from a previous environmentd or replica
incarnation are not recorded. That freshness is reset only in
reinstall_introspection_subscribe, after an introspection subscribe returns
ERROR_TARGET_REPLICA_FAILED (src/adapter/src/coord/introspection.rs:421-426,
447-453). Cluster status events already tell the coordinator that a replica
went offline or restarted, but message_cluster_event only updates replica
status and does not clear first_data_at for that replica
(src/adapter/src/coord/message_handler.rs:924-1040).

That leaves a reachable stale-history window on normal replica failure:

1. A replica has been reporting arrangement sizes for more than the 10s margin,
   so its arrangement-size and hydration subscribes are considered fresh.
2. The replica dies or restarts. The unified introspection storage collection
   still intentionally retains the previous subscribe's rows until the
   replacement subscribe reports data.
3. Before the arrangement-size subscribe's failure response is handled and
   first_data_at is reset, an arrangement-size history tick runs. The new
   arrangement_sizes_snapshot reads fresh_size_replicas from the stale
   first_data_at, snapshots the retained old rows, and sends them to
   ArrangementSizesWrite (message_handler.rs:487-559).
4. The write stamps those old measurements with a new collection_timestamp
   and appends them to mz_object_arrangement_size_history (message_handler.rs:578-626).

The resulting rows are the same kind of poison rows SQL-218 is trying to avoid:
they can be marked hydration_complete = true, look like post-restart
measurements, and persist until the retention pruner removes them. The new
restart workflow covers environmentd restarts, where all first_data_at values
start as None, but it does not cover a replica reconnect while environmentd
continues running. This is also distinct from SQL-379, which is about cleanup
after the replacement subscribe snapshots empty. Here the stale row can be
recorded before the subscribe error/reinstall path runs at all.

A robust fix would invalidate freshness for all introspection subscribes on the
replica as soon as the coordinator observes an offline status or restart-count
change, or carry a per-subscribe generation/first_data_at token into the
snapshot task and revalidate it on the coordinator before appending records.

[ggevay]

Looks good overall! Claude and I have some minor comments.

And how should we handle the flag flips in LD, i.e. turning arrangement_size_history_collection_interval back on? Someone will need to:

• turn it on in staging and prod canaries after the fix is rolled there
• turn it on in prod after we verify in staging and prod canaries that this indeed fixed the slow coordinator messages issue and the fix is rolled out to prod.

You could write about this in the release-discuss channel, either asking the release manager to do this, or just noting that you'll do this yourself.

[ggevay]

I'm not sure if this schedule_arrangement_sizes_collection() call is needed here. When envd comes out of read-only mode, it restarts.

[ggevay]

And a nit:

The catalog server is not writable in read-only mode.

I'd say "builtin collections are not writable in read-only mode".

[ggevay]

I think the same comment change is needed on mz_object_arrangement_size_history as well.

[ggevay]

This assertion can false-positive, and the CI failure on this build is an instance of it, rather than the fix being broken. For a row to be recorded at all, the replica's size subscribe must be fresh, and freshness is only set once its first batch is appended, which is also when the deferred delete clears the pre-restart shard rows. So the flagged 120 MiB rows are genuine post-restart measurements: the indexes hydrate at their pre-insert state (the dataflow as-of predates the post-restart INSERT), correctly report hydration_complete = true, and a snapshot catches them before the insert has propagated into the arrangements. That state is indistinguishable by value from a stale row.

Suggestion: drop a couple of the indexes right after each restart and assert those get zero post-restart history rows. Without the fix their lingering shard rows would get recorded; with the fix nothing can be, since the new subscribes never report a dropped object. No value ambiguity.

That would also remove the need for the cumulative inserts: by the last round they add up to ~7 GB raw (likely 10+ GB actual) of arrangements on the 16 GB agents, and that memory pressure widens exactly this timing window (the CI failure hit in round 3 as the machine slowed down). If you'd rather keep the size-mismatch approach, deleting the previous round's insert before adding the new one would bound the peak instead.

[ggevay]

Nothing pins the since across this gap, so the reads rely on both collections being retained-metrics objects, whose since lags the upper by metrics_retention (default 30 days) rather than tracking it closely. Could you record that assumption here, so it's visible if someone later repoints this at a normally-compacted collection? E.g.:

// No read hold is taken, so the reads rely on both collections being
// retained-metrics objects, whose since lags the upper by
// `metrics_retention` rather than tracking it closely. If the since
// still overtakes `read_ts`, the snapshot fails, and the cycle is
// skipped and retried at the next interval.

[ggevay]

nit: The two error paths above use let _ = internal_cmd_tx.send(...) while this one warns, and the comment says outliving the receiver is not an error. Consider one treatment for all three (the silent form, going by the comment).

[ggevay]

nit: The real mz_compute_hydration_times.time_ns column is uint8, so Datum::UInt64 would keep the fixture faithful to the schema. Works today because the code under test only checks is_null.

[ggevay]

nit: The help text of mz_arrangement_sizes_collection_time_seconds ("…and prepare history-table updates…") now slightly overstates the timer's scope, since row packing moved to arrangement_sizes_write outside it. Fine to fold in only if you touch metrics.rs anyway.

[ggevay]

All reachable failure modes of snapshot indicate invariant violations for these two collections: ReadBeforeSince cannot happen short of a read-policy bug, since both collections are retained-metrics objects whose since lags the upper by metrics_retention (default 30 days), IdentifierMissing cannot happen for a builtin collection, and a decode error would mean error rows in an introspection collection. Consider soft_panic_or_log instead of warn! here and in the arm below, so a violation is loud in CI and reaches Sentry in production. Since soft_panic_or_log returns in production, keep the ArrangementSizesSchedule send after it, so production degrades to a skipped cycle rather than the collection stopping.