From 55862f1ab16faa1025b564b234e8a4e43e4d988d Mon Sep 17 00:00:00 2001 From: pingqiu Date: Thu, 9 Apr 2026 16:18:44 -0700 Subject: [PATCH] fix: rebuild base-only completion + protocol handshake + direct ack events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fixes for the remote rebuild path: 1. Base-only completion: when BaseLSN == TargetLSN, the base image covers all data — no WAL tail needed. MarkBaseComplete now auto-satisfies the WAL condition and calls TryComplete so the session completes immediately after the base transfer finishes. 2. Base lane protocol handshake: runBaseLaneClient now sends MsgRebuildReq {Type: RebuildSessionBase} before reading. The RebuildServer requires this handshake to dispatch to ServeBaseBlocks. Without it, the server received raw frames it couldn't understand. 3. Direct ack events: OnAck emits engine events directly (SessionCompleted, SessionProgressObserved, SessionFailed) instead of routing through ObserveReplicaRebuildSessionAck which requires the sender in the orchestrator registry. The remote coordinator owns the session — no registry lookup needed. Also adds diagnostic logging on both sides: - Replica: logs parsed RebuildAddr and base lane client start - Primary: logs sender state after installSession Co-Authored-By: Claude Opus 4.6 (1M context) --- weed/server/block_recovery.go | 50 ++++++++++++++++++++---- weed/storage/blockvol/rebuild_session.go | 9 +++++ weed/storage/blockvol/replica_barrier.go | 16 ++++++++ 3 files changed, 67 insertions(+), 8 deletions(-) diff --git a/weed/server/block_recovery.go b/weed/server/block_recovery.go index 7a98d780f..79e14d069 100644 --- a/weed/server/block_recovery.go +++ b/weed/server/block_recovery.go @@ -862,22 +862,56 @@ func (rm *RecoveryManager) buildRemoteRebuildIO(replicaID, volPath, rebuildAddr Epoch: epoch, SessionID: sessionID, OnAck: func(ack blockvol.SessionAckMsg) error { - err := bs.ObserveReplicaRebuildSessionAck(volPath, replicaID, ack) - // On completion, store the replica's achieved LSN so - // readRebuildStatus uses the replica's proof, not the primary's vol. - if err == nil && ack.Phase == blockvol.SessionAckCompleted { - achieved := ack.AchievedLSN - if achieved == 0 { - achieved = ack.WALAppliedLSN + // Emit engine events directly. The remote coordinator owns the + // session — no sender registry lookup needed. This avoids the + // "sender not found" failure when the registry is reconciled + // between session install and ack arrival. + if bs == nil || bs.v2Core == nil { + return nil + } + achieved := ack.AchievedLSN + if achieved == 0 { + achieved = ack.WALAppliedLSN + } + switch ack.Phase { + case blockvol.SessionAckAccepted: + // No engine event needed — shipper transition handled by caller. + case blockvol.SessionAckRunning, blockvol.SessionAckBaseComplete: + if achieved > 0 { + bs.applyCoreEvent(engine.SessionProgressObserved{ + ID: volPath, + ReplicaID: replicaID, + Kind: engine.SessionRebuild, + AchievedLSN: achieved, + }) } + case blockvol.SessionAckCompleted: + // Store achieved LSN for OnRebuildCompleted (skip local vol read). rm.mu.Lock() if rm.remoteRebuildAchieved == nil { rm.remoteRebuildAchieved = make(map[string]uint64) } rm.remoteRebuildAchieved[replicaID] = achieved rm.mu.Unlock() + bs.applyCoreEvent(engine.SessionCompleted{ + ID: volPath, + ReplicaID: replicaID, + Kind: engine.SessionRebuild, + AchievedLSN: achieved, + }) + case blockvol.SessionAckFailed: + reason := "rebuild_failed" + if ack.BaseComplete { + reason = "rebuild_failed_post_base" + } + bs.applyCoreEvent(engine.SessionFailed{ + ID: volPath, + ReplicaID: replicaID, + Kind: engine.SessionRebuild, + Reason: reason, + }) } - return err + return nil }, TransitionShipper: func(state blockvol.ReplicaState) { if shipperRef != nil { diff --git a/weed/storage/blockvol/rebuild_session.go b/weed/storage/blockvol/rebuild_session.go index 402aa2d1b..eb5ca0719 100644 --- a/weed/storage/blockvol/rebuild_session.go +++ b/weed/storage/blockvol/rebuild_session.go @@ -217,9 +217,18 @@ func (s *RebuildSession) MarkBaseComplete(totalBlocks uint64) { if s.phase == RebuildPhaseRunning { s.phase = RebuildPhaseBaseComplete } + // When BaseLSN == TargetLSN, the base image covers all data — no WAL + // tail needed. Auto-satisfy the WAL condition so TryComplete succeeds + // immediately after base transfer. + if s.config.BaseLSN == s.config.TargetLSN && s.walAppliedLSN < s.config.TargetLSN { + s.walAppliedLSN = s.config.TargetLSN + } ack := s.sessionAckLocked() s.mu.Unlock() s.vol.emitRebuildSessionAck(ack) + // Try to complete immediately — covers the BaseLSN == TargetLSN case + // where no WAL entries will arrive. + s.TryComplete() } // TryComplete checks if both completion conditions are met: diff --git a/weed/storage/blockvol/replica_barrier.go b/weed/storage/blockvol/replica_barrier.go index 420fe3149..b9f982628 100644 --- a/weed/storage/blockvol/replica_barrier.go +++ b/weed/storage/blockvol/replica_barrier.go @@ -67,6 +67,8 @@ func (r *ReplicaReceiver) handleSessionControl(conn net.Conn, payload []byte, wr } switch ctrl.Command { case SessionCmdStartRebuild: + log.Printf("replica: handleSessionControl start_rebuild session=%d epoch=%d base=%d target=%d rebuildAddr=%q", + ctrl.SessionID, ctrl.Epoch, ctrl.BaseLSN, ctrl.TargetLSN, ctrl.RebuildAddr) r.vol.SetOnRebuildSessionAck(func(ack SessionAckMsg) { if ack.SessionID != ctrl.SessionID { return @@ -111,12 +113,26 @@ func (r *ReplicaReceiver) handleSessionControl(conn net.Conn, payload []byte, wr // session control message. On failure, the rebuild session will eventually time // out or be cancelled by the primary. func (r *ReplicaReceiver) runBaseLaneClient(sessionID uint64, rebuildAddr string) { + log.Printf("replica: base lane client starting session=%d addr=%s", sessionID, rebuildAddr) conn, err := net.DialTimeout("tcp", rebuildAddr, 5*time.Second) if err != nil { log.Printf("replica: base lane dial %s: %v", rebuildAddr, err) return } defer conn.Close() + + // Send rebuild request so the RebuildServer dispatches to ServeBaseBlocks. + epoch := r.vol.Epoch() + req := RebuildRequest{ + Type: RebuildSessionBase, + Epoch: epoch, + FromLSN: 0, // full base + } + if err := WriteFrame(conn, MsgRebuildReq, EncodeRebuildRequest(req)); err != nil { + log.Printf("replica: base lane send request to %s: %v", rebuildAddr, err) + return + } + client := NewRebuildTransportClient(r.vol, sessionID) blocks, err := client.ReceiveBaseBlocks(conn) if err != nil {