feat: RemoteRebuildIO — primary coordinates rebuild, replica installs

Replace the broken primary-local rebuild executor with RemoteRebuildIO,
a server-side engine.RebuildIO implementation that coordinates remotely.
The primary sends SessionControlV2 (with RebuildAddr trailer) to the
replica's control channel; the replica starts a local rebuild session
and auto-connects to the primary's rebuild server for the base lane.

Single rebuild route: ALL core-present rebuilds use RemoteRebuildIO.
The entire command chain is preserved unchanged:
  PlanRebuild → pending → RebuildStarted → StartRebuildCommand
  → ExecutePendingRebuild → RemoteRebuildIO.TransferFullBase

Key changes:
- SessionControlMsg v2: optional RebuildAddr trailer (len-based decode)
- ReplicaRebuilding shipper state: session-gated live WAL lane
- RemoteRebuildIO: dials replica ctrl, sends session control, reads acks
- Ack forwarding through ObserveReplicaRebuildSessionAck (pins/watchdog)
- Completion proof from replica's achievedLSN, not primary's local vol
- Transport failures emit SessionFailed (no double-emit on ack failures)
- Progress ack rejection fails closed (stale session = abort)
- Replica auto-starts base lane client on v2 session control

State transitions:
  NeedsRebuild → [accepted ack] → Rebuilding → [completed] → InSync
  Rebuilding → [failed/EOF] → NeedsRebuild → [next probe] → retry

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

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)
+	}
+}

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":

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

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)
 			}
 		}
 

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 {

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))

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 {

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 {