compute: retry metric sink collector registration - #38910

Left a comment; I think we can miss metrics with this approach. Do you mind checking this?

Checked this against the SQL-680 occurrence in nightly 18362, "Checks 0dt restart of the entire Mz with forced migrations 2" (panic at services.log line 514103). The collision there came from curated sinks. On boot, environmentd reinstalled them under new transient ids (for replica u1, mz_metric_arrangement_sizes went from t291 to t273), and they share the definition-name label, so both incarnations have one Desc id. The registering worker is sink_id.hashed() % peers (metric_sink.rs:73), so old and new usually land on different workers. Worker::reconcile runs per worker without a barrier, so the new one can register before the other worker drops the old handle. User sinks cannot hit this: their durable id keeps both incarnations on one worker, which drops before it re-creates. The retry converges for this race, so it fixes that path.

1. The concern from def-'s QA review on 09-18 is still open on the current head. The registration guard lives in the operator closure, and the operator has no outputs or capabilities. Once both input frontiers close, for example with a constant-folded source, the closure and the guard drop, and the sink silently stops publishing, taking any in-flight retry with it. Keeping the guard in sink_token, shared with the operator as Rc<RefCell<PendingRegistration>>, still converges and restores unregistering on token drop.
2. A permanent AlreadyReg collision now retries every second forever, with one info! and an unlabeled counter. Today it soft-panics and CI catches it. Suggest escalating after a bound, e.g. soft_panic_or_log! after 30-60 s. The doc claim that a predecessor always eventually drops is not established. The original SQL-680 report (plain restart, u3 missing, only u4 visible) fits neither this race nor a late drop, so rerunning its minimal repro on this branch would show whether that case is permanent.
3. The NOTE at metric_sink.rs:90-96, the doc of try_register_collector_with_dropper and the PR description say nothing sequences teardown before render. Within a worker it is sequenced, and the gap is across workers. Naming that, and noting that user sinks are not exposed, would make the contract clear. Transient ids are also reused across boots, so "fresh transient GlobalId" is misleading.
4. The tests cover the PendingRegistration state machine only. A test that the registration survives the input closing would have caught 1, and a multi-worker reconcile in clusterd-test-driver with a shared label could cover the race.

Posted by Claude Code.

The fallible try_register_collector_with_dropper split is clean, and replacing the old "never collides" NOTE is right.

Should fix

1. Retries aren't rate-limited. try_register runs on every activation of the active worker, and each failure schedules another activate_after(1s). With a busy input during the collision window, that means a registration attempt (collector clone plus a registry write lock) per activation, a pile-up of delayed activations, and a mz_compute_metric_sink_registration_retries_total that counts activations rather than retries, so its help text is misleading. Suggest a next_attempt: Instant on PendingRegistration: return early before it, and only schedule a retry when none is pending.
2. PR description is stale. It says "The guard lives in the operator closure", but the code deliberately returns it as the sink token, as the comment at the Rc<dyn Any> return explains.
3. Retries stop if the inputs close first. Timely drops the operator closure once both input frontiers close, which ends the retry loop. The token then holds a PendingRegistration with handle: None forever, and the sink never publishes. That's probably rare for curated sinks, but it contradicts "retried until the predecessor drops". Either note it as a known gap or keep an activation pending until the handle is set.

Nits

4. "Once per streak" doesn't apply. first_collision and escalated are never reset, and there are no more attempts after success, so there is only ever one streak.
5. New unchecked claim in the NOTE. "User sinks … both land on one worker that drops before it re-creates, and never collide" is the same kind of claim the old NOTE made, and that one turned out wrong. Point at what enforces it (the ordering in Worker::reconcile) or cover it with a test.
6. registration_survives_operator_closure_drop only tests Rc refcounting. It doesn't build an operator, so it wouldn't catch the operator no longer sharing the token. A dataflow-level test would, for example a constant-input sink whose series survives its input closing.
7. Repeated struct literal. The seven-field PendingRegistration { … } literal appears three times. A PendingRegistration::new(label, collector) constructor would remove that.

Generated by Claude Code