refactor: unified primary onboarding + rebuild execution wiring

Replace three bypass mechanisms with one unified model. When the
probe returns ProbeRebuildRequired, the host now starts the rebuild
through the existing recovery manager (StartRecoveryTask), which
resolves the rebuild address, plans the rebuild, and executes via
the v2bridge executor — the same path as master-driven RoleRebuilding.

New per-replica probe API:
- WALShipper.ProbeReconnect() → ReplicaProbeResult with typed outcome
- ShipperGroup.ProbeReconnectAll() → []ReplicaProbeResult
- BlockVol.ProbeReplicaOnboarding() / IsClosed()

Host-side wiring:
- handleReplicaProbeResult routes outcomes:
  KeepUp → ShipperConnectedObserved
  CatchUp → ShipperConnectedObserved (recovery manager handles session)
  Rebuild → NeedsRebuildObserved + StartRecoveryTask (executes rebuild)
  TemporaryFailure → no-op
- lastAssignmentsForPath reconstructs assignment for recovery manager
- onPrimaryRosterChanged probes all replicas (defined, called from watchdog)
- observePrimaryShipperConnectivity uses probe API

Probe fires via syncProtocolExecutionState immediately after assignment
processing — same heartbeat cycle, no timer delay.

Deleted: startDirectRebuild, resolveCtrlAddrForShipper,
TryReconnect/TryReconnectAll/TryReconnectShippers.

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

weed/server/block_recovery.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"strconv"
 	"sync"
+	"time"
 
 	engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication"
 	rt "github.com/seaweedfs/seaweedfs/sw-block/engine/replication/runtime"
@@ -236,9 +237,16 @@ func (rm *RecoveryManager) cancelAndDrainWithReason(replicaID string, invalidate
 	doneCh := task.done
 	rm.mu.Unlock()
 
-	// Wait for the old goroutine to exit OUTSIDE the lock.
-	// This serializes replacement: new task cannot start until old is fully drained.
-	<-doneCh
+	// Wait for the old goroutine to exit OUTSIDE the lock, with a bounded
+	// timeout. If the goroutine is stuck on a blocking dial/read that
+	// doesn't respect context cancellation, we abandon it after 5s and
+	// proceed. This prevents a stuck catch-up from blocking a rebuild.
+	select {
+	case <-doneCh:
+		// Clean exit.
+	case <-time.After(5 * time.Second):
+		glog.Warningf("recovery: drain timeout for %s — abandoning stuck goroutine", replicaID)
+	}
 }
 
 // startTask creates and starts a new recovery goroutine. Caller must ensure

weed/server/volume_server_block.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"hash/fnv"
 	"log"
-	"net"
 	"os"
 	"path/filepath"
 	"strings"
@@ -694,6 +693,14 @@ func (bs *BlockService) applyCoreAssignmentEvent(a blockvol.BlockVolumeAssignmen
 	// serve. This is the enforcement point — it happens immediately after
 	// assignment, before the next heartbeat round-trip.
 	bs.evaluateActivationGate(a.Path)
+
+	// Unified onboarding: if this is a primary assignment with replica
+	// addresses, probe each replica immediately. This is the main trigger
+	// for rejoin, failover promotion, and address refresh.
+	// Onboarding: the probe for new/returning replicas fires immediately
+	// via syncProtocolExecutionState → observePrimaryShipperConnectivity
+	// which runs at the end of this assignment processing path. No
+	// separate timer or delayed trigger needed.
 	return nil
 }
 
