diff --git a/weed/server/block_rebuild_remote.go b/weed/server/block_rebuild_remote.go new file mode 100644 index 000000000..db5e80541 --- /dev/null +++ b/weed/server/block_rebuild_remote.go @@ -0,0 +1,177 @@ +package weed_server + +import ( + "errors" + "fmt" + "net" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +// errRebuildAckFailed is returned by TransferFullBase when the rebuild failed +// via a replica SessionAckFailed. The ack observation path (ObserveReplicaRebuildSessionAck) +// already emitted engine.SessionFailed, so callers must NOT emit a second one. +var errRebuildAckFailed = errors.New("rebuild failed via session ack") + +// RemoteRebuildIO implements engine.RebuildIO for the primary-side remote +// rebuild path. Instead of installing base blocks locally (which is what the +// v2bridge.Executor does on the replica), this implementation coordinates +// remotely: it sends a session control message to the replica and monitors +// acks until the replica reports completion or failure. +// +// The base blocks are served by the primary's existing RebuildServer. +// The replica auto-starts a base lane client on receiving the session control. +// WAL entries continue flowing through the shipper's data channel. +// +// Ack forwarding: each ack from the replica is forwarded to onAck, which +// routes through ObserveReplicaRebuildSessionAck for pin/watchdog/engine +// integration. +type RemoteRebuildIO struct { + // ReplicaCtrlAddr is the replica's control channel address. + ReplicaCtrlAddr string + // RebuildAddr is the primary's rebuild server address (sent to replica + // so it knows where to connect for the base lane). + RebuildAddr string + // BaseLSN is the flushed/checkpoint boundary the extent can serve. + // Must be derived from the real checkpoint, not the plan target. + BaseLSN uint64 + // Epoch is the current volume epoch for session validation. + Epoch uint64 + // SessionID uniquely identifies this rebuild session. + SessionID uint64 + // OnAck forwards each replica ack to the observation layer + // (ObserveReplicaRebuildSessionAck). Returns error if the ack is + // rejected (stale session, etc.). Shipper state transitions happen + // only after successful observation. + OnAck func(blockvol.SessionAckMsg) error + // TransitionShipper changes the shipper state. Called only after + // OnAck succeeds. Used for: NeedsRebuild → Rebuilding (on accepted), + // Rebuilding → InSync (on completed), Rebuilding → NeedsRebuild (on failure). + TransitionShipper func(blockvol.ReplicaState) +} + +// TransferFullBase sends a session control to the replica and blocks until +// the rebuild completes or fails. committedLSN is the engine's frozen target +// from PlanRebuild — the replica must reach at least this LSN for completion. +// +// Protocol: +// 1. Dial replica ctrl addr (fresh connection, separate from barrier) +// 2. Send atomic SessionControlV2 with RebuildAddr trailer +// 3. Read acks in loop; forward each to OnAck for observation +// 4. On SessionAckAccepted: transition shipper to Rebuilding (live WAL lane opens) +// 5. On SessionAckCompleted: transition shipper to InSync, return achievedLSN +// 6. On SessionAckFailed or error: transition shipper to NeedsRebuild, return error +func (r *RemoteRebuildIO) TransferFullBase(committedLSN uint64) (uint64, error) { + conn, err := net.DialTimeout("tcp", r.ReplicaCtrlAddr, 5*time.Second) + if err != nil { + return 0, fmt.Errorf("remote rebuild: dial ctrl %s: %w", r.ReplicaCtrlAddr, err) + } + defer conn.Close() + + // Set a generous deadline for the entire rebuild session. + conn.SetDeadline(time.Now().Add(10 * time.Minute)) + + // Send atomic v2 session control with RebuildAddr trailer. + msg := blockvol.SessionControlMsg{ + Epoch: r.Epoch, + SessionID: r.SessionID, + Command: blockvol.SessionCmdStartRebuild, + BaseLSN: r.BaseLSN, + TargetLSN: committedLSN, // engine's frozen rebuild target + RebuildAddr: r.RebuildAddr, + } + if err := blockvol.SendSessionControlV2(conn, msg); err != nil { + return 0, fmt.Errorf("remote rebuild: send session control: %w", err) + } + + glog.V(0).Infof("remote rebuild: sent start_rebuild session=%d base=%d target=%d rebuild=%s → %s", + r.SessionID, r.BaseLSN, committedLSN, r.RebuildAddr, r.ReplicaCtrlAddr) + + // Read acks until terminal phase. + for { + msgType, payload, err := blockvol.ReadFrame(conn) + if err != nil { + r.transitionOnFailure() + return 0, fmt.Errorf("remote rebuild: read ack: %w", err) + } + if msgType != blockvol.MsgSessionAck { + continue + } + ack, err := blockvol.DecodeSessionAck(payload) + if err != nil { + r.transitionOnFailure() + return 0, fmt.Errorf("remote rebuild: decode ack: %w", err) + } + + // Forward to observation layer. Only transition shipper state + // if observation succeeds (Rule 1: ack-gated transitions). + ackErr := r.forwardAck(ack) + + switch ack.Phase { + case blockvol.SessionAckAccepted: + if ackErr != nil { + r.transitionOnFailure() + return 0, fmt.Errorf("remote rebuild: accepted ack rejected by observation: %w", ackErr) + } + if r.TransitionShipper != nil { + r.TransitionShipper(blockvol.ReplicaRebuilding) + } + glog.V(0).Infof("remote rebuild: session %d accepted by replica", r.SessionID) + + case blockvol.SessionAckRunning, blockvol.SessionAckBaseComplete: + // Fail closed: if observation rejects a progress ack, the primary + // no longer considers this session valid (stale ID, wrong kind, etc.). + if ackErr != nil { + r.transitionOnFailure() + return 0, fmt.Errorf("remote rebuild: progress ack rejected by observation: %w", ackErr) + } + + case blockvol.SessionAckCompleted: + if ackErr != nil { + r.transitionOnFailure() + return 0, fmt.Errorf("remote rebuild: completed ack rejected by observation: %w", ackErr) + } + if r.TransitionShipper != nil { + r.TransitionShipper(blockvol.ReplicaInSync) + } + achieved := ack.AchievedLSN + if achieved == 0 { + achieved = ack.WALAppliedLSN + } + glog.V(0).Infof("remote rebuild: session %d completed (achieved=%d)", r.SessionID, achieved) + return achieved, nil + + case blockvol.SessionAckFailed: + // Forward the ack for observation cleanup (pins, watchdog, engine + // SessionFailed). Then return the sentinel error so ExecutePendingRebuild + // knows NOT to emit a second SessionFailed. + r.transitionOnFailure() + return 0, fmt.Errorf("remote rebuild: session %d: %w", r.SessionID, errRebuildAckFailed) + } + } +} + +// TransferSnapshot is not supported in the V1 remote rebuild path. +func (r *RemoteRebuildIO) TransferSnapshot(snapshotLSN uint64) error { + return fmt.Errorf("remote rebuild: TransferSnapshot not supported (v1 full-base only)") +} + +// StreamWALEntries is not supported — WAL flows through the shipper's data channel. +func (r *RemoteRebuildIO) StreamWALEntries(startExclusive, endInclusive uint64) (uint64, error) { + return 0, fmt.Errorf("remote rebuild: StreamWALEntries not supported (WAL flows through shipper)") +} + +func (r *RemoteRebuildIO) forwardAck(ack blockvol.SessionAckMsg) error { + if r.OnAck == nil { + return nil + } + return r.OnAck(ack) +} + +func (r *RemoteRebuildIO) transitionOnFailure() { + if r.TransitionShipper != nil { + r.TransitionShipper(blockvol.ReplicaNeedsRebuild) + } +} diff --git a/weed/server/block_recovery.go b/weed/server/block_recovery.go index 2e5cf8d3d..349342200 100644 --- a/weed/server/block_recovery.go +++ b/weed/server/block_recovery.go @@ -2,6 +2,7 @@ package weed_server import ( "context" + "errors" "fmt" "net" "strconv" @@ -35,10 +36,11 @@ type recoveryTask struct { type RecoveryManager struct { bs *BlockService - mu sync.Mutex - tasks map[string]*recoveryTask - coord *rt.PendingCoordinator - wg sync.WaitGroup + mu sync.Mutex + tasks map[string]*recoveryTask + remoteRebuildAchieved map[string]uint64 // replicaID → achievedLSN from remote rebuild + coord *rt.PendingCoordinator + wg sync.WaitGroup // TestHook: if set, called before execution starts. Tests use this // to hold the goroutine alive for serialized-replacement proofs. @@ -200,6 +202,20 @@ func (rm *RecoveryManager) StartRecoveryTask(replicaID string, assignments []blo rm.startTask(replicaID, assignments) } +// StartRebuildFromProbe is the primary-direct rebuild entry point. +// Called when ProbeReconnect determines a replica needs a full rebuild. +// It installs a rebuild session on the orchestrator, then starts the +// recovery task. No fake assignment needed — deriveRebuildAddr computes +// the primary's rebuild server address from ReplicationPorts. +func (rm *RecoveryManager) StartRebuildFromProbe(replicaID string) { + if err := rm.installSession(replicaID, engine.SessionRebuild); err != nil { + glog.Warningf("recovery: install rebuild session for probe %s: %v", replicaID, err) + return + } + rm.cancelAndDrain(replicaID, false) + rm.startTask(replicaID, nil) +} + // DrainRecoveryTask drains removed recovery work from an explicit core-owned // command seam on the core-present path. func (rm *RecoveryManager) DrainRecoveryTask(replicaID, reason string) { @@ -362,8 +378,8 @@ type recoveryContext struct { volPath string rebuildAddr string driver *engine.RecoveryDriver - executor *v2bridge.Executor - replicaFlushedLSN uint64 // catch-up start point (0 if no session) + executor *v2bridge.Executor // catch-up IO (reads primary WAL, ships to replica) + replicaFlushedLSN uint64 // catch-up start point (0 if no session) } // resolveRecoveryContext resolves everything needed for recovery execution: @@ -542,13 +558,29 @@ func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID string, ass rm.executeLegacyRebuild(ctx, rctx.volPath, replicaID, rctx.driver, plan, rctx.executor) return } + + // Single rebuild route: always use RemoteRebuildIO on the core-present path. + // Primary coordinates, replica installs. Tests can override pe.RebuildIO + // via OnPendingExecution hook after the pending is stored. + // + // Rule 3: RemoteRebuildIO is full-base-only in V1. If the plan requests + // snapshot/tail-replay, RemoteRebuildIO.TransferSnapshot will return an error + // at execution time. Tests that need snapshot plans inject fakeRebuildIO + // via OnPendingExecution. + remote := rm.buildRemoteRebuildIO(replicaID, rctx.volPath, rctx.rebuildAddr) + if remote == nil { + glog.Warningf("recovery: cannot build remote rebuild IO for %s — no reachable replica", replicaID) + return + } + var rebuildIO engine.RebuildIO = remote + pe := &rt.PendingExecution{ VolumeID: rctx.volPath, ReplicaID: replicaID, RebuildTargetLSN: plan.RebuildTargetLSN, Driver: rctx.driver, Plan: plan, - RebuildIO: rctx.executor, + RebuildIO: rebuildIO, } rm.coord.Store(replicaID, pe) if rm.OnPendingExecution != nil { @@ -575,7 +607,22 @@ func (rm *RecoveryManager) ExecutePendingRebuild(replicaID string, targetLSN uin if pe == nil || pe.Driver == nil || pe.Plan == nil { return nil } - return rt.ExecuteRebuildPlan(pe.Driver, pe.Plan, pe.RebuildIO, pe.VolumeID, pe.ReplicaID, rm) + err := rt.ExecuteRebuildPlan(pe.Driver, pe.Plan, pe.RebuildIO, pe.VolumeID, pe.ReplicaID, rm) + if err != nil { + glog.Warningf("recovery: rebuild execution failed for %s: %v", replicaID, err) + // Emit SessionFailed only for transport errors (dial/EOF/decode). + // Ack-driven failures (errRebuildAckFailed) already emitted SessionFailed + // through ObserveReplicaRebuildSessionAck — don't double-emit. + if !errors.Is(err, errRebuildAckFailed) && rm.bs != nil && rm.bs.v2Core != nil { + rm.bs.applyCoreEvent(engine.SessionFailed{ + ID: pe.VolumeID, + ReplicaID: replicaID, + Kind: engine.SessionRebuild, + Reason: err.Error(), + }) + } + } + return err } // RecoveryCallbacks implementation — host-side completion notifications. @@ -617,7 +664,23 @@ func (rm *RecoveryManager) OnRebuildCompleted(volumeID, replicaID string, plan * if rm.bs == nil || rm.bs.v2Core == nil { return } - status := rm.readRebuildStatus(volumeID) + // For remote rebuilds, use the replica's achieved LSN (stored by onAck + // callback on SessionAckCompleted) instead of reading the primary's local + // vol — the primary's vol is the source, not the rebuilt destination. + rm.mu.Lock() + remoteAchieved, isRemote := rm.remoteRebuildAchieved[replicaID] + delete(rm.remoteRebuildAchieved, replicaID) // consumed + rm.mu.Unlock() + + var status rt.RebuildCompletionStatus + if isRemote { + // Use replica's completion proof. CommittedLSN = CheckpointLSN = achievedLSN. + status.CommittedLSN = remoteAchieved + status.CheckpointLSN = remoteAchieved + } else { + // Legacy/local path: read from primary vol (backwards compat). + status = rm.readRebuildStatus(volumeID) + } ev := rt.DeriveRebuildCommitted(volumeID, replicaID, status, plan) rm.bs.applyCoreEvent(ev) } @@ -708,6 +771,100 @@ func (rm *RecoveryManager) deriveRebuildAddr(replicaID string, assignments []blo return net.JoinHostPort(host, strconv.Itoa(rebuildPort)) } +// buildRemoteRebuildIO creates a RemoteRebuildIO for the primary-side remote +// rebuild path. It resolves the replica's ctrl address from the shipper, +// derives BaseLSN from the actual checkpoint, and wires the onAck callback +// through ObserveReplicaRebuildSessionAck for pin/watchdog/engine integration. +func (rm *RecoveryManager) buildRemoteRebuildIO(replicaID, volPath, rebuildAddr string) *RemoteRebuildIO { + if rm == nil || rm.bs == nil { + return nil + } + // Resolve replica ctrl address from the shipper. + var ctrlAddr string + var shipperRef *blockvol.WALShipper + if err := rm.bs.blockStore.WithVolume(volPath, func(vol *blockvol.BlockVol) error { + sg := vol.GetShipperGroup() + if sg == nil { + return nil + } + for i := 0; i < sg.Len(); i++ { + s := sg.Shipper(i) + if s != nil && s.ReplicaID() == replicaID { + ctrlAddr = s.CtrlAddr() + shipperRef = s + break + } + // Also match by suffix (engine replicaID = "path/serverID", shipper = "serverID") + if s != nil && len(replicaID) > len(volPath)+1 { + shipperServerID := replicaID[len(volPath)+1:] + if s.ReplicaID() == shipperServerID { + ctrlAddr = s.CtrlAddr() + shipperRef = s + break + } + } + } + return nil + }); err != nil || ctrlAddr == "" { + glog.Warningf("recovery: cannot resolve ctrl addr for %s in %s", replicaID, volPath) + return nil + } + + // Derive BaseLSN from the real flushed checkpoint boundary. + var baseLSN uint64 + _ = rm.bs.blockStore.WithVolume(volPath, func(vol *blockvol.BlockVol) error { + baseLSN = vol.CheckpointLSN() + return nil + }) + + // Resolve epoch and session ID from the orchestrator sender. + // The session ID must match the active orchestrator session so ack + // observation (ObserveReplicaRebuildSessionAck) accepts the acks. + var epoch, sessionID uint64 + if s := rm.bs.v2Orchestrator.Registry.Sender(replicaID); s != nil { + epoch = s.Epoch() + if snap := s.SessionSnapshot(); snap != nil && snap.Active { + sessionID = snap.ID + } + } + if sessionID == 0 { + glog.Warningf("recovery: no active session for %s — cannot build remote rebuild IO", replicaID) + return nil + } + + bs := rm.bs + return &RemoteRebuildIO{ + ReplicaCtrlAddr: ctrlAddr, + RebuildAddr: rebuildAddr, + BaseLSN: baseLSN, + 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 + } + rm.mu.Lock() + if rm.remoteRebuildAchieved == nil { + rm.remoteRebuildAchieved = make(map[string]uint64) + } + rm.remoteRebuildAchieved[replicaID] = achieved + rm.mu.Unlock() + } + return err + }, + TransitionShipper: func(state blockvol.ReplicaState) { + if shipperRef != nil { + shipperRef.TransitionState(state) + } + }, + } +} + func shouldReenterRecoveryFromFailure(reason string) bool { switch reason { case "recoverability_lost", "retention_lost", "truncation_unsafe": diff --git a/weed/server/block_recovery_test.go b/weed/server/block_recovery_test.go index 0d4efcab1..0c9b3dddd 100644 --- a/weed/server/block_recovery_test.go +++ b/weed/server/block_recovery_test.go @@ -795,6 +795,197 @@ func TestP4_ShutdownDrain(t *testing.T) { // --- Rebuild address scoped --- +// ============================================================ +// Probe → rebuild path: component tests for StartRebuildFromProbe +// and the full handleReplicaProbeResult → recovery chain. +// +// These tests close the gap where bugs were only caught on hardware. +// ============================================================ + +// TestStartRebuildFromProbe_InstallsSessionAndReachesPlan verifies the new +// primary-direct rebuild entry point: installSession(SessionRebuild) + startTask. +// The goroutine must find the session, derive the rebuild address from +// ReplicationPorts (no assignments), and reach rebuild planning. +func TestStartRebuildFromProbe_InstallsSessionAndReachesPlan(t *testing.T) { + bs, volPath := createTestBlockServiceWithVolCoreNoRecovery(t) + + // Write data so the volume has content for rebuild planning. + if err := bs.blockStore.WithVolume(volPath, func(vol *blockvol.BlockVol) error { + for i := 0; i < 5; i++ { + if err := vol.WriteLBA(uint64(i), make([]byte, 4096)); err != nil { + return err + } + } + return vol.ForceFlush() + }); err != nil { + t.Fatalf("write+flush: %v", err) + } + + // Set up primary assignment with a replica. + bs.ProcessAssignments([]blockvol.BlockVolumeAssignment{{ + Path: volPath, + Epoch: 1, + Role: uint32(blockvol.RolePrimary), + ReplicaServerID: "vs2", + ReplicaDataAddr: "10.0.0.2:9333", + ReplicaCtrlAddr: "10.0.0.2:9334", + }}) + + replicaID := volPath + "/vs2" + sender := bs.v2Orchestrator.Registry.Sender(replicaID) + if sender == nil { + t.Fatal("expected sender for replica after assignment") + } + + // Fire NeedsRebuildObserved (what handleReplicaProbeResult does before calling us). + bs.applyCoreEvent(engine.NeedsRebuildObserved{ + ID: volPath, + ReplicaID: replicaID, + Reason: "gap_exceeds_retained_wal", + }) + + // Create recovery manager with fake rebuild IO so planning completes. + rm := NewRecoveryManager(bs) + bs.v2Recovery = rm + t.Cleanup(func() { rm.Shutdown() }) + + rebuildReached := make(chan string, 1) + rm.OnPendingExecution = func(volumeID string, pending *rt.PendingExecution) { + if pending != nil && pending.Plan != nil { + pending.RebuildIO = fakeRebuildIO{achievedLSN: pending.Plan.RebuildTargetLSN} + select { + case rebuildReached <- volumeID: + default: + } + } + } + + // Call StartRebuildFromProbe — the method under test. + rm.StartRebuildFromProbe(replicaID) + + // 1. Session must be installed. + sender = bs.v2Orchestrator.Registry.Sender(replicaID) + if sender == nil { + t.Fatal("sender missing after StartRebuildFromProbe") + } + snap := sender.SessionSnapshot() + if snap == nil || !snap.Active || snap.Kind != engine.SessionRebuild { + t.Fatalf("session=%+v, want active rebuild session", snap) + } + + // 2. Recovery goroutine must reach rebuild planning. + select { + case vol := <-rebuildReached: + if vol != volPath { + t.Fatalf("rebuild reached for %s, want %s", vol, volPath) + } + case <-time.After(5 * time.Second): + t.Fatal("rebuild planning not reached within 5s") + } +} + +// TestStartRebuildFromProbe_DeriveRebuildAddr_NilAssignments verifies that +// deriveRebuildAddr produces a valid address when assignments are nil, +// using the ReplicationPorts fallback. +func TestStartRebuildFromProbe_DeriveRebuildAddr_NilAssignments(t *testing.T) { + bs, volPath := createTestBlockServiceWithVol(t) + rm := bs.v2Recovery + + replicaID := volPath + "/vs2" + addr := rm.deriveRebuildAddr(replicaID, nil) + if addr == "" { + t.Fatal("deriveRebuildAddr with nil assignments returned empty — fallback broken") + } + + // The fallback should use the same port as ReplicationPorts. + _, _, rebuildPort := bs.ReplicationPorts(volPath) + expected := fmt.Sprintf("127.0.0.1:%d", rebuildPort) + if addr != expected { + t.Fatalf("deriveRebuildAddr=%s, want %s", addr, expected) + } +} + +// TestHandleReplicaProbeResult_RebuildRequired_FullPath exercises the full +// probe → NeedsRebuildObserved → StartRebuildFromProbe → recovery chain. +// This is the integration seam that was previously untested and required +// hardware runs to find bugs. +func TestHandleReplicaProbeResult_RebuildRequired_FullPath(t *testing.T) { + bs, volPath := createTestBlockServiceWithVolCoreNoRecovery(t) + + if err := bs.blockStore.WithVolume(volPath, func(vol *blockvol.BlockVol) error { + for i := 0; i < 5; i++ { + if err := vol.WriteLBA(uint64(i), make([]byte, 4096)); err != nil { + return err + } + } + return vol.ForceFlush() + }); err != nil { + t.Fatalf("write+flush: %v", err) + } + + bs.ProcessAssignments([]blockvol.BlockVolumeAssignment{{ + Path: volPath, + Epoch: 1, + Role: uint32(blockvol.RolePrimary), + ReplicaServerID: "vs2", + ReplicaDataAddr: "10.0.0.2:9333", + ReplicaCtrlAddr: "10.0.0.2:9334", + }}) + + replicaID := volPath + "/vs2" + + rm := NewRecoveryManager(bs) + bs.v2Recovery = rm + t.Cleanup(func() { rm.Shutdown() }) + + rebuildReached := make(chan string, 1) + rm.OnPendingExecution = func(volumeID string, pending *rt.PendingExecution) { + if pending != nil && pending.Plan != nil { + pending.RebuildIO = fakeRebuildIO{achievedLSN: pending.Plan.RebuildTargetLSN} + select { + case rebuildReached <- volumeID: + default: + } + } + } + + // Simulate what the shipper's ProbeReconnect returns. + bs.handleReplicaProbeResult(volPath, blockvol.ReplicaProbeResult{ + ReplicaID: "vs2", // shipper format, not engine format + Outcome: blockvol.ProbeRebuildRequired, + ReplicaFlushedLSN: 0, + }) + + // 1. Engine should record NeedsRebuild. + proj, ok := bs.CoreProjection(volPath) + if !ok { + t.Fatal("expected core projection") + } + if proj.Recovery.Phase != engine.RecoveryNeedsRebuild { + t.Fatalf("recovery.phase=%s, want %s", proj.Recovery.Phase, engine.RecoveryNeedsRebuild) + } + + // 2. Rebuild session should be installed. + sender := bs.v2Orchestrator.Registry.Sender(replicaID) + if sender == nil { + t.Fatal("sender missing after probe result") + } + snap := sender.SessionSnapshot() + if snap == nil || !snap.Active || snap.Kind != engine.SessionRebuild { + t.Fatalf("session=%+v, want active rebuild", snap) + } + + // 3. Recovery goroutine should reach planning. + select { + case vol := <-rebuildReached: + if vol != volPath { + t.Fatalf("rebuild for %s, want %s", vol, volPath) + } + case <-time.After(5 * time.Second): + t.Fatal("rebuild planning not reached within 5s") + } +} + func TestP4_RebuildAddrScoped(t *testing.T) { bs, _ := createTestBlockServiceWithVol(t) rm := bs.v2Recovery diff --git a/weed/server/volume_server_block.go b/weed/server/volume_server_block.go index d8dc87021..5469de4f0 100644 --- a/weed/server/volume_server_block.go +++ b/weed/server/volume_server_block.go @@ -1420,9 +1420,11 @@ func (bs *BlockService) handleReplicaProbeResult(path string, r blockvol.Replica ReplicaID: engineReplicaID, Reason: "gap_exceeds_retained_wal", }) - // Start the rebuild through the existing recovery manager. + // Install rebuild session then start recovery. No fake + // assignment needed — deriveRebuildAddr computes the address + // from ReplicationPorts. if bs.v2Recovery != nil { - bs.v2Recovery.StartRecoveryTask(engineReplicaID, bs.lastAssignmentsForPath(path)) + bs.v2Recovery.StartRebuildFromProbe(engineReplicaID) } } diff --git a/weed/storage/blockvol/blockvol.go b/weed/storage/blockvol/blockvol.go index 1b2bf3162..15189042c 100644 --- a/weed/storage/blockvol/blockvol.go +++ b/weed/storage/blockvol/blockvol.go @@ -1533,6 +1533,15 @@ func (v *BlockVol) StartRebuildServer(addr string) error { return nil } +// RebuildServerAddr returns the listening address of the rebuild server, +// or empty string if no rebuild server is running. +func (v *BlockVol) RebuildServerAddr() string { + if v.rebuildServer == nil { + return "" + } + return v.rebuildServer.Addr() +} + // StopRebuildServer stops the rebuild server if running. func (v *BlockVol) StopRebuildServer() { if v.rebuildServer != nil { diff --git a/weed/storage/blockvol/rebuild_transport.go b/weed/storage/blockvol/rebuild_transport.go index 73d47e4b6..e8b49b9bd 100644 --- a/weed/storage/blockvol/rebuild_transport.go +++ b/weed/storage/blockvol/rebuild_transport.go @@ -32,21 +32,25 @@ const ( // SessionControlMsg is the wire message for session control commands. // -// Wire version: v1 (33 bytes). This is a new protocol with no deployed peers. -// The initial design had a 37-byte format with a SnapshotID field that was -// removed because the protocol contract uses flushed checkpoint boundaries, -// not explicit snapshot IDs. If future versions need additional fields, the -// decoder should check len(buf) and handle both sizes. +// Wire format: +// v1: 33 bytes [8B epoch][8B sessionID][1B cmd][8B baseLSN][8B targetLSN] +// v2: 33 + 2 + len(addr) bytes: v1 header + [2B addrLen][addrBytes...] +// +// The decoder uses len(buf) to detect the v2 format. v2 adds an optional +// RebuildAddr trailer for start_rebuild commands so the replica knows where +// to connect for the base lane. This keeps start_rebuild atomic — one frame, +// one decode, no multi-message fragility. type SessionControlMsg struct { - Epoch uint64 - SessionID uint64 - Command byte - BaseLSN uint64 // flushed/checkpoint boundary for start_rebuild - TargetLSN uint64 // WAL target for start_rebuild + Epoch uint64 + SessionID uint64 + Command byte + BaseLSN uint64 // flushed/checkpoint boundary for start_rebuild + TargetLSN uint64 // WAL target for start_rebuild + RebuildAddr string // optional: primary's rebuild server address (v2 trailer) } -// EncodeSessionControl serializes a session control message. -// Wire: [8B epoch][8B sessionID][1B cmd][8B baseLSN][8B targetLSN] = 33 bytes. +// EncodeSessionControl serializes a session control message (v1, 33 bytes). +// For messages with RebuildAddr, use EncodeSessionControlV2 instead. func EncodeSessionControl(msg SessionControlMsg) []byte { buf := make([]byte, 33) binary.BigEndian.PutUint64(buf[0:8], msg.Epoch) @@ -57,18 +61,49 @@ func EncodeSessionControl(msg SessionControlMsg) []byte { return buf } +// EncodeSessionControlV2 serializes a session control message with optional +// RebuildAddr trailer. If RebuildAddr is empty, produces the same 33 bytes as v1. +// Wire: [33B v1 header][2B addrLen][addrBytes...] where addrLen may be 0. +func EncodeSessionControlV2(msg SessionControlMsg) []byte { + header := EncodeSessionControl(msg) + if msg.RebuildAddr == "" { + return header + } + addrBytes := []byte(msg.RebuildAddr) + if len(addrBytes) > 65535 { + addrBytes = addrBytes[:65535] + } + buf := make([]byte, 33+2+len(addrBytes)) + copy(buf, header) + binary.BigEndian.PutUint16(buf[33:35], uint16(len(addrBytes))) + copy(buf[35:], addrBytes) + return buf +} + // DecodeSessionControl deserializes a session control message. +// Handles both v1 (33 bytes) and v2 (33 + trailer) formats via len-based detection. func DecodeSessionControl(buf []byte) (SessionControlMsg, error) { if len(buf) < 33 { return SessionControlMsg{}, fmt.Errorf("session control: short message (%d bytes)", len(buf)) } - return SessionControlMsg{ + msg := SessionControlMsg{ Epoch: binary.BigEndian.Uint64(buf[0:8]), SessionID: binary.BigEndian.Uint64(buf[8:16]), Command: buf[16], BaseLSN: binary.BigEndian.Uint64(buf[17:25]), TargetLSN: binary.BigEndian.Uint64(buf[25:33]), - }, nil + } + // v2 trailer: [2B addrLen][addrBytes...] + if len(buf) >= 35 { + addrLen := int(binary.BigEndian.Uint16(buf[33:35])) + if len(buf) < 35+addrLen { + return SessionControlMsg{}, fmt.Errorf("session control: trailer truncated (need %d, have %d)", 35+addrLen, len(buf)) + } + if addrLen > 0 { + msg.RebuildAddr = string(buf[35 : 35+addrLen]) + } + } + return msg, nil } // SessionAckMsg is the wire message for session progress/result. @@ -289,11 +324,17 @@ func (c *RebuildTransportClient) ReceiveBaseBlocksWithStatus(conn net.Conn) (uin return totalBlocks, achievedLSN, nil } -// SendSessionControl sends a session control message on the control connection. +// SendSessionControl sends a v1 session control message on the control connection. func SendSessionControl(conn net.Conn, msg SessionControlMsg) error { return WriteFrame(conn, MsgSessionControl, EncodeSessionControl(msg)) } +// SendSessionControlV2 sends a v2 session control message with optional +// RebuildAddr trailer. Atomic: one frame, one decode on the replica side. +func SendSessionControlV2(conn net.Conn, msg SessionControlMsg) error { + return WriteFrame(conn, MsgSessionControl, EncodeSessionControlV2(msg)) +} + // SendSessionAck sends a session ack message on the control connection. func SendSessionAck(conn net.Conn, msg SessionAckMsg) error { return WriteFrame(conn, MsgSessionAck, EncodeSessionAck(msg)) diff --git a/weed/storage/blockvol/replica_barrier.go b/weed/storage/blockvol/replica_barrier.go index 9a41f2015..420fe3149 100644 --- a/weed/storage/blockvol/replica_barrier.go +++ b/weed/storage/blockvol/replica_barrier.go @@ -89,6 +89,11 @@ func (r *ReplicaReceiver) handleSessionControl(conn net.Conn, payload []byte, wr })) return err } + // If the primary included a RebuildAddr trailer (v2 session control), + // auto-start the base lane client so the replica pulls base blocks. + if ctrl.RebuildAddr != "" { + go r.runBaseLaneClient(ctrl.SessionID, ctrl.RebuildAddr) + } return nil case SessionCmdCancel: if err := r.vol.CancelRebuildSession(ctrl.SessionID, "remote_cancel"); err != nil { @@ -100,6 +105,27 @@ func (r *ReplicaReceiver) handleSessionControl(conn net.Conn, payload []byte, wr } } +// runBaseLaneClient connects to the primary's rebuild server and pulls base +// blocks for the active rebuild session. Runs as a background goroutine started +// by handleSessionControl when the primary includes a RebuildAddr in the v2 +// 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) { + 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() + client := NewRebuildTransportClient(r.vol, sessionID) + blocks, err := client.ReceiveBaseBlocks(conn) + if err != nil { + log.Printf("replica: base lane receive from %s: %v", rebuildAddr, err) + return + } + log.Printf("replica: base lane complete from %s: %d blocks", rebuildAddr, blocks) +} + // handleBarrier waits until all WAL entries up to req.LSN have been received, // then fsyncs the WAL to ensure durability. func (r *ReplicaReceiver) handleBarrier(req BarrierRequest) BarrierResponse { diff --git a/weed/storage/blockvol/wal_shipper.go b/weed/storage/blockvol/wal_shipper.go index 14902355f..54441b3cc 100644 --- a/weed/storage/blockvol/wal_shipper.go +++ b/weed/storage/blockvol/wal_shipper.go @@ -28,6 +28,7 @@ const ( ReplicaInSync ReplicaState = 3 // eligible for sync_all barriers ReplicaDegraded ReplicaState = 4 // transient failure, retry allowed ReplicaNeedsRebuild ReplicaState = 5 // WAL gap too large, rebuild required (CP13-7) + ReplicaRebuilding ReplicaState = 6 // active rebuild session accepted by replica ) func (s ReplicaState) String() string { @@ -44,6 +45,8 @@ func (s ReplicaState) String() string { return "degraded" case ReplicaNeedsRebuild: return "needs_rebuild" + case ReplicaRebuilding: + return "rebuilding" default: return fmt.Sprintf("unknown(%d)", s) } @@ -71,7 +74,8 @@ type WALShipper struct { state atomic.Uint32 // ReplicaState catchupFailures int // consecutive catch-up failures; reset on success lastContactTime atomic.Value // time.Time: last successful barrier/handshake/catch-up - stopped atomic.Bool + stopped atomic.Bool + activeRebuildSession atomic.Bool // true when ReplicaRebuilding AND session confirmed // onStateChange is called when the shipper transitions between states. // Used to trigger immediate heartbeat on degradation/recovery. @@ -127,6 +131,24 @@ func (s *WALShipper) SetLiveShippingPolicy(fn func(replicaID string, entryLSN ui s.liveShippingPolicy = fn } +// TransitionState sets the shipper state and fires the onStateChange callback. +// Used by external coordinators (e.g., RemoteRebuildIO) to drive rebuild state +// transitions: NeedsRebuild → Rebuilding → InSync, or Rebuilding → NeedsRebuild on failure. +// Also manages the activeRebuildSession flag: set on entering Rebuilding, +// cleared on leaving it. +func (s *WALShipper) TransitionState(to ReplicaState) { + from := ReplicaState(s.state.Swap(uint32(to))) + // Manage session flag: entering Rebuilding sets it, leaving clears it. + if to == ReplicaRebuilding { + s.activeRebuildSession.Store(true) + } else if from == ReplicaRebuilding { + s.activeRebuildSession.Store(false) + } + if from != to && s.onStateChange != nil { + s.onStateChange(from, to) + } +} + const maxCatchupRetries = 3 // NewWALShipper creates a WAL shipper. Connections are established lazily on @@ -152,19 +174,28 @@ func NewWALShipper(dataAddr, controlAddr string, epochFn func() uint64, walAcces // the full reconnect protocol. See design/sync-all-reconnect-protocol.md. func (s *WALShipper) Ship(entry *WALEntry) error { st := s.State() - // Ship allowed from Disconnected (bootstrap: data must flow before first barrier) - // and InSync (steady state). All other states reject. - if s.stopped.Load() || (st != ReplicaInSync && st != ReplicaDisconnected) { + // Ship allowed from Disconnected (bootstrap), InSync (steady state), + // and Rebuilding (live WAL lane during active rebuild session). + if s.stopped.Load() || (st != ReplicaInSync && st != ReplicaDisconnected && st != ReplicaRebuilding) { return nil } - if s.liveShippingPolicy != nil { - if allow, reason := s.liveShippingPolicy(s.replicaID, entry.LSN); !allow { - if reason == "" { - reason = "live_shipping_blocked" + // Rebuilding: session-gated authorization. The state alone is not sufficient — + // activeRebuildSession must also be set (confirmed via TransitionState from + // an accepted ack). This prevents stale transitions from opening the live lane. + if st == ReplicaRebuilding && !s.activeRebuildSession.Load() { + return nil + } + // Protocol-level liveShippingPolicy applies only to non-rebuild states. + if st != ReplicaRebuilding { + if s.liveShippingPolicy != nil { + if allow, reason := s.liveShippingPolicy(s.replicaID, entry.LSN); !allow { + if reason == "" { + reason = "live_shipping_blocked" + } + log.Printf("wal_shipper: live ship gated (replica=%s data=%s ctrl=%s lsn=%d reason=%s)", + s.replicaID, s.dataAddr, s.controlAddr, entry.LSN, reason) + return nil } - log.Printf("wal_shipper: live ship gated (replica=%s data=%s ctrl=%s lsn=%d reason=%s)", - s.replicaID, s.dataAddr, s.controlAddr, entry.LSN, reason) - return nil } } // Fresh or late-attached replicas must consume the retained backlog before @@ -451,7 +482,7 @@ func (s *WALShipper) HasTransportContact() bool { switch s.State() { case ReplicaDegraded, ReplicaNeedsRebuild: return false - case ReplicaConnecting, ReplicaCatchingUp, ReplicaInSync: + case ReplicaConnecting, ReplicaCatchingUp, ReplicaInSync, ReplicaRebuilding: return true } if s.ShippedLSN() > 0 {