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 {