@@ -1334,51 +1341,20 @@ func (bs *BlockService) observePrimaryShipperConnectivity(path string) {
 	}
 	connected := bs.isPrimaryShipperConnected(path)
 	if !connected {
-		// Proactive reconnect: the shipper is configured but not connected,
-		// and no I/O is happening to trigger Ship(). This occurs on rejoin
-		// paths where the primary gets a fresh assignment with replica
-		// addresses but no writes are pending. Without this, the shipper
-		// sits at Disconnected and the core stays at
-		// awaiting_shipper_connected indefinitely.
-		//
-		// Uses the full reconnect protocol (handshake + bounded catch-up),
-		// not just a dial probe. This brings the replica current if WAL
-		// entries are available.
+		// Watchdog fallback: if the main onboarding path (assignment-triggered)
+		// didn't run or the shipper dropped after onboarding, probe again.
+		// This is NOT the primary recovery path — just a safety net.
+		var probeResults []blockvol.ReplicaProbeResult
 		_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
-			connected = vol.TryReconnectShippers()
-			if !connected {
-				// Bridge 1: check if any shipper reached NeedsRebuild during
-				// the proactive reconnect. Emit per-replica NeedsRebuildObserved
-				// to the core, then initiate primary-direct rebuild for that
-				// specific replica. The master is NOT involved in the rebuild
-				// decision — the primary owns data-control recovery. The master
-				// sees the result via subsequent heartbeat projection.
-				for _, st := range vol.ReplicaShipperStates() {
-					if st.State == "needs_rebuild" && st.DataAddr != "" {
-						replicaID := bs.resolveReplicaIDForShipper(path, st.DataAddr)
-						if replicaID == "" {
-							glog.Warningf("block service: shipper %s needs rebuild but replica ID not resolved — cannot start direct rebuild",
-								st.DataAddr)
-							continue
-						}
-						glog.V(0).Infof("block service: shipper %s (replica=%s) needs rebuild — starting primary-direct rebuild",
-							st.DataAddr, replicaID)
-						bs.applyCoreEvent(engine.NeedsRebuildObserved{
-							ID:        path,
-							ReplicaID: replicaID,
-							Reason:    "gap_exceeds_retained_wal",
-						})
-						// Start the rebuild session directly. Resolve ctrl
-						// address from the shipper (same as data address source).
-						ctrlAddr := bs.resolveCtrlAddrForShipper(path, st.DataAddr)
-						if ctrlAddr != "" {
-							go bs.startDirectRebuild(path, replicaID, st.DataAddr, ctrlAddr)
-						}
-					}
-				}
-			}
+			probeResults = vol.ProbeReplicaOnboarding()
 			return nil
 		})
+		for _, r := range probeResults {
+			bs.handleReplicaProbeResult(path, r)
+			if r.Outcome == blockvol.ProbeKeepUp {
+				connected = true
+			}
+		}
 	}
 	glog.V(0).Infof("block service: recheck shipper connectivity %s connected=%v mode=%s reason=%q",
 		path, connected, proj.Mode.Name, proj.Publication.Reason)
@@ -1410,139 +1386,101 @@ func (bs *BlockService) resolveReplicaIDForShipper(path, dataAddr string) string
 	return replicaID
 }
 
