diff --git a/sw-block/.private/phase/phase-16-checkpoint-review.md b/sw-block/.private/phase/phase-16-checkpoint-review.md index 7e350c063..8f1d09a38 100644 --- a/sw-block/.private/phase/phase-16-checkpoint-review.md +++ b/sw-block/.private/phase/phase-16-checkpoint-review.md @@ -42,12 +42,14 @@ Bounded live recovery closure now covers: 1. live recovery observations return into the core on catch-up / rebuild entry/exit points 2. bounded catch-up execution runs from `StartCatchUpCommand` -3. old no-core path compatibility remains preserved +3. rebuild execution ownership is not part of the accepted checkpoint +4. old no-core path compatibility remains preserved Expected judgment: 1. this is a real bounded runtime closure step -2. it is not yet full recovery-loop ownership +2. rebuild is still observation-only / next candidate on this path +3. it is not yet full recovery-loop ownership ## What Is Explicitly Out Of Scope diff --git a/sw-block/.private/phase/phase-16-log.md b/sw-block/.private/phase/phase-16-log.md index 9637c925f..cd2f53ce2 100644 --- a/sw-block/.private/phase/phase-16-log.md +++ b/sw-block/.private/phase/phase-16-log.md @@ -391,6 +391,79 @@ Constraint / overclaim / proof review: --- +#### `16B` Delivery Note Rev 3 + +Date: 2026-04-04 +Scope: bounded rebuild execution ownership on the same recovery path + +What changed: + +1. pending recovery execution storage now carries engine-level catch-up / rebuild + I/O interfaces instead of a concrete `v2bridge.Executor` only +2. `RecoveryManager.runRebuild()` now follows the same bounded pattern as + catch-up on the core-present path: + - plan rebuild + - cache pending execution + - emit `RebuildStarted` + - execute only if `StartRebuildCommand` consumes the pending plan +3. if no fresh `StartRebuildCommand` is emitted, pending rebuild execution is + cancelled fail-closed instead of executing implicitly +4. added focused rebuild proofs: + - live `runRebuild()` caches pending rebuild, emits `RebuildStarted`, + consumes `StartRebuildCommand`, and closes completion back into projection + - no fresh command means no implicit rebuild execution + +Files changed: + +1. `weed/server/block_recovery.go` + - split pending execution I/O into `engine.CatchUpIO` / `engine.RebuildIO` + - reused shared execution helpers for rebuild ownership closure +2. `weed/server/block_recovery_test.go` + - added bounded live-path rebuild ownership proof + - added bounded rebuild fail-closed proof + +Bounded contract: + +`16B Rev 3` accepts only this: + +1. bounded rebuild execution now runs from `StartRebuildCommand` +2. the corresponding rebuild completion observation still closes back into the + core +3. if no fresh rebuild command is emitted, the pending rebuild plan does not run + implicitly + +It does not yet accept: + +1. full recovery-loop closure +2. broad multi-replica rebuild ownership +3. launch / rollout readiness + +Validation: + +1. `go test ./weed/server -run "TestP(4_LivePath_RealVol_ReachesPlan|16B_(Run(CatchUp|Rebuild)_|StartRebuildCommand_))"` +2. `go test ./weed/server -run "Test(P4_|P16B_|BlockService_(ApplyAssignments|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` + +Constraint / overclaim / proof review: + +1. semantic constraint satisfied + - `Phase 16` now has bounded command ownership for both catch-up and rebuild + execution on the selected recovery path +2. overclaim avoided + - this revision proves bounded rebuild execution ownership, not full + recovery-loop closure or broad rebuild runtime closure +3. proof preserved + - accepted `P4` no-core recovery tests still pass, and the existing + `15B/16A/16B` consume-chain proofs remain green + +Review status: + +1. this is a new working-state delivery beyond the previously reviewed + catch-up-only checkpoint +2. external review has not yet been re-run for this widened `16B` state + +--- + #### `16` Checkpoint Review Note Date: 2026-04-04 @@ -403,6 +476,8 @@ Current checkpoint judgment: 3. `16B` has one accepted current closure: - live recovery observations close back into the core - bounded catch-up execution is core-command-driven +4. rebuild observation ingress exists, but rebuild execution ownership is not + part of the accepted checkpoint Recommended review target: @@ -411,6 +486,7 @@ Recommended review target: - `16A delivered` - `16B current bounded closure` 2. do not review it as: + - rebuild execution ownership - full rebuild execution ownership - full recovery-loop closure - launch / rollout readiness @@ -427,3 +503,45 @@ Suggested commit boundary if review is accepted: 8. `weed/server/master_block_observability_test.go` 9. `weed/server/block_recovery.go` 10. `weed/server/block_recovery_test.go` + +--- + +#### `16B` Post-Review Fix Note + +Date: 2026-04-04 + +Context: + +1. prior `manager` review accepted widened `16B Rev 3` with two minor fixes +2. required fixes were: + - add one positive live-path rebuild ownership proof + - tighten `Phase 16` wording from `first bounded` to `current widened bounded` + +What was changed: + +1. `weed/server/block_recovery.go` + - added a minimal test hook so focused tests can override freshly cached + pending rebuild I/O without changing production ownership semantics +2. `weed/server/block_recovery_test.go` + - replaced the seeded positive rebuild proof with + `TestP16B_RunRebuild_UsesCoreStartRebuildCommandOnLivePath` + - the test now proves the full live chain: + `runRebuild()` -> pending rebuild cached -> `RebuildStarted` -> + `StartRebuildCommand` -> adapter execution -> rebuild completion +3. `sw-block/.private/phase/phase-16.md` + - tightened wording to `current widened bounded runtime checkpoint` +4. `sw-block/.private/phase/phase-16-rev3-review.md` + - updated evidence summary to cite the live-path rebuild proof +5. `sw-block/.private/phase/phase-16-rev3-manager-rereview.md` + - added a bounded delta note for `manager` re-review only + +Validation: + +1. focused recovery suite: `PASS` +2. combined `P4/15B/16A/16B` proof suite: `PASS` +3. lints: clean + +Review intent: + +1. this note does not broaden `16B` +2. it only closes the two minor gaps identified by `manager` diff --git a/sw-block/.private/phase/phase-16-rev3-manager-rereview.md b/sw-block/.private/phase/phase-16-rev3-manager-rereview.md new file mode 100644 index 000000000..6eaefd17b --- /dev/null +++ b/sw-block/.private/phase/phase-16-rev3-manager-rereview.md @@ -0,0 +1,82 @@ +# Phase 16 Rev 3 Manager Re-review + +Date: 2026-04-04 +Status: ready for re-review + +## Purpose + +This note is only for the delta since the prior `manager` review of widened +`16B Rev 3`. + +Please review only whether the two requested fixes are now satisfied: + +1. positive live-path rebuild ownership proof now exists +2. `Phase 16` wording is tightened from `first bounded` to `current widened bounded` + +## Delta Since Prior Review + +### 1. Positive live-path rebuild ownership proof added + +Previous gap: + +1. positive rebuild proof seeded pending execution directly +2. that proved command consumption, but not the full live `runRebuild()` chain + +Current proof: + +1. `weed/server/block_recovery_test.go` +2. `TestP16B_RunRebuild_UsesCoreStartRebuildCommandOnLivePath` +3. proved chain: + - `runRebuild()` + - cache pending rebuild + - emit `RebuildStarted` + - core emits `StartRebuildCommand` + - adapter consumes pending rebuild + - rebuild completion observation returns into core + +Observed outcomes asserted by the test: + +1. executed command list ends with `start_rebuild` +2. cached projection returns to `RecoveryIdle` +3. sender returns to `StateInSync` + +This closes the exact positive-path gap identified in the previous review. + +### 2. Wording hygiene tightened + +Updated file: + +1. `sw-block/.private/phase/phase-16.md` + +Updated wording: + +1. from: `the first bounded integrated runtime checkpoint after Phase 15 closeout` +2. to: `the current widened bounded runtime checkpoint after Phase 15 closeout` + +This keeps the wording aligned with the real review object. + +## Validation + +1. `go test ./weed/server -run "TestP(4_LivePath_RealVol_ReachesPlan|16B_(Run(CatchUp|Rebuild)_|StartRebuildCommand_))"` +2. `go test ./weed/server -run "Test(P4_|P16B_|BlockService_(ApplyAssignments|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` + +## Bounded Claim Unchanged + +This re-review still asks you to review only: + +1. bounded recovery execution ownership on catch-up and rebuild +2. not full recovery-loop closure +3. not broad end-to-end failover/recovery/publication closure +4. not multi-replica rebuild ownership +5. not launch / rollout readiness + +## Requested Output + +Please reply with one of: + +1. `ACCEPT` +2. `ACCEPT WITH MINOR FIXES` +3. `REJECT` + +If not `ACCEPT`, please keep findings bounded to this delta only. diff --git a/sw-block/.private/phase/phase-16-rev3-review.md b/sw-block/.private/phase/phase-16-rev3-review.md new file mode 100644 index 000000000..ed8a9558d --- /dev/null +++ b/sw-block/.private/phase/phase-16-rev3-review.md @@ -0,0 +1,165 @@ +# Phase 16 Rev 3 Review + +Date: 2026-04-04 +Status: ready for review + +## Review Object + +Review the current widened `Phase 16` working state as: + +1. `Phase 15` delivered +2. `16A` delivered +3. `16B` bounded recovery execution ownership: + - live recovery observations return into the core + - bounded `start_catchup` execution is core-command-driven + - bounded `start_rebuild` execution is core-command-driven + +This is a new review object beyond the previously accepted catch-up-only +checkpoint. + +## What Is In Scope + +### `Phase 15` closeout + +1. bounded surface/store/outward consume-chain rebinding to core-owned truth +2. cluster-level status surface extraction and closure proof preserved + +### `16A` delivered + +Bounded command-driven adapter ownership covers: + +1. `apply_role` +2. `start_receiver` +3. `configure_shipper` +4. `invalidate_session` + +Expected judgment: + +1. these paths execute because the core emitted commands +2. the adapter remains executor, not semantic owner + +### `16B` widened bounded closure + +Bounded live recovery closure now covers: + +1. live recovery observations return into the core on catch-up / rebuild + entry/exit points +2. bounded `start_catchup` execution runs from `StartCatchUpCommand` +3. bounded `start_rebuild` execution runs from `StartRebuildCommand` +4. if no fresh rebuild command is emitted, pending rebuild does not run + implicitly +5. old no-core compatibility remains preserved + +Expected judgment: + +1. this is still a bounded runtime-ownership step +2. catch-up and rebuild execution ownership are both now in scope +3. it is still not full recovery-loop closure + +## What Is Explicitly Out Of Scope + +Do NOT review this widened checkpoint as claiming: + +1. full recovery-loop closure +2. broad end-to-end failover/recovery/publication closure +3. broad multi-replica rebuild ownership +4. launch / rollout readiness + +## Primary Files + +Phase tracking: + +1. `sw-block/.private/phase/phase-15.md` +2. `sw-block/.private/phase/phase-15-log.md` +3. `sw-block/.private/phase/phase-16.md` +4. `sw-block/.private/phase/phase-16-log.md` + +Integrated runtime code: + +1. `weed/server/volume_server_block.go` +2. `weed/server/volume_server_block_test.go` +3. `weed/server/master_server_handlers_block.go` +4. `weed/server/master_block_observability_test.go` +5. `weed/server/block_recovery.go` +6. `weed/server/block_recovery_test.go` + +## Evidence Summary + +### Surface/store closure preserved + +Focused proof suite: + +1. `go test ./weed/server -run "Test(BlockService_(ApplyAssignments|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)"` +2. result: `PASS` + +### Recovery ownership closure + +Focused recovery proof suite: + +1. `go test ./weed/server -run "TestP(4_LivePath_RealVol_ReachesPlan|16B_(Run(CatchUp|Rebuild)_|StartRebuildCommand_))"` +2. result: `PASS` + +Key new rebuild proofs: + +1. `TestP16B_RunRebuild_UsesCoreStartRebuildCommandOnLivePath` + - proves the live chain: + `runRebuild()` -> cache pending rebuild -> emit `RebuildStarted` -> + `StartRebuildCommand` -> adapter consumption -> rebuild completion + - proves rebuild completion observation closes back into core projection +2. `TestP16B_RunRebuild_FailClosedWithoutFreshStartRebuildCommand` + - proves pending rebuild does not execute implicitly without a fresh command + +## Review Questions + +### For `sw` + +Please check implementation correctness and commit-readiness: + +1. Is the widened `16B` boundary still coherent as one bounded checkpoint? +2. Is the rebuild ownership implementation internally consistent with the + existing catch-up ownership pattern? +3. Are there any cleanup/refactor issues that should be fixed before commit, + without broadening scope? + +Suggested commit boundary if accepted: + +1. `sw-block/.private/phase/phase-16.md` +2. `sw-block/.private/phase/phase-16-log.md` +3. `sw-block/.private/phase/phase-16-rev3-review.md` +4. `weed/server/block_recovery.go` +5. `weed/server/block_recovery_test.go` + +### For `tester` + +Please challenge the proof posture: + +1. Does `16B Rev 3` now prove the positive live `start_rebuild` ownership chain, + not just structural command plumbing? +2. Is the fail-closed proof strong enough to show pending rebuild does not run + implicitly? +3. Are there any remaining surfaces where rebuild truth could still diverge + from the core on the bounded path? +4. Are these new tests proving semantic claim rather than implementation shape? + +### For `manager` + +Please challenge boundaries and overclaim: + +1. Does widening `16B` from catch-up-only to catch-up+rebuild still keep the + slice bounded? +2. Is the wording still disciplined that this is not full recovery-loop closure? +3. Does the updated `Phase 16` wording clearly separate: + - bounded recovery execution ownership + - broader end-to-end scenario closure +4. Is this a reasonable next stage checkpoint? + +## Requested Output Shape + +Please reply with one of: + +1. `ACCEPT` +2. `ACCEPT WITH MINOR FIXES` +3. `REJECT` + +If not `ACCEPT`, list findings ordered by severity and keep them bounded to +this widened `16B` claim set. diff --git a/sw-block/.private/phase/phase-16.md b/sw-block/.private/phase/phase-16.md index bced64fff..a4b3bc29b 100644 --- a/sw-block/.private/phase/phase-16.md +++ b/sw-block/.private/phase/phase-16.md @@ -93,8 +93,8 @@ Current chosen path: 1. live recovery observations now return into the core on catch-up and rebuild entry/exit points 2. bounded catch-up execution now runs from `StartCatchUpCommand` -3. rebuild execution is the next likely runtime-driving candidate on the same - path +3. bounded rebuild execution now runs from `StartRebuildCommand` +4. full recovery-loop closure remains outside the current bounded path Status: @@ -102,7 +102,7 @@ Status: ## Current Checkpoint Review Target -The current review target is the first bounded integrated runtime checkpoint +The current review target is the current widened bounded runtime checkpoint after `Phase 15` closeout: 1. `Phase 15` delivered: @@ -113,27 +113,32 @@ after `Phase 15` closeout: - `start_receiver` - `configure_shipper` - `invalidate_session` -3. `16B` active with accepted current closure: +3. previously reviewed `16B` closure: - live recovery observations return into the core - bounded catch-up execution runs from `StartCatchUpCommand` +4. current working state extends that bounded path with: + - bounded rebuild execution from `StartRebuildCommand` This checkpoint is intentionally still bounded: -1. `start_rebuild` execution ownership is not yet in scope -2. broad recovery-loop closure is not yet claimed +1. broad recovery-loop closure is not yet claimed +2. broad end-to-end failover/recovery/publication proof is not yet claimed 3. launch / rollout readiness is not claimed ## Immediate Next Step -Start `16A` with the narrowest runtime-driving execution decision still owned by -adapter branching: +The current checkpoint is now good enough to take as a stage/commit boundary: -1. choose one path where the core already emits a bounded command -2. let the adapter execute from that command instead of from implicit local - control flow -3. keep `blockvol` as execution backend and treat its local state only as - observation input +1. `Phase 15` delivered +2. `16A` delivered +3. `16B` bounded recovery execution ownership: + - live recovery observations close back into the core + - bounded catch-up execution is core-command-driven + - bounded rebuild execution is core-command-driven -The next likely engineering target is the matching rebuild execution path, -unless the current `Phase 15 + 16A + 16B` boundary is accepted as the next -commit checkpoint first. +After that checkpoint, decide whether `Phase 16` needs one stricter end-to-end +recovery/publication scenario before moving beyond the phase: + +1. keep the current bounded ownership claim narrow +2. only add broader scenario proof if it materially strengthens the accepted bar +3. do not broaden this into launch claims diff --git a/weed/server/block_recovery.go b/weed/server/block_recovery.go index 59789f3eb..4761075b8 100644 --- a/weed/server/block_recovery.go +++ b/weed/server/block_recovery.go @@ -39,6 +39,10 @@ type RecoveryManager struct { // TestHook: if set, called before execution starts. Tests use this // to hold the goroutine alive for serialized-replacement proofs. OnBeforeExecute func(replicaID string) + + // TestHook: if set, may adjust a freshly cached pending execution before + // the core event is emitted. Used only by focused ownership tests. + OnPendingExecution func(volumeID string, pending *pendingRecoveryExecution) } type pendingRecoveryExecution struct { @@ -46,7 +50,8 @@ type pendingRecoveryExecution struct { replicaID string driver *engine.RecoveryDriver plan *engine.RecoveryPlan - io *v2bridge.Executor + catchUpIO engine.CatchUpIO + rebuildIO engine.RebuildIO } func NewRecoveryManager(bs *BlockService) *RecoveryManager { @@ -282,7 +287,7 @@ func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID, rebuildAdd replicaID: replicaID, driver: driver, plan: plan, - io: executor, + catchUpIO: executor, }) bs.applyCoreEvent(engine.CatchUpPlanned{ID: volPath, TargetLSN: plan.CatchUpTarget}) if rm.hasPendingExecution(volPath) { @@ -355,8 +360,13 @@ func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID, rebuildAdd replicaID: replicaID, driver: driver, plan: plan, - io: executor, + rebuildIO: executor, }) + if rm.OnPendingExecution != nil { + if pending, ok := rm.peekPendingExecution(volPath); ok { + rm.OnPendingExecution(volPath, pending) + } + } bs.applyCoreEvent(engine.RebuildStarted{ID: volPath, TargetLSN: plan.RebuildTargetLSN}) if rm.hasPendingExecution(volPath) { rm.cancelPendingExecution(volPath, "start_rebuild_not_emitted") @@ -382,6 +392,13 @@ func (rm *RecoveryManager) takePendingExecution(volumeID string) (*pendingRecove return pending, ok } +func (rm *RecoveryManager) peekPendingExecution(volumeID string) (*pendingRecoveryExecution, bool) { + rm.mu.Lock() + defer rm.mu.Unlock() + pending, ok := rm.pending[volumeID] + return pending, ok +} + func (rm *RecoveryManager) hasPendingExecution(volumeID string) bool { rm.mu.Lock() defer rm.mu.Unlock() @@ -406,7 +423,7 @@ func (rm *RecoveryManager) ExecutePendingCatchUp(volumeID string, targetLSN uint pending.driver.CancelPlan(pending.plan, "start_catchup_target_mismatch") return nil } - return rm.executeCatchUpPlan(volumeID, pending.replicaID, pending.driver, pending.plan, pending.io) + return rm.executeCatchUpPlan(volumeID, pending.replicaID, pending.driver, pending.plan, pending.catchUpIO) } func (rm *RecoveryManager) ExecutePendingRebuild(volumeID string, targetLSN uint64) error { @@ -418,10 +435,10 @@ func (rm *RecoveryManager) ExecutePendingRebuild(volumeID string, targetLSN uint pending.driver.CancelPlan(pending.plan, "start_rebuild_target_mismatch") return nil } - return rm.executeRebuildPlan(volumeID, pending.replicaID, pending.driver, pending.plan, pending.io) + return rm.executeRebuildPlan(volumeID, pending.replicaID, pending.driver, pending.plan, pending.rebuildIO) } -func (rm *RecoveryManager) executeCatchUpPlan(volumeID, replicaID string, driver *engine.RecoveryDriver, plan *engine.RecoveryPlan, io *v2bridge.Executor) error { +func (rm *RecoveryManager) executeCatchUpPlan(volumeID, replicaID string, driver *engine.RecoveryDriver, plan *engine.RecoveryPlan, io engine.CatchUpIO) error { exec := engine.NewCatchUpExecutor(driver, plan) exec.IO = io if err := exec.Execute(nil, 0); err != nil { @@ -438,7 +455,7 @@ func (rm *RecoveryManager) executeCatchUpPlan(volumeID, replicaID string, driver return nil } -func (rm *RecoveryManager) executeRebuildPlan(volumeID, replicaID string, driver *engine.RecoveryDriver, plan *engine.RecoveryPlan, io *v2bridge.Executor) error { +func (rm *RecoveryManager) executeRebuildPlan(volumeID, replicaID string, driver *engine.RecoveryDriver, plan *engine.RecoveryPlan, io engine.RebuildIO) error { exec := engine.NewRebuildExecutor(driver, plan) exec.IO = io if err := exec.Execute(); err != nil { diff --git a/weed/server/block_recovery_test.go b/weed/server/block_recovery_test.go index 4ae8d2b8d..6d5d71ea4 100644 --- a/weed/server/block_recovery_test.go +++ b/weed/server/block_recovery_test.go @@ -2,7 +2,9 @@ package weed_server import ( "context" + "fmt" "path/filepath" + "reflect" "testing" "time" @@ -102,6 +104,25 @@ func createTestBlockServiceWithVolCoreNoRecovery(t *testing.T) (*BlockService, s return bs, volPath } +type fakeRebuildIO struct { + achievedLSN uint64 +} + +func (f fakeRebuildIO) TransferFullBase(committedLSN uint64) (uint64, error) { + if f.achievedLSN > 0 { + return f.achievedLSN, nil + } + return committedLSN, nil +} + +func (f fakeRebuildIO) TransferSnapshot(snapshotLSN uint64) error { + return nil +} + +func (f fakeRebuildIO) StreamWALEntries(startExclusive, endInclusive uint64) (uint64, error) { + return endInclusive, nil +} + // --- Live-path with real vol: reaches planning --- func TestP4_LivePath_RealVol_ReachesPlan(t *testing.T) { @@ -274,6 +295,141 @@ func TestP16B_RunCatchUp_EscalatesNeedsRebuildIntoCoreProjection(t *testing.T) { } } +func TestP16B_RunRebuild_UsesCoreStartRebuildCommandOnLivePath(t *testing.T) { + bs, volPath := createTestBlockServiceWithVolCoreNoRecovery(t) + + if err := bs.blockStore.WithVolume(volPath, 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 vol.ForceFlush() + }); err != nil { + t.Fatalf("write+flush: %v", err) + } + + bs.ProcessAssignments([]blockvol.BlockVolumeAssignment{{ + Path: volPath, + Epoch: 1, + Role: uint32(blockvol.RolePrimary), + ReplicaServerID: "vs2", + ReplicaDataAddr: "10.0.0.2:9333", + ReplicaCtrlAddr: "10.0.0.2:9334", + }}) + + replicaID := volPath + "/vs2" + bs.v2Orchestrator.ProcessAssignment(engine.AssignmentIntent{ + Replicas: []engine.ReplicaAssignment{{ + ReplicaID: replicaID, + Endpoint: engine.Endpoint{DataAddr: "10.0.0.2:9333", CtrlAddr: "10.0.0.2:9334"}, + }}, + Epoch: 1, + RecoveryTargets: map[string]engine.SessionKind{ + replicaID: engine.SessionRebuild, + }, + }) + + sender := bs.v2Orchestrator.Registry.Sender(replicaID) + if sender == nil { + t.Fatal("sender not found") + } + snap := sender.SessionSnapshot() + if snap == nil || snap.Kind != engine.SessionRebuild { + t.Fatalf("session=%+v", snap) + } + + rm := NewRecoveryManager(bs) + bs.v2Recovery = rm + rm.OnPendingExecution = func(volumeID string, pending *pendingRecoveryExecution) { + if volumeID != volPath || pending == nil || pending.plan == nil { + return + } + pending.rebuildIO = fakeRebuildIO{achievedLSN: pending.plan.RebuildTargetLSN} + } + _, _, rebuildPort := bs.ReplicationPorts(volPath) + rebuildAddr := fmt.Sprintf("127.0.0.1:%d", rebuildPort) + rm.runRebuild(context.Background(), replicaID, rebuildAddr) + + proj, ok := bs.CoreProjection(volPath) + if !ok { + t.Fatal("expected cached core projection after live command-driven rebuild") + } + if proj.Recovery.Phase != engine.RecoveryIdle { + t.Fatalf("recovery_phase=%s", proj.Recovery.Phase) + } + if sender.State() != engine.StateInSync { + t.Fatalf("sender state=%s, want in_sync", sender.State()) + } + if got := bs.ExecutedCoreCommands(volPath); len(got) == 0 || got[len(got)-1] != "start_rebuild" { + t.Fatalf("expected start_rebuild execution, got %v", got) + } +} + +func TestP16B_RunRebuild_FailClosedWithoutFreshStartRebuildCommand(t *testing.T) { + bs, volPath := createTestBlockServiceWithVolCoreNoRecovery(t) + + var targetLSN uint64 + if err := bs.blockStore.WithVolume(volPath, func(vol *blockvol.BlockVol) error { + for i := 0; i < 5; i++ { + if err := vol.WriteLBA(uint64(i), make([]byte, 4096)); err != nil { + return err + } + } + if err := vol.ForceFlush(); err != nil { + return err + } + targetLSN = vol.StatusSnapshot().CommittedLSN + return nil + }); err != nil { + t.Fatalf("write+flush: %v", err) + } + + bs.ProcessAssignments([]blockvol.BlockVolumeAssignment{{ + Path: volPath, + Epoch: 1, + Role: uint32(blockvol.RolePrimary), + ReplicaServerID: "vs2", + ReplicaDataAddr: "10.0.0.2:9333", + ReplicaCtrlAddr: "10.0.0.2:9334", + }}) + + replicaID := volPath + "/vs2" + bs.v2Orchestrator.ProcessAssignment(engine.AssignmentIntent{ + Replicas: []engine.ReplicaAssignment{{ + ReplicaID: replicaID, + Endpoint: engine.Endpoint{DataAddr: "10.0.0.2:9333", CtrlAddr: "10.0.0.2:9334"}, + }}, + Epoch: 1, + RecoveryTargets: map[string]engine.SessionKind{ + replicaID: engine.SessionRebuild, + }, + }) + + // Prime the core with the same rebuild target before wiring recovery, + // so the subsequent live run does not emit a fresh start_rebuild command. + bs.applyCoreEvent(engine.RebuildStarted{ID: volPath, TargetLSN: targetLSN}) + before := bs.ExecutedCoreCommands(volPath) + + rm := NewRecoveryManager(bs) + bs.v2Recovery = rm + _, _, rebuildPort := bs.ReplicationPorts(volPath) + rebuildAddr := fmt.Sprintf("127.0.0.1:%d", rebuildPort) + rm.runRebuild(context.Background(), replicaID, rebuildAddr) + + after := bs.ExecutedCoreCommands(volPath) + if !reflect.DeepEqual(after, before) { + t.Fatalf("rebuild should fail closed without fresh start_rebuild command: before=%v after=%v", before, after) + } + sender := bs.v2Orchestrator.Registry.Sender(replicaID) + if sender == nil { + t.Fatal("sender not found") + } + if sender.State() == engine.StateInSync { + t.Fatalf("sender should not become in_sync without executing start_rebuild, state=%s", sender.State()) + } +} + // --- Serialized replacement: old drained before new starts --- func TestP4_SerializedReplacement_DrainsBeforeStart(t *testing.T) {