diff --git a/sw-block/.private/phase/phase-16-log.md b/sw-block/.private/phase/phase-16-log.md index 10e716b77..668566d22 100644 --- a/sw-block/.private/phase/phase-16-log.md +++ b/sw-block/.private/phase/phase-16-log.md @@ -782,3 +782,73 @@ Validation: 1. `go test ./weed/server -run "TestBlockService_ApplyAssignments_RebuildingRole_(UsesCoreRecoveryPathWithoutLegacyDirectStart|PreservesLegacyFallbackWithoutCore)"` 2. `go test ./weed/server -run "Test(P4_|P16B_|BlockService_(ApplyAssignments_(RebuildingRole_|ExecutesCoreCommands_)|BarrierRejected|DebugInfoForVolume|CollectBlockVolumeHeartbeat|ReadinessSnapshot|HeartbeatReplicaDegraded)|Registry_(ReplicaReadyRequiresReplicaHeartbeat|UpdateFullHeartbeat|UpdateFullHeartbeat_ConsumesCoreInfluencedReplicaDegraded|UpdateFullHeartbeat_ConsumesCoreInfluencedReplicaReady)|EntryToVolumeInfo_(IncludesHealthState|ReflectsCoreInfluencedReadyConsume|ReflectsCoreInfluencedDegradedConsume)|BlockVolume(LookupHandler_ReflectsCoreInfluencedReadyConsume|ListHandler_ReflectsCoreInfluencedDegradedConsume)|BlockStatusHandler_(IncludesHealthCounts|ReflectsCoreInfluencedConsumeCounts)|LookupResponseFromEntry_PublicationMinimalSurface)"` 3. result: `PASS` + +--- + +#### `16F` Start Note Rev 1 + +Date: 2026-04-04 +Scope: replica-scoped recovery command addressing on the bounded core-present +paths + +Why this slice exists: + +1. `16E` moved recovery-task startup into the core command path +2. but `start_catchup` / `start_rebuild` and pending execution still address + only one volume-wide slot +3. that volume-scoped slot blocks any later widening toward multi-replica + startup ownership because concurrent replica targets would overwrite each + other + +Chosen implementation rule: + +1. do not yet broaden startup ownership claims +2. first make the bounded recovery execution commands replica-scoped +3. keep proof posture narrow to the already-accepted single-replica primary + path plus the bounded rebuilding path + +--- + +#### `16F` Delivery Note Rev 1 + +Date: 2026-04-04 +Scope: replica-scoped recovery command addressing on the bounded core-present +paths + +What changed: + +1. `sw-block/engine/replication/command.go` + - `StartCatchUpCommand` now carries `replicaID` + - `StartRebuildCommand` now carries `replicaID` +2. `sw-block/engine/replication/state.go` + - command memory now tracks recovery-task / catch-up / rebuild issuance by + replica target instead of one volume-scoped slot +3. `sw-block/engine/replication/engine.go` + - bounded catch-up/rebuild execution commands now emit replica-scoped + addressing + - startup breadth is still bounded; this slice does not yet claim broad + multi-replica startup ownership +4. `sw-block/engine/replication/runtime/pending.go` + - `PendingCoordinator` now stores and matches pending execution by replica + target +5. `weed/server/blockcmd` + - recovery execution dispatch now forwards replica-scoped addressing +6. `weed/server/block_recovery.go` + - pending catch-up/rebuild execution now takes and executes plans by + `replicaID` + +Proof / evidence: + +1. `go test ./...` from `sw-block/engine/replication` +2. `go test ./weed/server/blockcmd -count=1 -timeout 60s` +3. `go test ./weed/server -count=1 -timeout 120s -run "Test(P16B_|BlockService_(ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart|RebuildingRole_PreservesLegacyFallbackWithoutCore)|DebugInfoForVolume|CollectBlockVolumeHeartbeat|ReadinessSnapshot|HeartbeatReplicaDegraded))"` +4. result: `PASS` + +Conclusion: + +1. the bounded core-present recovery path no longer relies on a volume-scoped + pending slot +2. the structural blocker for any future multi-replica startup-ownership + widening is reduced +3. this slice still does not claim broad multi-replica startup ownership or + full recovery-loop closure diff --git a/sw-block/.private/phase/phase-16.md b/sw-block/.private/phase/phase-16.md index 2ee271b74..efe498b5c 100644 --- a/sw-block/.private/phase/phase-16.md +++ b/sw-block/.private/phase/phase-16.md @@ -196,7 +196,51 @@ Current chosen path: Status: -1. active +1. delivered + +### `16F`: Replica-Scoped Recovery Command Addressing + +Goal: + +1. remove the remaining volume-scoped recovery command/pending slot from the + bounded core-present recovery path +2. make `start_catchup` / `start_rebuild` address the intended replica + explicitly, even before broad multi-replica ownership is claimed + +Acceptance object: + +1. bounded core-emitted recovery execution commands identify the target + `replicaID` +2. pending recovery execution is keyed by replica target instead of a single + volume-wide slot +3. current single-replica catch-up and rebuilding paths remain green +4. this slice does not yet claim broad multi-replica startup ownership + +Current chosen path: + +1. `StartCatchUpCommand` carries `replicaID` on the bounded single-replica + primary path +2. `StartRebuildCommand` carries `replicaID` on the bounded rebuilding path +3. `PendingCoordinator` matches pending execution by replica target +4. command dispatch / recovery manager execution follow the same replica-scoped + addressing + +Status: + +1. delivered + +Delivered result: + +1. `StartCatchUpCommand` and `StartRebuildCommand` now carry `replicaID` +2. pending recovery execution is matched by replica target instead of one + volume-wide slot +3. the bounded single-replica primary catch-up path and bounded rebuilding path + continue to run unchanged in behavior, but now through replica-scoped + recovery addressing + +Evidence: + +1. focused working-tree change after `145327498` ## Current Checkpoint Review Target @@ -220,6 +264,7 @@ after `Phase 15` closeout: - rebuild recovery-task startup ownership - bounded catch-up recovery-task startup ownership on the single-replica primary path + - replica-scoped recovery command addressing on those same bounded paths This checkpoint is intentionally still bounded: @@ -247,11 +292,15 @@ boundary: 6. `16E` current bounded refinement: - catch-up recovery-task startup is core-command-driven on the single-replica primary path +7. `16F` delivered: + - recovery execution commands / pending matching are replica-scoped on the + same bounded paths After this checkpoint: 1. keep `legacy P4` only as a compatibility guard -2. decide whether to widen startup ownership beyond the bounded single-replica - catch-up path +2. the next bounded semantic/runtime decision is whether to widen startup + ownership beyond the single-replica catch-up path now that recovery + addressing is replica-scoped 3. do not yet claim full recovery-loop closure 4. do not broaden into launch claims diff --git a/sw-block/engine/replication/command.go b/sw-block/engine/replication/command.go index 6290e0c91..c3ba0a8b5 100644 --- a/sw-block/engine/replication/command.go +++ b/sw-block/engine/replication/command.go @@ -37,6 +37,7 @@ func (StartRecoveryTaskCommand) commandName() string { return "start_recovery_ta type StartCatchUpCommand struct { VolumeID string + ReplicaID string TargetLSN uint64 } @@ -44,6 +45,7 @@ func (StartCatchUpCommand) commandName() string { return "start_catchup" } type StartRebuildCommand struct { VolumeID string + ReplicaID string TargetLSN uint64 } diff --git a/sw-block/engine/replication/engine.go b/sw-block/engine/replication/engine.go index 7ed9385c5..9e6487398 100644 --- a/sw-block/engine/replication/engine.go +++ b/sw-block/engine/replication/engine.go @@ -91,12 +91,16 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { st.Boundary.TargetLSN = v.TargetLSN } st.Recovery.Reason = "" - if st.shouldStartCatchUp(v.TargetLSN) { + if replicaID, ok := st.recoveryCommandReplicaID(); ok && st.shouldStartCatchUp(replicaID, v.TargetLSN) { cmds = append(cmds, StartCatchUpCommand{ VolumeID: st.VolumeID, + ReplicaID: replicaID, TargetLSN: v.TargetLSN, }) - st.commands.CatchUpTargetLSN = v.TargetLSN + if st.commands.CatchUpTargets == nil { + st.commands.CatchUpTargets = make(map[string]uint64) + } + st.commands.CatchUpTargets[replicaID] = v.TargetLSN } case RecoveryProgressObserved: @@ -119,7 +123,7 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { } st.Recovery.Phase = RecoveryIdle st.Recovery.Reason = "" - st.commands.CatchUpTargetLSN = 0 + st.commands.CatchUpTargets = nil case NeedsRebuildObserved: st.needsRebuild = true @@ -148,12 +152,16 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { if v.TargetLSN > st.Boundary.TargetLSN { st.Boundary.TargetLSN = v.TargetLSN } - if st.shouldStartRebuild(v.TargetLSN) { + if replicaID, ok := st.recoveryCommandReplicaID(); ok && st.shouldStartRebuild(replicaID, v.TargetLSN) { cmds = append(cmds, StartRebuildCommand{ VolumeID: st.VolumeID, + ReplicaID: replicaID, TargetLSN: v.TargetLSN, }) - st.commands.RebuildTargetLSN = v.TargetLSN + if st.commands.RebuildTargets == nil { + st.commands.RebuildTargets = make(map[string]uint64) + } + st.commands.RebuildTargets[replicaID] = v.TargetLSN } case RebuildCommitted: @@ -180,7 +188,7 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { if achievedLSN > st.Boundary.AchievedLSN { st.Boundary.AchievedLSN = achievedLSN } - st.commands.RebuildTargetLSN = 0 + st.commands.RebuildTargets = nil } e.recompute(st) @@ -310,10 +318,9 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) [] st.Boundary.TargetLSN = 0 st.Boundary.AchievedLSN = 0 st.commands.RecoveryTaskEpoch = 0 - st.commands.RecoveryTaskReplicaID = "" - st.commands.RecoveryTaskKind = "" - st.commands.CatchUpTargetLSN = 0 - st.commands.RebuildTargetLSN = 0 + st.commands.RecoveryTaskTargets = nil + st.commands.CatchUpTargets = nil + st.commands.RebuildTargets = nil } var cmds []Command @@ -338,16 +345,17 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) [] st.commands.ShipperConfigEpoch = st.Epoch st.commands.ShipperConfigReplicas = append([]ReplicaAssignment(nil), st.DesiredReplicas...) } - if st.shouldStartRecoveryTask() { - replicaID := st.DesiredReplicas[0].ReplicaID + if replicaID, ok := st.recoveryCommandReplicaID(); ok && st.shouldStartRecoveryTask(replicaID) { cmds = append(cmds, StartRecoveryTaskCommand{ VolumeID: st.VolumeID, ReplicaID: replicaID, Kind: st.recoveryTarget, }) st.commands.RecoveryTaskEpoch = st.Epoch - st.commands.RecoveryTaskReplicaID = replicaID - st.commands.RecoveryTaskKind = st.recoveryTarget + if st.commands.RecoveryTaskTargets == nil { + st.commands.RecoveryTaskTargets = make(map[string]SessionKind) + } + st.commands.RecoveryTaskTargets[replicaID] = st.recoveryTarget } return cmds } @@ -389,14 +397,17 @@ func (st *VolumeState) shouldStartReceiver() bool { st.commands.ReceiverStartEpoch != st.Epoch } -func (st *VolumeState) shouldStartRecoveryTask() bool { - if st.recoveryTarget == "" || len(st.DesiredReplicas) != 1 { +func (st *VolumeState) shouldStartRecoveryTask(replicaID string) bool { + if st.recoveryTarget == "" || replicaID == "" { return false } - replicaID := st.DesiredReplicas[0].ReplicaID - return st.commands.RecoveryTaskEpoch != st.Epoch || - st.commands.RecoveryTaskReplicaID != replicaID || - st.commands.RecoveryTaskKind != st.recoveryTarget + if st.commands.RecoveryTaskEpoch != st.Epoch { + return true + } + if st.commands.RecoveryTaskTargets == nil { + return true + } + return st.commands.RecoveryTaskTargets[replicaID] != st.recoveryTarget } func (st *VolumeState) shouldConfigureShipper() bool { @@ -411,12 +422,24 @@ func (st *VolumeState) shouldInvalidate(reason string) bool { return !st.commands.InvalidationIssued || st.commands.InvalidationReason != reason } -func (st *VolumeState) shouldStartCatchUp(targetLSN uint64) bool { - return targetLSN > 0 && st.commands.CatchUpTargetLSN != targetLSN +func (st *VolumeState) shouldStartCatchUp(replicaID string, targetLSN uint64) bool { + if targetLSN == 0 || replicaID == "" { + return false + } + if st.commands.CatchUpTargets == nil { + return true + } + return st.commands.CatchUpTargets[replicaID] != targetLSN } -func (st *VolumeState) shouldStartRebuild(targetLSN uint64) bool { - return targetLSN > 0 && st.commands.RebuildTargetLSN != targetLSN +func (st *VolumeState) shouldStartRebuild(replicaID string, targetLSN uint64) bool { + if targetLSN == 0 || replicaID == "" { + return false + } + if st.commands.RebuildTargets == nil { + return true + } + return st.commands.RebuildTargets[replicaID] != targetLSN } func (st *VolumeState) resetInvalidation() { @@ -428,6 +451,13 @@ func (st *VolumeState) hasReplicas() bool { return len(st.DesiredReplicas) > 0 } +func (st *VolumeState) recoveryCommandReplicaID() (string, bool) { + if len(st.DesiredReplicas) != 1 || st.DesiredReplicas[0].ReplicaID == "" { + return "", false + } + return st.DesiredReplicas[0].ReplicaID, true +} + func (st *VolumeState) bootstrapReason() string { switch { case !st.Readiness.RoleApplied: diff --git a/sw-block/engine/replication/phase14_command_test.go b/sw-block/engine/replication/phase14_command_test.go index 033b65394..953bc3016 100644 --- a/sw-block/engine/replication/phase14_command_test.go +++ b/sw-block/engine/replication/phase14_command_test.go @@ -150,6 +150,13 @@ func TestPhase14_CommandSequence_CatchUpStartIsBounded(t *testing.T) { "start_catchup", "publish_projection", }) + start, ok := result.Commands[0].(StartCatchUpCommand) + if !ok { + t.Fatalf("cmd0=%T", result.Commands[0]) + } + if start.ReplicaID != "replica-1" { + t.Fatalf("replica_id=%q", start.ReplicaID) + } result = core.ApplyEvent(CatchUpPlanned{ID: "vol-cmd-catchup", TargetLSN: 55}) assertCommandNames(t, result.Commands, nil) @@ -174,6 +181,13 @@ func TestPhase14_CommandSequence_RebuildStartIsBounded(t *testing.T) { "start_rebuild", "publish_projection", }) + start, ok := result.Commands[0].(StartRebuildCommand) + if !ok { + t.Fatalf("cmd0=%T", result.Commands[0]) + } + if start.ReplicaID != "replica-1" { + t.Fatalf("replica_id=%q", start.ReplicaID) + } result = core.ApplyEvent(RebuildStarted{ID: "vol-cmd-rebuild", TargetLSN: 80}) assertCommandNames(t, result.Commands, nil) diff --git a/sw-block/engine/replication/runtime/pending.go b/sw-block/engine/replication/runtime/pending.go index 4eef5291e..2a7108937 100644 --- a/sw-block/engine/replication/runtime/pending.go +++ b/sw-block/engine/replication/runtime/pending.go @@ -10,7 +10,7 @@ import ( ) // PendingExecution holds the state needed to execute a planned recovery -// action. The coordinator stores one pending execution per volume and +// action. The coordinator stores one pending execution per replica target and // matches it against incoming commands. // // All fields are typed — no interface{} handles. The host adapter builds @@ -63,22 +63,23 @@ func NewPendingCoordinator(cancelFn CancelFunc) *PendingCoordinator { } } -// Store caches a pending execution for a volume, replacing any previous one. -func (pc *PendingCoordinator) Store(volumeID string, pe *PendingExecution) { +// Store caches a pending execution for one replica target, replacing any +// previous one. +func (pc *PendingCoordinator) Store(replicaID string, pe *PendingExecution) { pc.mu.Lock() defer pc.mu.Unlock() - pc.pending[volumeID] = pe + pc.pending[replicaID] = pe } -// TakeCatchUp takes the pending execution for the volume if the catch-up +// TakeCatchUp takes the pending execution for the replica if the catch-up // target matches. If there's a mismatch, the pending execution is cancelled // (fail-closed) and nil is returned. If no pending execution exists, nil // is returned. -func (pc *PendingCoordinator) TakeCatchUp(volumeID string, targetLSN uint64) *PendingExecution { +func (pc *PendingCoordinator) TakeCatchUp(replicaID string, targetLSN uint64) *PendingExecution { pc.mu.Lock() - pe, ok := pc.pending[volumeID] + pe, ok := pc.pending[replicaID] if ok { - delete(pc.pending, volumeID) + delete(pc.pending, replicaID) } pc.mu.Unlock() @@ -94,13 +95,13 @@ func (pc *PendingCoordinator) TakeCatchUp(volumeID string, targetLSN uint64) *Pe return pe } -// TakeRebuild takes the pending execution for the volume if the rebuild +// TakeRebuild takes the pending execution for the replica if the rebuild // target matches. Same fail-closed semantics as TakeCatchUp. -func (pc *PendingCoordinator) TakeRebuild(volumeID string, targetLSN uint64) *PendingExecution { +func (pc *PendingCoordinator) TakeRebuild(replicaID string, targetLSN uint64) *PendingExecution { pc.mu.Lock() - pe, ok := pc.pending[volumeID] + pe, ok := pc.pending[replicaID] if ok { - delete(pc.pending, volumeID) + delete(pc.pending, replicaID) } pc.mu.Unlock() @@ -116,20 +117,21 @@ func (pc *PendingCoordinator) TakeRebuild(volumeID string, targetLSN uint64) *Pe return pe } -// Has returns true if a pending execution exists for the volume. -func (pc *PendingCoordinator) Has(volumeID string) bool { +// Has returns true if a pending execution exists for the replica target. +func (pc *PendingCoordinator) Has(replicaID string) bool { pc.mu.Lock() defer pc.mu.Unlock() - _, ok := pc.pending[volumeID] + _, ok := pc.pending[replicaID] return ok } -// Cancel explicitly cancels and removes the pending execution for a volume. -func (pc *PendingCoordinator) Cancel(volumeID, reason string) { +// Cancel explicitly cancels and removes the pending execution for a replica +// target. +func (pc *PendingCoordinator) Cancel(replicaID, reason string) { pc.mu.Lock() - pe, ok := pc.pending[volumeID] + pe, ok := pc.pending[replicaID] if ok { - delete(pc.pending, volumeID) + delete(pc.pending, replicaID) } pc.mu.Unlock() @@ -139,10 +141,10 @@ func (pc *PendingCoordinator) Cancel(volumeID, reason string) { } // Peek returns the pending execution without removing it. Returns nil if none. -func (pc *PendingCoordinator) Peek(volumeID string) *PendingExecution { +func (pc *PendingCoordinator) Peek(replicaID string) *PendingExecution { pc.mu.Lock() defer pc.mu.Unlock() - return pc.pending[volumeID] + return pc.pending[replicaID] } // CancelAll cancels and removes all pending executions. diff --git a/sw-block/engine/replication/runtime/pending_test.go b/sw-block/engine/replication/runtime/pending_test.go index 6f480ea47..b2385e190 100644 --- a/sw-block/engine/replication/runtime/pending_test.go +++ b/sw-block/engine/replication/runtime/pending_test.go @@ -144,6 +144,21 @@ func TestPendingCoordinator_StoreReplaces(t *testing.T) { } } +func TestPendingCoordinator_ReplicaScopedKeysDoNotCollide(t *testing.T) { + pc := NewPendingCoordinator(nil) + pc.Store("vol1/r1", &PendingExecution{VolumeID: "vol1", ReplicaID: "vol1/r1", CatchUpTarget: 10}) + pc.Store("vol1/r2", &PendingExecution{VolumeID: "vol1", ReplicaID: "vol1/r2", CatchUpTarget: 20}) + + pe1 := pc.TakeCatchUp("vol1/r1", 10) + if pe1 == nil || pe1.ReplicaID != "vol1/r1" { + t.Fatalf("pe1=%+v", pe1) + } + pe2 := pc.TakeCatchUp("vol1/r2", 20) + if pe2 == nil || pe2.ReplicaID != "vol1/r2" { + t.Fatalf("pe2=%+v", pe2) + } +} + func TestPendingCoordinator_TakeFromEmpty_ReturnsNil(t *testing.T) { pc := NewPendingCoordinator(nil) if pc.TakeCatchUp("vol1", 100) != nil { diff --git a/sw-block/engine/replication/state.go b/sw-block/engine/replication/state.go index 7904c2647..6354feac6 100644 --- a/sw-block/engine/replication/state.go +++ b/sw-block/engine/replication/state.go @@ -93,10 +93,9 @@ type commandState struct { ShipperConfigEpoch uint64 ShipperConfigReplicas []ReplicaAssignment RecoveryTaskEpoch uint64 - RecoveryTaskReplicaID string - RecoveryTaskKind SessionKind - CatchUpTargetLSN uint64 - RebuildTargetLSN uint64 + RecoveryTaskTargets map[string]SessionKind + CatchUpTargets map[string]uint64 + RebuildTargets map[string]uint64 InvalidationIssued bool InvalidationReason string } @@ -146,5 +145,23 @@ func (s *VolumeState) Snapshot() VolumeState { if s.commands.ShipperConfigReplicas != nil { out.commands.ShipperConfigReplicas = append([]ReplicaAssignment(nil), s.commands.ShipperConfigReplicas...) } + if s.commands.RecoveryTaskTargets != nil { + out.commands.RecoveryTaskTargets = make(map[string]SessionKind, len(s.commands.RecoveryTaskTargets)) + for replicaID, kind := range s.commands.RecoveryTaskTargets { + out.commands.RecoveryTaskTargets[replicaID] = kind + } + } + if s.commands.CatchUpTargets != nil { + out.commands.CatchUpTargets = make(map[string]uint64, len(s.commands.CatchUpTargets)) + for replicaID, target := range s.commands.CatchUpTargets { + out.commands.CatchUpTargets[replicaID] = target + } + } + if s.commands.RebuildTargets != nil { + out.commands.RebuildTargets = make(map[string]uint64, len(s.commands.RebuildTargets)) + for replicaID, target := range s.commands.RebuildTargets { + out.commands.RebuildTargets[replicaID] = target + } + } return out } diff --git a/weed/server/block_recovery.go b/weed/server/block_recovery.go index 6672e241f..23e5cbb12 100644 --- a/weed/server/block_recovery.go +++ b/weed/server/block_recovery.go @@ -303,7 +303,7 @@ func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID string, ass rm.executeLegacyCatchUp(ctx, rctx.volPath, replicaID, rctx.driver, plan, rctx.executor) return } - rm.coord.Store(rctx.volPath, &rt.PendingExecution{ + rm.coord.Store(replicaID, &rt.PendingExecution{ VolumeID: rctx.volPath, ReplicaID: replicaID, CatchUpTarget: plan.CatchUpTarget, @@ -312,8 +312,8 @@ func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID string, ass CatchUpIO: rctx.executor, }) bs.applyCoreEvent(engine.CatchUpPlanned{ID: rctx.volPath, TargetLSN: plan.CatchUpTarget}) - if rm.coord.Has(rctx.volPath) { - rm.coord.Cancel(rctx.volPath, "start_catchup_not_emitted") + if rm.coord.Has(replicaID) { + rm.coord.Cancel(replicaID, "start_catchup_not_emitted") return } case engine.OutcomeNeedsRebuild: @@ -361,32 +361,32 @@ func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID string, ass Plan: plan, RebuildIO: rctx.executor, } - rm.coord.Store(rctx.volPath, pe) + rm.coord.Store(replicaID, pe) if rm.OnPendingExecution != nil { rm.OnPendingExecution(rctx.volPath, pe) } bs.applyCoreEvent(engine.RebuildStarted{ID: rctx.volPath, TargetLSN: plan.RebuildTargetLSN}) - if rm.coord.Has(rctx.volPath) { - rm.coord.Cancel(rctx.volPath, "start_rebuild_not_emitted") + if rm.coord.Has(replicaID) { + rm.coord.Cancel(replicaID, "start_rebuild_not_emitted") } } // === Core-present pending execution (delegates to runtime.PendingCoordinator) === -func (rm *RecoveryManager) ExecutePendingCatchUp(volumeID string, targetLSN uint64) error { - pe := rm.coord.TakeCatchUp(volumeID, targetLSN) +func (rm *RecoveryManager) ExecutePendingCatchUp(replicaID string, targetLSN uint64) error { + pe := rm.coord.TakeCatchUp(replicaID, targetLSN) if pe == nil || pe.Driver == nil || pe.Plan == nil { return nil } - return rt.ExecuteCatchUpPlan(pe.Driver, pe.Plan, pe.CatchUpIO, volumeID, rm) + return rt.ExecuteCatchUpPlan(pe.Driver, pe.Plan, pe.CatchUpIO, pe.VolumeID, rm) } -func (rm *RecoveryManager) ExecutePendingRebuild(volumeID string, targetLSN uint64) error { - pe := rm.coord.TakeRebuild(volumeID, targetLSN) +func (rm *RecoveryManager) ExecutePendingRebuild(replicaID string, targetLSN uint64) error { + pe := rm.coord.TakeRebuild(replicaID, targetLSN) if pe == nil || pe.Driver == nil || pe.Plan == nil { return nil } - return rt.ExecuteRebuildPlan(pe.Driver, pe.Plan, pe.RebuildIO, volumeID, rm) + return rt.ExecuteRebuildPlan(pe.Driver, pe.Plan, pe.RebuildIO, pe.VolumeID, rm) } // RecoveryCallbacks implementation — host-side completion notifications. @@ -471,4 +471,3 @@ func (rm *RecoveryManager) volumePathForReplica(replicaID string) string { } // --- Bridge shims --- - diff --git a/weed/server/blockcmd/dispatch.go b/weed/server/blockcmd/dispatch.go index b39ced611..3057648f3 100644 --- a/weed/server/blockcmd/dispatch.go +++ b/weed/server/blockcmd/dispatch.go @@ -15,8 +15,8 @@ type Ops interface { ConfigureShipper(volumeID string, replicas []engine.ReplicaAssignment) (executed bool, shipperConnected bool, err error) StartRecoveryTask(replicaID string, assignment blockvol.BlockVolumeAssignment) (bool, error) InvalidateSession(volumeID, reason string) (bool, error) - StartCatchUp(volumeID string, targetLSN uint64) (bool, error) - StartRebuild(volumeID string, targetLSN uint64) (bool, error) + StartCatchUp(replicaID string, targetLSN uint64) (bool, error) + StartRebuild(replicaID string, targetLSN uint64) (bool, error) } // HostEffects applies server-adapter side effects after concrete command @@ -104,7 +104,10 @@ func (d *Dispatcher) Run(cmds []engine.Command, assignment *blockvol.BlockVolume d.effects.RecordCommand(v.VolumeID, "invalidate_session") } case engine.StartCatchUpCommand: - executed, err := d.ops.StartCatchUp(v.VolumeID, v.TargetLSN) + if v.ReplicaID == "" { + continue + } + executed, err := d.ops.StartCatchUp(v.ReplicaID, v.TargetLSN) if err != nil { return err } @@ -112,7 +115,10 @@ func (d *Dispatcher) Run(cmds []engine.Command, assignment *blockvol.BlockVolume d.effects.RecordCommand(v.VolumeID, "start_catchup") } case engine.StartRebuildCommand: - executed, err := d.ops.StartRebuild(v.VolumeID, v.TargetLSN) + if v.ReplicaID == "" { + continue + } + executed, err := d.ops.StartRebuild(v.ReplicaID, v.TargetLSN) if err != nil { return err } diff --git a/weed/server/blockcmd/dispatch_test.go b/weed/server/blockcmd/dispatch_test.go index 6ce3c43e8..bd5a6c750 100644 --- a/weed/server/blockcmd/dispatch_test.go +++ b/weed/server/blockcmd/dispatch_test.go @@ -54,18 +54,18 @@ func (f fakeOps) InvalidateSession(volumeID, reason string) (bool, error) { return f.invalidateSessionFn(volumeID, reason) } -func (f fakeOps) StartCatchUp(volumeID string, targetLSN uint64) (bool, error) { +func (f fakeOps) StartCatchUp(replicaID string, targetLSN uint64) (bool, error) { if f.startCatchUpFn == nil { return false, nil } - return f.startCatchUpFn(volumeID, targetLSN) + return f.startCatchUpFn(replicaID, targetLSN) } -func (f fakeOps) StartRebuild(volumeID string, targetLSN uint64) (bool, error) { +func (f fakeOps) StartRebuild(replicaID string, targetLSN uint64) (bool, error) { if f.startRebuildFn == nil { return false, nil } - return f.startRebuildFn(volumeID, targetLSN) + return f.startRebuildFn(replicaID, targetLSN) } type fakeEffects struct { @@ -168,17 +168,17 @@ func TestDispatcher_PublishProjectionUsesHostEffect(t *testing.T) { func TestDispatcher_StopsOnFirstError(t *testing.T) { effects := &fakeEffects{} d := NewDispatcher(fakeOps{ - startCatchUpFn: func(volumeID string, targetLSN uint64) (bool, error) { + startCatchUpFn: func(replicaID string, targetLSN uint64) (bool, error) { return false, errors.New("boom") }, - startRebuildFn: func(volumeID string, targetLSN uint64) (bool, error) { + startRebuildFn: func(replicaID string, targetLSN uint64) (bool, error) { t.Fatal("should not execute after first error") return false, nil }, }, effects) err := d.Run([]engine.Command{ - engine.StartCatchUpCommand{VolumeID: "vol1", TargetLSN: 10}, - engine.StartRebuildCommand{VolumeID: "vol1", TargetLSN: 20}, + engine.StartCatchUpCommand{VolumeID: "vol1", ReplicaID: "vol1/r1", TargetLSN: 10}, + engine.StartRebuildCommand{VolumeID: "vol1", ReplicaID: "vol1/r1", TargetLSN: 20}, }, nil) if err == nil { t.Fatal("expected error") @@ -189,11 +189,11 @@ type fakeRecoveryCoordinator struct { startedReplica string startedAssigns []blockvol.BlockVolumeAssignment catchUpCalls []struct { - volumeID string + replicaID string targetLSN uint64 } rebuildCalls []struct { - volumeID string + replicaID string targetLSN uint64 } } @@ -203,19 +203,19 @@ func (f *fakeRecoveryCoordinator) StartRecoveryTask(replicaID string, assignment f.startedAssigns = assignments } -func (f *fakeRecoveryCoordinator) ExecutePendingCatchUp(volumeID string, targetLSN uint64) error { +func (f *fakeRecoveryCoordinator) ExecutePendingCatchUp(replicaID string, targetLSN uint64) error { f.catchUpCalls = append(f.catchUpCalls, struct { - volumeID string + replicaID string targetLSN uint64 - }{volumeID: volumeID, targetLSN: targetLSN}) + }{replicaID: replicaID, targetLSN: targetLSN}) return nil } -func (f *fakeRecoveryCoordinator) ExecutePendingRebuild(volumeID string, targetLSN uint64) error { +func (f *fakeRecoveryCoordinator) ExecutePendingRebuild(replicaID string, targetLSN uint64) error { f.rebuildCalls = append(f.rebuildCalls, struct { - volumeID string + replicaID string targetLSN uint64 - }{volumeID: volumeID, targetLSN: targetLSN}) + }{replicaID: replicaID, targetLSN: targetLSN}) return nil } diff --git a/weed/server/blockcmd/service_ops.go b/weed/server/blockcmd/service_ops.go index 2025d3ed3..6e04ea285 100644 --- a/weed/server/blockcmd/service_ops.go +++ b/weed/server/blockcmd/service_ops.go @@ -16,8 +16,8 @@ type BackendOps interface { // RecoveryCoordinator is the runtime recovery surface used by command ops. type RecoveryCoordinator interface { StartRecoveryTask(replicaID string, assignments []blockvol.BlockVolumeAssignment) - ExecutePendingCatchUp(volumeID string, targetLSN uint64) error - ExecutePendingRebuild(volumeID string, targetLSN uint64) error + ExecutePendingCatchUp(replicaID string, targetLSN uint64) error + ExecutePendingRebuild(replicaID string, targetLSN uint64) error } // ProjectionReader provides access to current core publication state. @@ -103,21 +103,21 @@ func (ops *ServiceOps) InvalidateSession(volumeID, reason string) (bool, error) return true, nil } -func (ops *ServiceOps) StartCatchUp(volumeID string, targetLSN uint64) (bool, error) { +func (ops *ServiceOps) StartCatchUp(replicaID string, targetLSN uint64) (bool, error) { if ops == nil || ops.recovery == nil { return false, nil } - if err := ops.recovery.ExecutePendingCatchUp(volumeID, targetLSN); err != nil { + if err := ops.recovery.ExecutePendingCatchUp(replicaID, targetLSN); err != nil { return false, err } return true, nil } -func (ops *ServiceOps) StartRebuild(volumeID string, targetLSN uint64) (bool, error) { +func (ops *ServiceOps) StartRebuild(replicaID string, targetLSN uint64) (bool, error) { if ops == nil || ops.recovery == nil { return false, nil } - if err := ops.recovery.ExecutePendingRebuild(volumeID, targetLSN); err != nil { + if err := ops.recovery.ExecutePendingRebuild(replicaID, targetLSN); err != nil { return false, err } return true, nil