diff --git a/sw-block/.private/phase/phase-16-log.md b/sw-block/.private/phase/phase-16-log.md index 0f7babc37..ac617403d 100644 --- a/sw-block/.private/phase/phase-16-log.md +++ b/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 diff --git a/sw-block/.private/phase/phase-16.md b/sw-block/.private/phase/phase-16.md index 2e27e67a8..a12f1775d 100644 --- a/sw-block/.private/phase/phase-16.md +++ b/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 diff --git a/sw-block/engine/replication/engine.go b/sw-block/engine/replication/engine.go index ddb1b97e5..4b2eff665 100644 --- a/sw-block/engine/replication/engine.go +++ b/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 { diff --git a/sw-block/engine/replication/phase14_command_test.go b/sw-block/engine/replication/phase14_command_test.go index 3bcd669b5..7a8f1d776 100644 --- a/sw-block/engine/replication/phase14_command_test.go +++ b/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 +} diff --git a/weed/server/volume_server_block.go b/weed/server/volume_server_block.go index 9e59b6802..33ce9166d 100644 --- a/weed/server/volume_server_block.go +++ b/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 != "" { diff --git a/weed/server/volume_server_block_test.go b/weed/server/volume_server_block_test.go index c47c25c93..e19451839 100644 --- a/weed/server/volume_server_block_test.go +++ b/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 +}