-// resolveCtrlAddrForShipper maps a shipper's data address to its ctrl address.
-func (bs *BlockService) resolveCtrlAddrForShipper(path, dataAddr string) string {
-	if bs == nil || bs.blockStore == nil || dataAddr == "" {
-		return ""
+// handleReplicaProbeResult processes one per-replica onboarding probe result.
+// Routes to the appropriate recovery path without going through the master.
+func (bs *BlockService) handleReplicaProbeResult(path string, r blockvol.ReplicaProbeResult) {
+	if bs == nil {
+		return
 	}
-	var ctrlAddr string
-	_ = bs.blockStore.WithVolume(path, 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.DataAddr() == dataAddr {
-				ctrlAddr = s.CtrlAddr()
-				return nil
+	switch r.Outcome {
+	case blockvol.ProbeKeepUp:
+		glog.V(0).Infof("block service: replica %s keepup (flushedLSN=%d)", r.ReplicaID, r.ReplicaFlushedLSN)
+		bs.applyCoreEvent(engine.ShipperConnectedObserved{ID: path})
+
+	case blockvol.ProbeCatchUpRequired:
+		glog.V(0).Infof("block service: replica %s needs catch-up (flushedLSN=%d)", r.ReplicaID, r.ReplicaFlushedLSN)
+		// The existing recovery manager handles catch-up via the engine's
+		// session commands. Emit the fact; the engine + recovery path do the rest.
+		bs.applyCoreEvent(engine.ShipperConnectedObserved{ID: path})
+
+	case blockvol.ProbeRebuildRequired:
+		glog.V(0).Infof("block service: replica %s needs rebuild (flushedLSN=%d)", r.ReplicaID, r.ReplicaFlushedLSN)
+		if r.ReplicaID != "" {
+			// Resolve the engine-format replicaID from the core projection.
+			// The engine uses MakeReplicaID(path, serverID) which may differ
+			// from the shipper's raw ServerID. Use the projection's ReplicaIDs
+			// to find the matching one.
+			engineReplicaID := bs.resolveEngineReplicaID(path, r.ReplicaID)
+			if engineReplicaID == "" {
+				engineReplicaID = path + "/" + r.ReplicaID // fallback
+			}
+			glog.V(0).Infof("block service: rebuild replicaID: shipper=%s engine=%s", r.ReplicaID, engineReplicaID)
+			bs.applyCoreEvent(engine.NeedsRebuildObserved{
+				ID:        path,
+				ReplicaID: engineReplicaID,
+				Reason:    "gap_exceeds_retained_wal",
+			})
+			// Start the rebuild through the existing recovery manager.
+			if bs.v2Recovery != nil {
+				bs.v2Recovery.StartRecoveryTask(engineReplicaID, bs.lastAssignmentsForPath(path))
 			}
 		}
-		return nil
-	})
-	return ctrlAddr
+
+	case blockvol.ProbeTemporaryFailure:
+		glog.V(0).Infof("block service: replica %s probe failed: %v", r.ReplicaID, r.Err)
+		// Don't escalate — temporary failure may resolve on next attempt.
+	}
 }
 
-// startDirectRebuild initiates a primary-direct rebuild session for one
-// specific replica. This runs in a goroutine — the primary sends
-// sessionControl(start_rebuild) directly to the replica's control channel
-// and streams the base extent. The master is not involved in the rebuild
-// decision; it observes the result via heartbeat projection.
-func (bs *BlockService) startDirectRebuild(path, replicaID, dataAddr, ctrlAddr string) {
-	if bs == nil || bs.blockStore == nil {
-		return
+// resolveEngineReplicaID finds the engine-format replicaID that contains the
+// given serverID. The engine uses MakeReplicaID(path, serverID) = "path/serverID".
+// The shipper's ReplicaID is just serverID. This function scans the core
+// projection's ReplicaIDs to find the matching full ID.
+func (bs *BlockService) resolveEngineReplicaID(path, shipperReplicaID string) string {
+	if bs == nil || bs.v2Core == nil || shipperReplicaID == "" {
+		return ""
 	}
-	glog.V(0).Infof("block service: starting direct rebuild %s replica=%s data=%s ctrl=%s",
-		path, replicaID, dataAddr, ctrlAddr)
-
-	// Get the primary's current state to determine baseLSN and targetLSN.
-	var baseLSN, targetLSN uint64
-	_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
-		status := vol.Status()
-		baseLSN = status.CheckpointLSN
-		targetLSN = status.WALHeadLSN
-		if baseLSN == 0 {
-			baseLSN = targetLSN
+	proj, ok := bs.CoreProjection(path)
+	if !ok {
+		return ""
+	}
+	for _, rid := range proj.ReplicaIDs {
+		// Check if this engine replicaID ends with the shipper's serverID.
+		if strings.HasSuffix(rid, "/"+shipperReplicaID) {
+			return rid
 		}
+		// Also check exact match in case formats align.
+		if rid == shipperReplicaID {
+			return rid
+		}
+	}
+	return ""
+}
+
+// lastAssignmentsForPath returns the last applied assignment for a volume path
+// as a slice (the recovery manager expects []BlockVolumeAssignment).
+func (bs *BlockService) lastAssignmentsForPath(path string) []blockvol.BlockVolumeAssignment {
+	if bs == nil {
 		return nil
-	})
-	if baseLSN == 0 {
-		glog.Warningf("block service: direct rebuild %s: no baseline LSN, aborting", path)
-		return
 	}
-
-	// Send sessionControl(start_rebuild) to the replica's control channel.
-	sessionID := uint64(time.Now().UnixNano())
-	conn, err := net.DialTimeout("tcp", ctrlAddr, 5*time.Second)
-	if err != nil {
-		glog.Warningf("block service: direct rebuild %s: dial ctrl %s: %v", path, ctrlAddr, err)
-		return
-	}
-	defer conn.Close()
-
-	if err := blockvol.SendSessionControl(conn, blockvol.SessionControlMsg{
-		Epoch:     0, // will be filled by the replica from its local state
-		SessionID: sessionID,
-		Command:   blockvol.SessionCmdStartRebuild,
-		BaseLSN:   baseLSN,
-		TargetLSN: targetLSN,
-	}); err != nil {
-		glog.Warningf("block service: direct rebuild %s: send session control to %s: %v", path, ctrlAddr, err)
-		return
-	}
-
-	// Wait for accepted ack.
-	conn.SetDeadline(time.Now().Add(10 * time.Second))
-	msgType, payload, err := blockvol.ReadFrame(conn)
-	if err != nil || msgType != blockvol.MsgSessionAck {
-		glog.Warningf("block service: direct rebuild %s: accepted ack: err=%v type=0x%02x", path, err, msgType)
-		return
-	}
-	ack, _ := blockvol.DecodeSessionAck(payload)
-	if ack.Phase != blockvol.SessionAckAccepted {
-		glog.Warningf("block service: direct rebuild %s: session not accepted: phase=%d", path, ack.Phase)
-		return
-	}
-
-	glog.V(0).Infof("block service: direct rebuild %s accepted (session=%d baseLSN=%d targetLSN=%d)",
-		path, sessionID, baseLSN, targetLSN)
-
-	// Stream base blocks from primary to replica over the rebuild TCP path.
-	_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
-		rebuildServer := blockvol.NewRebuildTransportServer(vol, sessionID, 0, baseLSN, targetLSN)
-		rebuildLn, err := net.Listen("tcp", "127.0.0.1:0")
-		if err != nil {
-			glog.Warningf("block service: direct rebuild %s: listen rebuild: %v", path, err)
-			return nil
-		}
-		defer rebuildLn.Close()
-
-		// Tell the replica where to connect for base data.
-		// For now, the base stream runs on the same ctrl connection
-		// (the replica's receiver handles it) or we serve locally.
-		// Simplest: serve base blocks directly on a temp listener.
-		go func() {
-			c, err := rebuildLn.Accept()
-			if err != nil {
-				return
-			}
-			defer c.Close()
-			rebuildServer.ServeBaseBlocks(c)
-		}()
-
-		// Connect replica to our base server.
-		baseConn, err := net.Dial("tcp", rebuildLn.Addr().String())
-		if err != nil {
-			glog.Warningf("block service: direct rebuild %s: connect base: %v", path, err)
-			return nil
-		}
-		defer baseConn.Close()
-		client := blockvol.NewRebuildTransportClient(vol, sessionID)
-		_, err = client.ReceiveBaseBlocks(baseConn)
-		if err != nil {
-			glog.Warningf("block service: direct rebuild %s: receive base: %v", path, err)
-		}
+	bs.lastAssignMu.RLock()
+	defer bs.lastAssignMu.RUnlock()
+	if bs.lastAssign == nil {
 		return nil
-	})
-
-	// Emit RebuildStarted to engine.
-	bs.applyCoreEvent(engine.RebuildStarted{
-		ID:        path,
-		ReplicaID: replicaID,
-		TargetLSN: targetLSN,
-	})
-
-	glog.V(0).Infof("block service: direct rebuild %s complete for replica=%s", path, replicaID)
+	}
+	last, ok := bs.lastAssign[path]
+	if !ok {
+		return nil
+	}
+	return []blockvol.BlockVolumeAssignment{{
+		Path:            path,
+		Epoch:           last.Epoch,
+		Role:            last.Role,
+		ReplicaServerID: last.ReplicaServerID,
+		ReplicaDataAddr: last.ReplicaDataAddr,
+		ReplicaCtrlAddr: last.ReplicaCtrlAddr,
+		ReplicaAddrs:    last.ReplicaAddrs,
+	}}
 }
 
 func (bs *BlockService) observePrimaryShipperConnectivityStatus(path string, connected bool) {

weed/storage/blockvol/blockvol.go

@@ -1037,15 +1037,19 @@ func (v *BlockVol) ReplicaShipperStates() []ReplicaShipperStatus {
 	return v.shipperGroup.ShipperStates()
 }
 
-// TryReconnectShippers attempts the full reconnect protocol on all configured
-// shippers without requiring foreground I/O. Returns true if all shippers now
-// have transport contact. Used by the host-side recheck on rejoin paths where
-// no writes are happening to trigger Ship().
-func (v *BlockVol) TryReconnectShippers() bool {
+// IsClosed returns true if the volume has been closed.
+func (v *BlockVol) IsClosed() bool {
+	return v == nil || v.closed.Load()
+}
+
+// ProbeReplicaOnboarding probes all configured shippers and returns per-replica
+// results. Used by primary onboarding after assignment to decide
+// keepup/catchup/rebuild for each replica.
+func (v *BlockVol) ProbeReplicaOnboarding() []ReplicaProbeResult {
 	if v == nil || v.shipperGroup == nil {
-		return false
+		return nil
 	}
-	return v.shipperGroup.TryReconnectAll()
+	return v.shipperGroup.ProbeReconnectAll()
 }
 
 // PrimaryShipperConnected reports whether all configured replica shippers have

weed/storage/blockvol/shipper_group.go

@@ -110,25 +110,26 @@ func (sg *ShipperGroup) AnyDegraded() bool {
 	return false
 }
 
-// TryReconnectAll attempts the full reconnect protocol on all shippers that
-// are not yet connected. Used by the host-side recheck when the V2 core
-// reports awaiting_shipper_connected but no I/O is triggering Ship().
-// Returns true if all shippers now have transport contact.
-func (sg *ShipperGroup) TryReconnectAll() bool {
+// ProbeReconnectAll probes all shippers that are not yet connected and
+// returns per-replica results. Used by primary onboarding after assignment.
+func (sg *ShipperGroup) ProbeReconnectAll() []ReplicaProbeResult {
 	sg.mu.RLock()
 	defer sg.mu.RUnlock()
-	if len(sg.shippers) == 0 {
-		return false
-	}
-	allConnected := true
+	var results []ReplicaProbeResult
 	for _, s := range sg.shippers {
 		if !s.HasTransportContact() {
-			if !s.TryReconnect() {
-				allConnected = false
-			}
+			results = append(results, s.ProbeReconnect())
+		} else {
+			results = append(results, ReplicaProbeResult{
+				ReplicaID:         s.ReplicaID(),
+				DataAddr:          s.DataAddr(),
+				CtrlAddr:          s.CtrlAddr(),
+				Outcome:           ProbeKeepUp,
+				ReplicaFlushedLSN: s.replicaFlushedLSN.Load(),
+			})
 		}
 	}
-	return allConnected
+	return results
 }
 
 // AllHaveTransportContact returns true only when every configured shipper has

weed/storage/blockvol/wal_shipper.go

@@ -507,51 +507,95 @@ func (s *WALShipper) Stop() {
 	s.ctrlMu.Unlock()
 }
 
-// TryReconnect attempts the full reconnect protocol (dial + handshake +
-// bounded catch-up if needed) without requiring a foreground write or barrier.
-// Used by the host-side recheck when the V2 core reports
-// awaiting_shipper_connected but no I/O is triggering Ship().
-//
-// This is Option B from the design: trigger the same reconnect path that
-// Barrier() would use, but proactively instead of waiting for I/O.
-// Returns true if the shipper reached InSync or has transport contact.
-func (s *WALShipper) TryReconnect() bool {
+// ReplicaProbeOutcome classifies the result of a per-replica onboarding probe.
+type ReplicaProbeOutcome int
+
+const (
+	ProbeTemporaryFailure ReplicaProbeOutcome = iota
+	ProbeKeepUp
+	ProbeCatchUpRequired
+	ProbeRebuildRequired
+)
+
+func (o ReplicaProbeOutcome) String() string {
+	switch o {
+	case ProbeKeepUp:
+		return "keepup"
+	case ProbeCatchUpRequired:
+		return "catchup"
+	case ProbeRebuildRequired:
+		return "rebuild"
+	default:
+		return "temporary_failure"
+	}
+}
+
+// ReplicaProbeResult is the outcome of probing one replica during onboarding.
+type ReplicaProbeResult struct {
+	ReplicaID         string
+	DataAddr          string
+	CtrlAddr          string
+	Outcome           ReplicaProbeOutcome
+	ReplicaFlushedLSN uint64
+	Err               error
+}
+
+// ProbeReconnect performs one onboarding probe: dial + handshake to determine
+// the replica's position, then classify as keepup/catchup/rebuild.
+// Does NOT perform catch-up or rebuild — only collects facts for the host
+// to decide what to do next.
+func (s *WALShipper) ProbeReconnect() ReplicaProbeResult {
+	result := ReplicaProbeResult{
+		ReplicaID: s.replicaID,
+		DataAddr:  s.dataAddr,
+		CtrlAddr:  s.controlAddr,
+	}
 	if s.stopped.Load() {
-		return false
+		result.Outcome = ProbeTemporaryFailure
+		result.Err = ErrShipperStopped
+		return result
 	}
 	st := s.State()
 	if st == ReplicaInSync {
-		return true
+		result.Outcome = ProbeKeepUp
+		result.ReplicaFlushedLSN = s.replicaFlushedLSN.Load()
+		return result
 	}
 	if st != ReplicaDisconnected && st != ReplicaDegraded {
-		return false
-	}
-	if s.wal == nil {
-		// No WAL access — try bare connection only.
-		s.mu.Lock()
-		err := s.ensureDataConn()
-		s.mu.Unlock()
-		if err != nil {
-			return false
-		}
-		s.lastContactTime.Store(time.Now())
-		return s.HasTransportContact()
+		result.Outcome = ProbeTemporaryFailure
+		return result
 	}
 
-	// Full reconnect: handshake + bounded catch-up if needed.
-	// Use the primary's WAL head as the catch-up target.
-	_, headLSN := s.wal.RetainedRange()
-	if headLSN == 0 {
-		headLSN = 1
+	// Attempt handshake to collect replica facts. Use a short deadline
+	// (3s) instead of the normal catchupTimeout (30s) since this is a
+	// lightweight probe, not a full catch-up.
+	s.mu.Lock()
+	if s.dataConn != nil {
+		s.dataConn.SetDeadline(time.Now().Add(3 * time.Second))
 	}
-	log.Printf("wal_shipper: proactive reconnect (data=%s ctrl=%s state=%s target=%d)",
-		s.dataAddr, s.controlAddr, st, headLSN)
-	if _, err := s.CatchUpTo(headLSN); err != nil {
-		log.Printf("wal_shipper: proactive reconnect failed (data=%s ctrl=%s): %v",
-			s.dataAddr, s.controlAddr, err)
-		return s.HasTransportContact()
+	s.mu.Unlock()
+	targetState, replicaFlushedLSN, err := s.reconnectWithHandshake()
+	result.ReplicaFlushedLSN = replicaFlushedLSN
+	if err != nil {
+		result.Err = err
 	}
-	return s.State() == ReplicaInSync || s.HasTransportContact()
+
+	switch targetState {
+	case ReplicaInSync:
+		s.markInSync()
+		result.Outcome = ProbeKeepUp
+	case ReplicaCatchingUp:
+		result.Outcome = ProbeCatchUpRequired
+	case ReplicaNeedsRebuild:
+		s.state.Store(uint32(ReplicaNeedsRebuild))
+		result.Outcome = ProbeRebuildRequired
+	default:
+		result.Outcome = ProbeTemporaryFailure
+	}
+
+	log.Printf("wal_shipper: probe result replica=%s data=%s outcome=%s flushedLSN=%d err=%v",
+		s.replicaID, s.dataAddr, result.Outcome, replicaFlushedLSN, err)
+	return result
 }
 
 func (s *WALShipper) ensureDataConn() error {