refactor: make bounded recovery command addressing replica-scoped

Replace the remaining volume-scoped recovery command and pending slot
with replica-scoped addressing on the bounded core-present path. This
preserves the current single-replica catch-up and rebuilding behavior
while removing the structural blocker for later multi-replica startup
ownership.

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

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

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

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
 }
 

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:

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)

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.

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 {

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
 }

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

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
 			}

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
 }
 

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