refactor: widen bounded multi-replica catch-up startup ownership

Emit one core-owned start_recovery_task per primary catch-up replica so the bounded multi-replica startup path no longer depends on a single-replica assumption.

Made-with: Cursor

sw-block/.private/phase/phase-16-log.md

@@ -965,3 +965,63 @@ Conclusion:
    completion after the first replica finishes
 2. this slice is still only an enabling aggregation step, not broad
    multi-replica startup ownership
+
+---
+
+#### `16I` Start Note Rev 1
+
+Date: 2026-04-04
+Scope: bounded multi-replica primary catch-up recovery-task startup ownership
+
+Why this slice exists:
+
+1. `16E` only bounded `start_recovery_task` ownership on the single-replica
+   primary catch-up path
+2. `16F-16H` made the downstream command / event / observation seams
+   replica-scoped enough to support a bounded widening
+3. the remaining gap is that primary assignment delivery still only starts one
+   bounded recovery task even when multiple catch-up replicas are present
+
+Chosen implementation rule:
+
+1. keep the widening limited to the core-present primary catch-up path
+2. emit one bounded `start_recovery_task` command per desired catch-up replica
+   on assignment delivery
+3. do not claim broad multi-replica recovery-loop closure beyond that startup
+   ownership seam
+
+---
+
+#### `16I` Delivery Note Rev 1
+
+Date: 2026-04-04
+Scope: bounded multi-replica primary catch-up recovery-task startup ownership
+
+What changed:
+
+1. `sw-block/engine/replication/engine.go`
+   - widened assignment-time `start_recovery_task` emission from a single
+     bounded replica to all desired bounded recovery replicas
+2. `weed/server/volume_server_block.go`
+   - primary assignment delivery now marks bounded catch-up recovery intent when
+     any stable replica set is present, including multi-replica assignments
+3. `sw-block/engine/replication/phase14_command_test.go`
+   - added focused proof that primary assignment emits one recovery-task command
+     per bounded replica
+4. `weed/server/volume_server_block_test.go`
+   - added focused proof that the host starts two bounded catch-up paths from
+     two emitted `start_recovery_task` commands on the multi-replica primary
+     path
+
+Proof / evidence:
+
+1. `go test ./...` from `sw-block/engine/replication`
+2. `go test ./weed/server -count=1 -timeout 120s -run "TestBlockService_(ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|PrimaryMultiReplica_UsesCoreStartRecoveryTaskPerReplica|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart)|DebugInfoForVolume|CollectBlockVolumeHeartbeat|ReadinessSnapshot|HeartbeatReplicaDegraded)"`
+3. result: `PASS`
+
+Conclusion:
+
+1. bounded multi-replica primary catch-up startup ownership is now
+   core-command-driven on the core-present path
+2. this slice still does not claim broad multi-replica recovery-loop closure or
+   broad failover/publication closure

sw-block/.private/phase/phase-16.md

@@ -324,6 +324,52 @@ Evidence:
 
 1. focused working-tree change after `16ba70f85`
 
+### `16I`: Multi-Replica Catch-Up Task Startup Ownership
+
+Goal:
+
+1. widen bounded catch-up recovery-task startup ownership from the single-replica
+   primary path to the bounded multi-replica primary path
+2. keep the slice limited to task startup ownership, not broad multi-replica
+   execution closure
+
+Acceptance object:
+
+1. on the core-present primary path with multiple replicas, the core emits one
+   bounded `start_recovery_task` command per catch-up replica
+2. the adapter starts those recovery goroutines because of the emitted commands,
+   not from orchestrator create/supersede results
+3. current single-replica and rebuilding proofs remain green
+4. this slice still does not yet claim broad multi-replica recovery-loop
+   closure
+
+Current chosen path:
+
+1. primary assignment with `len(replicas) > 1` now marks bounded
+   `RecoveryTarget=SessionCatchUp` in the core assignment event
+2. core assignment command emission widens `start_recovery_task` from one
+   replica to all bounded catch-up replicas on that path
+3. bounded multi-replica catch-up execution still closes through the already
+   replica-scoped command / observation seams from `16F-16H`
+
+Status:
+
+1. delivered
+
+Delivered result:
+
+1. primary assignment delivery now marks bounded catch-up startup intent for all
+   desired replicas, not only the single-replica path
+2. the core emits one bounded `start_recovery_task` command per catch-up
+   replica on that widened path
+3. the bounded adapter path now starts multi-replica primary catch-up recovery
+   work from those emitted commands and closes back through the existing
+   replica-scoped observation seams
+
+Evidence:
+
+1. focused working-tree change after `92c006eb2`
+
 ## Current Checkpoint Review Target
 
 The current review target is the current widened bounded runtime checkpoint
