From 80be2ec05a54910130aec15bff7386bd7d4b89a8 Mon Sep 17 00:00:00 2001 From: pingqiu Date: Thu, 2 Apr 2026 22:39:08 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20CP13-5=20=E2=80=94=20reconnect=20handsh?= =?UTF-8?q?ake=20+=20WAL=20catch-up=20on=20SetReplicaAddrs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug: SetReplicaAddrs created fresh shippers (hasFlushedProgress=false), so after disconnect, the new shipper used bootstrap instead of reconnect handshake. Bootstrap doesn't replay missed WAL entries — barrier hung. Fix: - blockvol.go: SetReplicaAddrs checks if old shipper group had durable progress (AnyHasFlushedProgress). If so, seeds new shippers with hasFlushedProgress=true → they use reconnect handshake + catch-up. - shipper_group.go: add AnyHasFlushedProgress() helper. 3 baseline FAILs now PASS: - ReconnectUsesHandshakeNotBootstrap: reconnect path used, not bootstrap - CatchupMultipleDisconnects: repeated disconnect/reconnect recovers - CatchupDoesNotOverwriteNewerData: catch-up completes, safety exercised 7 tests promoted to CP13-5 primary proof. TestAdversarial_NeedsRebuildBlocksAllPaths still FAIL (CP13-7 scope). Co-Authored-By: Claude Opus 4.6 (1M context) --- .../.private/phase/phase-13-cp5-reconnect.md | 89 +++++++++++++++++++ weed/storage/blockvol/blockvol.go | 18 ++++ weed/storage/blockvol/shipper_group.go | 13 +++ 3 files changed, 120 insertions(+) create mode 100644 sw-block/.private/phase/phase-13-cp5-reconnect.md diff --git a/sw-block/.private/phase/phase-13-cp5-reconnect.md b/sw-block/.private/phase/phase-13-cp5-reconnect.md new file mode 100644 index 000000000..0e3bff472 --- /dev/null +++ b/sw-block/.private/phase/phase-13-cp5-reconnect.md @@ -0,0 +1,89 @@ +# CP13-5 Reconnect Handshake + WAL Catch-up — Contract Review + Proof Package + +Date: 2026-04-03 +Code change: `blockvol.go` SetReplicaAddrs + `shipper_group.go` AnyHasFlushedProgress + +## Reconnect Decision Matrix + +| Prior durable progress? | WAL covers gap? | Outcome | +|------------------------|----------------|---------| +| No (`hasFlushedProgress=false`) | N/A | Bootstrap: bare Ship + Barrier | +| Yes | Yes (gap within retained WAL) | Reconnect: ResumeShipReq handshake → catch-up replay → InSync | +| Yes | No (gap exceeds retained WAL) | Fail closed: NeedsRebuild (CP13-7 scope for full lifecycle) | + +## What Changed + +**Bug:** `SetReplicaAddrs` created fresh shippers with `hasFlushedProgress=false`, so after +disconnect + reconnect, the shipper used the bootstrap path instead of the reconnect handshake. +Bootstrap doesn't replay missed WAL entries, so the barrier waited forever for entries the +replica never received. + +**Fix (`blockvol.go`):** `SetReplicaAddrs` now checks if the old shipper group had any +shipper with durable progress (`AnyHasFlushedProgress`). If so, new shippers are seeded +with `hasFlushedProgress=true`, routing them through the reconnect handshake + catch-up path. + +**New helper (`shipper_group.go`):** `AnyHasFlushedProgress()` — returns true if any shipper +in the group has ever received a valid `FlushedLSN > 0` from a barrier response. + +## Reconnect Path (production flow) + +``` +SetReplicaAddrs(new addresses after reconnect) + ├─ old group had flushedProgress? → seed new shippers with hasFlushedProgress=true + └─ new shipper created with WAL access + +SyncCache → groupCommit.Submit → Barrier(lsnMax) + ├─ state=Disconnected + hasFlushedProgress=true + wal != nil + │ → doReconnectAndCatchUp() + │ → reconnectWithHandshake() + │ → TCP connect to new replica address + │ → ResumeShipReq{Epoch, PrimaryHeadLSN, RetainStart} + │ → replica responds with {Status, ReplicaFlushedLSN} + │ → gap analysis: R (replica flushed) vs H (primary head) vs S (retain start) + │ ├─ R >= H: already caught up → InSync + │ ├─ R >= S: recoverable gap → CatchingUp → runCatchUp(R) + │ └─ R < S: gap exceeds retention → NeedsRebuild + │ → runCatchUp: stream WAL entries from R to H → replica applies + │ → catch-up complete → InSync + └─ barrier request proceeds (InSync) +``` + +## Baseline FAILs Now Closed + +| Test | Was | Now | Why | +|------|-----|-----|-----| +| `TestAdversarial_ReconnectUsesHandshakeNotBootstrap` | FAIL | PASS | Seeded hasFlushedProgress → reconnect path used | +| `TestAdversarial_CatchupMultipleDisconnects` | FAIL | PASS | Repeated SetReplicaAddrs preserves progress seed | +| `TestAdversarial_CatchupDoesNotOverwriteNewerData` | FAIL | PASS | Catch-up now completes, safety invariant exercised | + +## Baseline Tests Promoted to CP13-5 Proof + +### Primary proofs + +| Test | What it proves | +|------|---------------| +| `TestAdversarial_ReconnectUsesHandshakeNotBootstrap` | Degraded shipper with prior progress reconnects via handshake + catch-up | +| `TestAdversarial_CatchupMultipleDisconnects` | Repeated disconnect/reconnect cycles recover cleanly | +| `TestAdversarial_CatchupDoesNotOverwriteNewerData` | Catch-up replays missing entries without overwriting newer replica data | +| `TestReconnect_CatchupFromRetainedWal` | Retained-WAL gap replays and returns to InSync | +| `TestReconnect_EpochChangeDuringCatchup_Aborts` | Epoch change during catch-up aborts cleanly | +| `TestReconnect_CatchupTimeout_TransitionsDegraded` | Catch-up timeout → Degraded (bounded failure) | +| `TestAdversarial_FreshShipperUsesBootstrapNotReconnect` | Fresh shipper (no prior progress) uses bootstrap, not reconnect | + +### Support evidence + +| Test | What it supports | +|------|-----------------| +| `TestReconnect_GapBeyondRetainedWal_NeedsRebuild` | PASS* — asserts barrier failure on large gap, but full NeedsRebuild lifecycle is CP13-7 | + +### Still FAIL (CP13-7 scope) + +| Test | Why still fails | +|------|----------------| +| `TestAdversarial_NeedsRebuildBlocksAllPaths` | Full NeedsRebuild lifecycle — lease expiry + WAL overflow timing; CP13-7 scope | + +## What CP13-5 Does NOT Close + +- Replica-aware WAL retention policy (CP13-6) +- Full NeedsRebuild lifecycle / rebuild execution (CP13-7) +- The `TestAdversarial_NeedsRebuildBlocksAllPaths` failure is a CP13-7 gap, not CP13-5 diff --git a/weed/storage/blockvol/blockvol.go b/weed/storage/blockvol/blockvol.go index 4ac27a0a5..667d2bd66 100644 --- a/weed/storage/blockvol/blockvol.go +++ b/weed/storage/blockvol/blockvol.go @@ -848,13 +848,31 @@ func (v *BlockVol) SetReplicaAddr(dataAddr, ctrlAddr string) { // SetReplicaAddrs configures N replica endpoints and creates a ShipperGroup // with distributed group commit. Creates fresh shippers (old group is GC'd). +// +// CP13-5: If the old shipper group had durable progress (any shipper was +// previously InSync), new shippers are seeded with hasFlushedProgress=true +// so they use the reconnect handshake + catch-up path instead of bare +// bootstrap. This is correct because the primary's WAL may contain entries +// the replica hasn't received yet (written during disconnect), and the +// handshake determines the exact gap to replay. func (v *BlockVol) SetReplicaAddrs(addrs []ReplicaAddr) { + // CP13-5: Check if old group had durable progress before replacing. + hadPriorProgress := false + if v.shipperGroup != nil { + hadPriorProgress = v.shipperGroup.AnyHasFlushedProgress() + } + wa := &walAccess{vol: v} shippers := make([]*WALShipper, len(addrs)) for i, a := range addrs { shippers[i] = NewWALShipper(a.DataAddr, a.CtrlAddr, func() uint64 { return v.epoch.Load() }, wa, v.Metrics) + // CP13-5: Seed new shippers with prior progress so reconnect + // path is used instead of bootstrap. + if hadPriorProgress { + shippers[i].hasFlushedProgress.Store(true) + } } v.shipperGroup = NewShipperGroup(shippers) diff --git a/weed/storage/blockvol/shipper_group.go b/weed/storage/blockvol/shipper_group.go index aede8f6f9..0ab34f6f0 100644 --- a/weed/storage/blockvol/shipper_group.go +++ b/weed/storage/blockvol/shipper_group.go @@ -72,6 +72,19 @@ func (sg *ShipperGroup) AllDegraded() bool { return true } +// AnyHasFlushedProgress returns true if any shipper has ever had durable +// progress (CP13-5: used to seed replacement shippers on SetReplicaAddrs). +func (sg *ShipperGroup) AnyHasFlushedProgress() bool { + sg.mu.RLock() + defer sg.mu.RUnlock() + for _, s := range sg.shippers { + if s.HasFlushedProgress() { + return true + } + } + return false +} + // AnyDegraded returns true if at least one shipper is degraded. func (sg *ShipperGroup) AnyDegraded() bool { sg.mu.RLock()