diff --git a/weed/server/block_recovery.go b/weed/server/block_recovery.go index 72c60a059..2e5cf8d3d 100644 --- a/weed/server/block_recovery.go +++ b/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 diff --git a/weed/server/volume_server_block.go b/weed/server/volume_server_block.go index 310b53e21..d8dc87021 100644 --- a/weed/server/volume_server_block.go +++ b/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) { diff --git a/weed/storage/blockvol/blockvol.go b/weed/storage/blockvol/blockvol.go index d92e7d8b7..1b2bf3162 100644 --- a/weed/storage/blockvol/blockvol.go +++ b/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 diff --git a/weed/storage/blockvol/shipper_group.go b/weed/storage/blockvol/shipper_group.go index 139c89b9f..61b44173d 100644 --- a/weed/storage/blockvol/shipper_group.go +++ b/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 diff --git a/weed/storage/blockvol/wal_shipper.go b/weed/storage/blockvol/wal_shipper.go index 8db3e9226..14902355f 100644 --- a/weed/storage/blockvol/wal_shipper.go +++ b/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 {