@@ -349,12 +395,14 @@ after `Phase 15` closeout:
    - replica-scoped recovery command addressing on those same bounded paths
    - conservative multi-replica catch-up observation aggregation on those same
      bounded paths
+   - bounded multi-replica catch-up recovery-task startup ownership on the
+     primary path
 
 This checkpoint is intentionally still bounded:
 
 1. broad recovery-loop closure is not yet claimed
 2. broad end-to-end failover/recovery/publication proof is not yet claimed
-3. multi-replica startup ownership is not yet claimed
+3. broad multi-replica startup ownership is not yet claimed
 4. launch / rollout readiness is not claimed
 
 ## Immediate Next Step
@@ -373,7 +421,7 @@ boundary:
    - rebuilding assignment no longer emits false `start_receiver`
 5. `16D` delivered:
    - rebuild recovery-task startup is core-command-driven
-6. `16E` current bounded refinement:
+6. `16E` delivered:
    - catch-up recovery-task startup is core-command-driven on the
      single-replica primary path
 7. `16F` delivered:
@@ -384,12 +432,13 @@ boundary:
 9. `16H` delivered:
    - multi-replica catch-up observation is aggregated conservatively at the
      volume projection layer
+10. `16I` delivered:
+   - multi-replica primary catch-up startup ownership is core-command-driven
 
 After this checkpoint:
 
 1. keep `legacy P4` only as a compatibility guard
-2. the next bounded semantic/runtime decision is whether to widen startup
-   ownership beyond the single-replica catch-up path now that multi-replica
-   catch-up aggregation no longer overclaims completion
+2. identify the next bounded runtime gap after multi-replica startup ownership,
+   most likely around broader recovery-loop closure rather than assignment entry
 3. do not yet claim full recovery-loop closure
 4. do not broaden into launch claims

sw-block/engine/replication/engine.go

@@ -385,7 +385,10 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) []
 		st.commands.ShipperConfigEpoch = st.Epoch
 		st.commands.ShipperConfigReplicas = append([]ReplicaAssignment(nil), st.DesiredReplicas...)
 	}
