GeezerVison

The PR's own GC can't delete any file of the newest checkpoint; I traced it end to end. But it shouldn't merge as-is: combined with the cloud S3 pull, it can roll a pipeline back to an older checkpoint on an ordinary restart, and I reproduced that. It also leaves a pre-existing multihost hole open, while the PR description says keeping two checkpoints prevents it.

1. Blocker: start_from_checkpoint: "<uuid>" rolls the pipeline back (reproduced)

1. A pipeline starts from checkpoint X, pulled from the bucket, and writes checkpoint Y.
2. RunningCheckpoint::finish calls Checkpointer::gc_keeping(_, 1), which deletes X locally.
3. On the next start (a crash, stop/start, or suspend/resume), the server calls ControllerBuilder::pull_once, which reaches the cloud's pull::pull_checkpoint.
4. pull::check_local_storage(Uuid(X)) doesn't find X, so the pull downloads X again. pull::write_local_catalog replaces the catalog with [X].
5. PipelineStorage::gc_startup then deletes Y and its batch files, and open_latest_checkpoint resumes from X.

Y is lost without any error. X stays in the bucket for at least 30 days and until 10 newer checkpoints are pushed (the default retention), so every restart in that period rolls back again.

I ran a probe: a real dbsp circuit, a local rclone serve s3 as the bucket, the cloud's real push_checkpoint and pull_checkpoint, and this PR's dbsp. "Keys lost" counts keys that were absent from the restored state.

┌────────────────────────┬─────────────────────┬───────────────────────┬────────────┬────────────┐
│        GC rule         │ Checkpoints after X │ Catalog after restart │ Y survived │ Keys lost  │
├────────────────────────┼─────────────────────┼───────────────────────┼────────────┼────────────┤
│ This PR (keep 1)       │ 1                   │ [X]                   │ no         │ 100 of 100 │
├────────────────────────┼─────────────────────┼───────────────────────┼────────────┼────────────┤
│ Before the PR (keep 2) │ 1                   │ [X, Y]                │ yes        │ 0          │
├────────────────────────┼─────────────────────┼───────────────────────┼────────────┼────────────┤
│ Before the PR (keep 2) │ 2                   │ [X]                   │ no         │ 200 of 200 │
└────────────────────────┴─────────────────────┴───────────────────────┴────────────┴────────────┘

- The root cause predates this PR. It is in the cloud pull, and the old rule fails the same way one checkpoint later.
- What the PR changes: a pipeline that checkpoints only at suspend used to survive its first suspend/resume, and now doesn't. With periodic checkpoints, the safe window shrinks by one interval.
- Fix, before or with this PR: make the pull treat a UUID as a one-time seed. Record the seeded UUID in a marker file, like standby's activation marker. On later starts, use the local catalog if it isn't empty.
- No existing test catches this. test_from_uuid restarts only once, after clearing storage.

2. Multihost: two failed checkpoints in a row lose all state (existed before the PR; inferred from code, not reproduced)

- Hosts trim without coordination: after every local success, a host trims to its 2 newest checkpoints, whatever happened on the other hosts.
- Failures don't stop checkpointing: the coordinator's main loop ignores a failed periodic checkpoint (let _ = self.checkpoint(...)).
- No common checkpoint means a fresh start: Coordinator::choose_checkpoint then returns None, and the pipeline starts fresh with no error.

┌────────────────────────┬───────────────────┬────────────────┬──────────────────────────┐
│ Coordinated checkpoint │ Host A (succeeds) │ Host B (fails) │ Checkpoint on every host │
├────────────────────────┼───────────────────┼────────────────┼──────────────────────────┤
│ after N and complete   │ N                 │ N              │ N                        │
├────────────────────────┼───────────────────┼────────────────┼──────────────────────────┤
│ N+1                    │ N, N+1            │ N              │ N                        │
├────────────────────────┼───────────────────┼────────────────┼──────────────────────────┤
│ N+2                    │ N+1, N+2          │ N              │ none                     │
└────────────────────────┴───────────────────┴────────────────┴──────────────────────────┘

Failures on two different hosts give the same result. I traced both rules through these sequences, and this PR doesn't make it worse. With the new complete message, the fix is cheap: a host keeps everything from the last checkpoint the coordinator confirmed, or from the one it was activated from. It falls back to keeping 2 only if it has never received a confirmation, as with an old coordinator.

3. Low: a sync can name a checkpoint that GC has just deleted (inferred, not reproduced)

- The window: the /checkpoint/sync handler reads the catalog from disk. The UUID is protected only once CircuitThread::run_commands registers it, so a checkpoint finishing in between can delete it.
- The result: the push fails ("failed to read checkpoint directory") and leaves an empty <uuid>.zip in the storage root that is never cleaned up. No data is lost.
- Before the PR, this took two checkpoints finishing in that window; now one is enough. The coordinator's periodic multihost push races with complete in the same way.