-	if replicaID, ok := st.recoveryCommandReplicaID(); ok && st.shouldStartRecoveryTask(replicaID) {
+	for _, replicaID := range st.recoveryTaskReplicaIDs() {
+		if !st.shouldStartRecoveryTask(replicaID) {
+			continue
+		}
 		cmds = append(cmds, StartRecoveryTaskCommand{
 			VolumeID:  st.VolumeID,
 			ReplicaID: replicaID,
@@ -498,6 +501,20 @@ func (st *VolumeState) recoveryCommandReplicaID() (string, bool) {
 	return st.DesiredReplicas[0].ReplicaID, true
 }
 
+func (st *VolumeState) recoveryTaskReplicaIDs() []string {
+	if st.recoveryTarget == "" || len(st.DesiredReplicas) == 0 {
+		return nil
+	}
+	replicaIDs := make([]string, 0, len(st.DesiredReplicas))
+	for _, replica := range st.DesiredReplicas {
+		if replica.ReplicaID == "" {
+			continue
+		}
+		replicaIDs = append(replicaIDs, replica.ReplicaID)
+	}
+	return replicaIDs
+}
+
 func (st *VolumeState) recoveryCommandReplicaIDFromEvent(replicaID string) (string, bool) {
 	if replicaID != "" {
 		for _, replica := range st.DesiredReplicas {

sw-block/engine/replication/phase14_command_test.go

@@ -29,6 +29,35 @@ func TestPhase14_CommandSequence_PrimaryAssignmentIsBounded(t *testing.T) {
 	assertCommandNames(t, result.Commands, nil)
 }
 
+func TestPhase14_CommandSequence_PrimaryMultiReplicaAssignmentStartsRecoveryPerReplica(t *testing.T) {
+	core := NewCoreEngine()
+	ev := AssignmentDelivered{
+		ID:             "vol-cmd-primary-multi",
+		Epoch:          1,
+		Role:           RolePrimary,
+		RecoveryTarget: SessionCatchUp,
+		Replicas: []ReplicaAssignment{
+			{ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.21:9333", CtrlAddr: "10.0.0.21:9334", Version: 1}},
+			{ReplicaID: "replica-2", Endpoint: Endpoint{DataAddr: "10.0.0.22:9333", CtrlAddr: "10.0.0.22:9334", Version: 1}},
+		},
+	}
+
+	result := core.ApplyEvent(ev)
+	assertCommandNames(t, result.Commands, []string{
+		"apply_role",
+		"configure_shipper",
+		"start_recovery_task",
+		"start_recovery_task",
+		"publish_projection",
+	})
+	if got := recoveryTaskReplicaIDs(result.Commands); !reflect.DeepEqual(got, []string{"replica-1", "replica-2"}) {
+		t.Fatalf("recovery task replicas=%v", got)
+	}
+
+	result = core.ApplyEvent(ev)
+	assertCommandNames(t, result.Commands, nil)
+}
+
 func TestPhase14_CommandSequence_ReplicaAssignmentIsBounded(t *testing.T) {
 	core := NewCoreEngine()
 	ev := AssignmentDelivered{
@@ -270,3 +299,15 @@ func assertCommandNames(t *testing.T, cmds []Command, want []string) {
 		t.Fatalf("commands=%v, want %v", got, want)
 	}
 }
+
+func recoveryTaskReplicaIDs(cmds []Command) []string {
+	var replicaIDs []string
+	for _, cmd := range cmds {
+		start, ok := cmd.(StartRecoveryTaskCommand)
+		if !ok {
+			continue
+		}
+		replicaIDs = append(replicaIDs, start.ReplicaID)
+	}
+	return replicaIDs
+}

weed/server/volume_server_block.go

@@ -710,7 +710,7 @@ func (bs *BlockService) coreAssignmentEvent(a blockvol.BlockVolumeAssignment) (e
 					CtrlAddr: ra.CtrlAddr,
 				}))
 			}
-			if len(ev.Replicas) == 1 {
+			if len(ev.Replicas) > 0 {
 				ev.RecoveryTarget = engine.SessionCatchUp
 			}
 		} else if a.ReplicaServerID != "" && a.ReplicaDataAddr != "" {

weed/server/volume_server_block_test.go

@@ -493,6 +493,96 @@ func TestBlockService_ApplyAssignments_PrimaryRole_UsesCoreStartRecoveryTaskForC
 	}
 }
 
+func TestBlockService_ApplyAssignments_PrimaryMultiReplica_UsesCoreStartRecoveryTaskPerReplica(t *testing.T) {
+	bs := newTestBlockServiceDirect(t)
+	bs.v2Bridge = newTestControlBridge()
+	bs.v2Orchestrator = newTestOrchestrator()
+	bs.v2Recovery = NewRecoveryManager(bs)
+	defer bs.v2Recovery.Shutdown()
+
+	path := createTestVolDirect(t, bs, "vol-core-cmd-catchup-start-multi")
+	if err := bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
+		for i := 0; i < 5; i++ {
+			if err := vol.WriteLBA(uint64(i), make([]byte, 4096)); err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	bs.v2Recovery.OnPendingExecution = func(volumeID string, pending *rt.PendingExecution) {
+		if volumeID == path && pending != nil && pending.Plan != nil {
+			pending.CatchUpIO = fakeCatchUpIO{transferredTo: pending.Plan.CatchUpTarget}
+		}
+	}
+
+	errs := bs.ApplyAssignments([]blockvol.BlockVolumeAssignment{{
+		Path:       path,
+		Epoch:      1,
+		Role:       blockvol.RoleToWire(blockvol.RolePrimary),
+		LeaseTtlMs: 30000,
+		ReplicaAddrs: []blockvol.ReplicaAddr{
+			{ServerID: "vs-2", DataAddr: "10.0.0.2:4260", CtrlAddr: "10.0.0.2:4261"},
+			{ServerID: "vs-3", DataAddr: "10.0.0.3:4260", CtrlAddr: "10.0.0.3:4261"},
+		},
+	}})
+	if len(errs) != 1 || errs[0] != nil {
+		t.Fatalf("apply errs=%v", errs)
+	}
+
+	var (
+		proj    engine.PublicationProjection
+		ok      bool
+		sender2 *engine.Sender
+		sender3 *engine.Sender
+	)
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		proj, ok = bs.CoreProjection(path)
+		sender2 = bs.v2Orchestrator.Registry.Sender(path + "/vs-2")
+		sender3 = bs.v2Orchestrator.Registry.Sender(path + "/vs-3")
+		if ok &&
+			proj.Recovery.Phase == engine.RecoveryIdle &&
+			sender2 != nil && sender2.State() == engine.StateInSync &&
+			sender3 != nil && sender3.State() == engine.StateInSync {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	if !ok {
+		t.Fatal("expected core projection")
+	}
+	if proj.Recovery.Phase != engine.RecoveryIdle {
+		t.Fatalf("recovery_phase=%s", proj.Recovery.Phase)
+	}
+	if proj.Boundary.DurableLSN == 0 {
+		t.Fatalf("durable_lsn=%d", proj.Boundary.DurableLSN)
+	}
+	if sender2 == nil || sender3 == nil {
+		t.Fatalf("senders not found: vs-2=%v vs-3=%v", sender2 != nil, sender3 != nil)
+	}
+	if sender2.State() != engine.StateInSync || sender3.State() != engine.StateInSync {
+		t.Fatalf("sender states: vs-2=%s vs-3=%s", sender2.State(), sender3.State())
+	}
+
+	cmds := bs.ExecutedCoreCommands(path)
+	if got := countCommandName(cmds, "apply_role"); got != 1 {
+		t.Fatalf("apply_role count=%d cmds=%v", got, cmds)
+	}
+	if got := countCommandName(cmds, "configure_shipper"); got != 1 {
+		t.Fatalf("configure_shipper count=%d cmds=%v", got, cmds)
+	}
+	if got := countCommandName(cmds, "start_recovery_task"); got != 2 {
+		t.Fatalf("start_recovery_task count=%d cmds=%v", got, cmds)
+	}
+	if got := countCommandName(cmds, "start_catchup"); got != 2 {
+		t.Fatalf("start_catchup count=%d cmds=%v", got, cmds)
+	}
+}
+
 func TestBlockService_ApplyAssignments_RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart(t *testing.T) {
 	bs := newTestBlockServiceDirect(t)
 	bs.v2Bridge = newTestControlBridge()
@@ -1068,3 +1158,13 @@ func TestBlockService_ProcessAssignment_InvalidTransition(t *testing.T) {
 		t.Fatalf("epoch should still be 5, got %d", s.Epoch)
 	}
 }
+
+func countCommandName(cmds []string, want string) int {
+	count := 0
+	for _, cmd := range cmds {
+		if cmd == want {
+			count++
+		}
+	}
+	return count
+}