From 11c6aaf316ed80fc31c8e65ff874536a10429260 Mon Sep 17 00:00:00 2001 From: pingqiu Date: Sat, 4 Apr 2026 02:13:08 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20Batch=207=20+=20Phase=2016C-E=20?= =?UTF-8?q?=E2=80=94=20command=20dispatch=20extraction=20+=20engine=20refi?= =?UTF-8?q?nements?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Batch 7: Command dispatch binding extraction - New weed/server/blockcmd package: CommandHandler interface + DispatchCommands - volume_server_block.go applyCoreCommandsWithAssignment delegates to dispatcher - weed/server still owns RecordCommand, EmitCoreEvent, PublishProjection - v2bridge NOT given command-switch or event-emission semantics Phase 16C: Rebuilding assignment enters core command path Phase 16D: Rebuild recovery-task startup is command-driven Phase 16E: Catch-up recovery-task startup is command-driven Engine refinements: - RecoveryTarget on AssignmentDelivered event - shouldStartRecoveryTask / shouldStartReceiver guards - bootstrapReason: awaiting_rebuild_start Bridge/contract updates: - control_adapter.go: refined translation helpers - contract.go: executor port alignment Migration design docs (Batch 1-3 delivered, design artifacts): - v2-first/second/third-migration-batch.md + task-pack.md - v2-assignment-translation-unification.md - v2-execution-muscles-inventory.md - v2-separation-port-layer-audit.md - v2-legacy-runtime-exit-criteria.md Co-Authored-By: Claude Opus 4.6 (1M context) --- sw-block/.private/phase/phase-16-log.md | 239 +++++++++++++- sw-block/.private/phase/phase-16.md | 127 +++++++- sw-block/.private/phase/phase-16e-review.md | 156 +++++++++ sw-block/bridge/blockvol/bridge_test.go | 21 ++ sw-block/bridge/blockvol/contract.go | 44 +-- sw-block/bridge/blockvol/control_adapter.go | 54 +-- sw-block/bridge/blockvol/doc.go | 24 +- .../v2-assignment-translation-unification.md | 108 ++++++ .../design/v2-execution-muscles-inventory.md | 149 +++++++++ sw-block/design/v2-first-migration-batch.md | 109 +++++++ .../design/v2-first-migration-task-pack.md | 308 ++++++++++++++++++ .../design/v2-legacy-runtime-exit-criteria.md | 99 ++++++ sw-block/design/v2-second-migration-batch.md | 109 +++++++ .../design/v2-second-migration-task-pack.md | 230 +++++++++++++ .../design/v2-separation-port-layer-audit.md | 135 ++++++++ sw-block/design/v2-third-migration-batch.md | 103 ++++++ .../design/v2-third-migration-task-pack.md | 218 +++++++++++++ sw-block/engine/replication/command.go | 8 + sw-block/engine/replication/engine.go | 33 +- sw-block/engine/replication/event.go | 9 +- .../replication/phase14_command_test.go | 71 ++-- sw-block/engine/replication/state.go | 14 +- weed/server/blockcmd/dispatch.go | 139 ++++++++ weed/server/blockcmd/dispatch_test.go | 186 +++++++++++ weed/server/volume_server_block.go | 272 +++++++++------- 25 files changed, 2757 insertions(+), 208 deletions(-) create mode 100644 sw-block/.private/phase/phase-16e-review.md create mode 100644 sw-block/design/v2-assignment-translation-unification.md create mode 100644 sw-block/design/v2-execution-muscles-inventory.md create mode 100644 sw-block/design/v2-first-migration-batch.md create mode 100644 sw-block/design/v2-first-migration-task-pack.md create mode 100644 sw-block/design/v2-legacy-runtime-exit-criteria.md create mode 100644 sw-block/design/v2-second-migration-batch.md create mode 100644 sw-block/design/v2-second-migration-task-pack.md create mode 100644 sw-block/design/v2-separation-port-layer-audit.md create mode 100644 sw-block/design/v2-third-migration-batch.md create mode 100644 sw-block/design/v2-third-migration-task-pack.md create mode 100644 weed/server/blockcmd/dispatch.go create mode 100644 weed/server/blockcmd/dispatch_test.go diff --git a/sw-block/.private/phase/phase-16-log.md b/sw-block/.private/phase/phase-16-log.md index cd2f53ce2..10e716b77 100644 --- a/sw-block/.private/phase/phase-16-log.md +++ b/sw-block/.private/phase/phase-16-log.md @@ -470,7 +470,7 @@ Date: 2026-04-04 Current checkpoint judgment: -1. this is the first bounded integrated runtime checkpoint after `Phase 15` +1. this is the current widened bounded runtime checkpoint after `Phase 15` closeout 2. `16A` is delivered 3. `16B` has one accepted current closure: @@ -545,3 +545,240 @@ Review intent: 1. this note does not broaden `16B` 2. it only closes the two minor gaps identified by `manager` + +--- + +#### `16C` Start Note Rev 1 + +Date: 2026-04-04 +Scope: bounded rebuilding-assignment entry ownership on the core-present path + +Why this slice exists: + +1. `16B Rev 3` closed bounded catch-up/rebuild command ownership once recovery + execution was already on the live path +2. one remaining adapter-local rebuild trigger still existed at assignment time: + `RoleRebuilding` assignments could call legacy `BlockService.startRebuild()` + directly +3. assignment handling also started recovery tasks before the local assignment + apply path had completed + +What changed: + +1. `weed/server/volume_server_block.go` + - rebuilding assignments no longer directly call legacy `startRebuild()` when + the core is present + - rebuilding assignments now enter `coreAssignmentEvent()` and therefore use + the same bounded `apply_role` command path as other integrated assignments + - recovery-task startup from orchestrator results is deferred until after the + assignment apply loop completes + - legacy direct `startRebuild()` is now explicitly reserved for no-core + fallback only +2. `weed/server/volume_server_block_test.go` + - added proof that core-present rebuilding assignment: + - does not use legacy direct rebuild start + - still applies local rebuilding role + - reaches core-driven `start_rebuild` + - added proof that no-core fallback still uses legacy direct rebuild start + +--- + +#### `16C` Delivery Note Rev 2 + +Date: 2026-04-04 +Scope: suppress false replica-ready receiver startup on rebuilding assignment + +What changed: + +1. `sw-block/engine/replication/event.go` + - `AssignmentDelivered` now carries `RecoveryTarget` +2. `sw-block/engine/replication/engine.go` + - rebuilding assignment with `RecoveryTarget=SessionRebuild` no longer emits + `start_receiver` + - bootstrap reason for that bounded state is now `awaiting_rebuild_start` + instead of `awaiting_receiver_ready` +3. `weed/server/volume_server_block.go` + - `RoleRebuilding` assignment now marks the core assignment event with + `SessionRebuild` +4. `weed/server/volume_server_block_test.go` + - tightened focused proof to require exactly: + `apply_role`, then `start_rebuild` + - explicitly rejects `start_receiver` on the rebuilding-assignment path + +Bounded contract refinement: + +`16C` now additionally accepts only this narrow improvement: + +1. rebuilding assignment is not misclassified as receiver-start work +2. command egress for that path is the minimum bounded set needed for: + - local rebuilding role apply + - rebuild execution ownership + +Validation: + +1. `go test ./sw-block/engine/replication/...` +2. `go test ./weed/server -run "TestBlockService_ApplyAssignments_RebuildingRole_(UsesCoreRecoveryPathWithoutLegacyDirectStart|PreservesLegacyFallbackWithoutCore)"` +3. `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)"` +4. result: `PASS` + +--- + +#### `16D` Start Note Rev 1 + +Date: 2026-04-04 +Scope: bounded rebuild recovery-task startup ownership + +Why this slice exists: + +1. `16C` closed rebuilding-assignment entry ownership +2. one bounded startup gap still remained: + - recovery goroutine startup on the core-present path still came directly from + orchestrator `SessionsCreated` / `SessionsSuperseded` + - the core only owned later rebuild execution (`start_rebuild`), not the task + startup itself +3. this is a runtime-loop ownership gap, but still narrow enough to fix only for + rebuilding-assignment startup + +What changed: + +1. `sw-block/engine/replication/command.go` + - added bounded `StartRecoveryTaskCommand` +2. `sw-block/engine/replication/state.go` + - added command dedupe state for recovery-task startup +3. `sw-block/engine/replication/engine.go` + - rebuilding assignment now emits `start_recovery_task` + - that command is deduped by epoch / replica / recovery kind +4. `sw-block/engine/replication/phase14_command_test.go` + - added command proof that rebuilding assignment emits: + - `apply_role` + - `start_recovery_task` + - `publish_projection` +5. `weed/server/volume_server_block.go` + - core-present path no longer starts recovery tasks from + `SessionsCreated` / `SessionsSuperseded` + - adapter now executes `start_recovery_task` + - removed-sender drain remains preserved + - legacy no-core startup path still uses `HandleAssignmentResult()` +6. `weed/server/block_recovery.go` + - restored `HandleAssignmentResult()` as a no-core compatibility entry + - added bounded `StartRecoveryTask()` entry for core-command execution +7. `weed/server/volume_server_block_test.go` + - rebuilding-assignment proof now requires: + - `apply_role` + - `start_recovery_task` + - `start_rebuild` + +Bounded contract: + +`16D Rev 1` currently accepts only this: + +1. rebuild recovery-task startup is core-command-driven on the core-present + rebuilding-assignment path +2. adapter no longer starts that task directly from orchestrator create/supersede + results on the bounded path +3. no-core / old `P4` live-path ownership proofs are preserved +4. those legacy `P4` proofs now serve as compatibility guards only; they are not + the semantic authority proof for the core-present `16D` path + +It does not yet accept: + +1. catch-up task startup ownership +2. full recovery-loop closure +3. broad end-to-end failover/recovery/publication closure +4. multi-replica recovery ownership +5. launch / rollout readiness + +Validation: + +1. `go test ./sw-block/engine/replication/...` +2. `go test ./weed/server -run "TestBlockService_ApplyAssignments_RebuildingRole_(UsesCoreRecoveryPathWithoutLegacyDirectStart|PreservesLegacyFallbackWithoutCore)"` +3. `go test ./weed/server -run "TestP4_(LivePath_RealVol_ReachesPlan|SerializedReplacement_DrainsBeforeStart|ShutdownDrain)"` +4. `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)"` +5. result: `PASS` + +--- + +#### `16E` Start Note Rev 1 + +Date: 2026-04-04 +Scope: bounded catch-up recovery-task startup ownership on the core-present path + +Why this slice exists: + +1. `16D` closed rebuild recovery-task startup ownership +2. the parallel catch-up startup path still started from orchestrator + `SessionsCreated` / `SessionsSuperseded` +3. this left the current `RF=2` primary-assignment path only partially + command-driven + +What changed: + +1. `sw-block/engine/replication/engine.go` + - `start_recovery_task` dedupe now resets on assignment change + - startup command remains bounded to a single desired replica + - startup command ordering now places `configure_shipper` before + `start_recovery_task` on the primary catch-up path +2. `sw-block/engine/replication/phase14_command_test.go` + - primary assignment now proves: + - `apply_role` + - `configure_shipper` + - `start_recovery_task` + - `publish_projection` + - assignment change proof now requires a fresh `start_recovery_task` + - catch-up planning proof now runs from the primary catch-up shape +3. `weed/server/volume_server_block.go` + - primary assignment now marks `RecoveryTarget=SessionCatchUp` in the core + assignment event on the bounded single-replica path +4. `weed/server/volume_server_block_test.go` + - added focused proof that the core-present primary path executes: + - `apply_role` + - `configure_shipper` + - `start_recovery_task` + - `start_catchup` + - proves sender reaches `in_sync` and projection returns to `RecoveryIdle` + +Bounded contract: + +`16E Rev 1` currently accepts only this: + +1. catch-up recovery-task startup is core-command-driven on the core-present + primary-assignment path +2. adapter no longer starts that task directly from orchestrator create/supersede + results on that bounded path +3. no-core / legacy `P4` compatibility remains preserved + +It does not yet accept: + +1. multi-replica catch-up startup ownership +2. full recovery-loop closure +3. broad end-to-end failover/recovery/publication closure +4. launch / rollout readiness + +Validation: + +1. `go test ./sw-block/engine/replication/...` +2. `go test ./weed/server -run "TestBlockService_ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart|RebuildingRole_PreservesLegacyFallbackWithoutCore)"` +3. `go test ./weed/server -run "Test(P4_|P16B_|BlockService_(ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|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)"` +4. result: `PASS` + +Bounded contract: + +`16C Rev 1` currently accepts only this: + +1. rebuilding-assignment entry no longer bypasses the core on the core-present + path +2. local role apply precedes recovery task start on that bounded path +3. old no-core fallback is preserved + +It does not yet accept: + +1. full recovery-loop closure +2. broad end-to-end failover/recovery/publication closure +3. multi-replica rebuild ownership +4. launch / rollout readiness + +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` diff --git a/sw-block/.private/phase/phase-16.md b/sw-block/.private/phase/phase-16.md index a4b3bc29b..2ee271b74 100644 --- a/sw-block/.private/phase/phase-16.md +++ b/sw-block/.private/phase/phase-16.md @@ -98,6 +98,104 @@ Current chosen path: Status: +1. delivered + +### `16C`: Rebuild Assignment Entry Ownership + +Goal: + +1. remove one remaining adapter-local rebuild entry trigger from the core-present + path +2. serialize rebuilding assignment local apply before the recovery task starts + +Acceptance object: + +1. `RoleRebuilding` assignment does not directly trigger legacy + `BlockService.startRebuild()` when the core is present +2. the same assignment still applies local role through the core-driven command + chain +3. recovery task start happens after the local assignment command path, not + before it +4. old no-core fallback remains preserved + +Current chosen path: + +1. rebuilding assignment now participates in the same core assignment command + chain as other bounded roles +2. direct legacy `startRebuild()` is now reserved for no-core fallback only +3. orchestrator-driven recovery task start is deferred until the assignment + apply path has completed +4. rebuilding assignment no longer emits `start_receiver` as a false + replica-ready side effect on the core-present path + +Status: + +1. delivered + +### `16D`: Rebuild Task Startup Ownership + +Goal: + +1. move bounded rebuild recovery-task startup from direct orchestrator-result + handling into the core command path +2. preserve old no-core recovery startup behavior for legacy proofs + +Acceptance object: + +1. on the core-present rebuilding-assignment path, recovery goroutine startup + happens because the core emitted a command +2. the adapter no longer starts rebuild recovery tasks directly from + `SessionsCreated` / `SessionsSuperseded` on that bounded path +3. no-core / older `P4` live-path proofs still pass unchanged + +Current chosen path: + +1. rebuilding assignment now emits a bounded `start_recovery_task` command +2. adapter executes that command by starting one recovery goroutine for the + already-attached rebuild session +3. core-present path now uses: + - `apply_role` + - `start_recovery_task` + - `start_rebuild` +4. old `HandleAssignmentResult()` startup behavior is retained only for no-core + compatibility and legacy `P4` proof preservation + +Status: + +1. delivered + +### `16E`: Catch-Up Task Startup Ownership + +Goal: + +1. move bounded catch-up recovery-task startup from direct orchestrator-result + handling into the core command path +2. keep the slice bounded to the single-replica `RF=2` chosen path + +Acceptance object: + +1. on the core-present primary-assignment path, catch-up recovery-task startup + happens because the core emitted a command +2. the bounded command sequence for that path becomes: + - `apply_role` + - `configure_shipper` + - `start_recovery_task` + - `start_catchup` +3. old no-core / legacy `P4` compatibility remains preserved + +Current chosen path: + +1. primary assignment with one replica now marks `RecoveryTarget=SessionCatchUp` + in the core assignment event +2. the core emits `start_recovery_task` for that bounded catch-up path +3. the adapter starts the recovery goroutine from that command, not from + orchestrator create/supersede results +4. assignment change resets the dedupe key for recovery-task startup, so endpoint + change / reassign still emits a fresh task-start command +5. multi-replica startup ownership remains outside the current bounded path + +Status: + 1. active ## Current Checkpoint Review Target @@ -118,16 +216,22 @@ after `Phase 15` closeout: - bounded catch-up execution runs from `StartCatchUpCommand` 4. current working state extends that bounded path with: - bounded rebuild execution from `StartRebuildCommand` + - rebuilding assignment entry ownership + - rebuild recovery-task startup ownership + - bounded catch-up recovery-task startup ownership on the single-replica + 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. launch / rollout readiness is not claimed +3. multi-replica startup ownership is not yet claimed +4. launch / rollout readiness is not claimed ## Immediate Next Step -The current checkpoint is now good enough to take as a stage/commit boundary: +The current checkpoint is now good enough to take as the next stage/review +boundary: 1. `Phase 15` delivered 2. `16A` delivered @@ -135,10 +239,19 @@ The current checkpoint is now good enough to take as a stage/commit boundary: - live recovery observations close back into the core - bounded catch-up execution is core-command-driven - bounded rebuild execution is core-command-driven +4. `16C` delivered: + - rebuilding assignment entry no longer bypasses the core + - rebuilding assignment no longer emits false `start_receiver` +5. `16D` delivered: + - rebuild recovery-task startup is core-command-driven +6. `16E` current bounded refinement: + - catch-up recovery-task startup is core-command-driven on the + single-replica primary path -After that checkpoint, decide whether `Phase 16` needs one stricter end-to-end -recovery/publication scenario before moving beyond the phase: +After this checkpoint: -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 +1. keep `legacy P4` only as a compatibility guard +2. decide whether to widen startup ownership beyond the bounded single-replica + catch-up path +3. do not yet claim full recovery-loop closure +4. do not broaden into launch claims diff --git a/sw-block/.private/phase/phase-16e-review.md b/sw-block/.private/phase/phase-16e-review.md new file mode 100644 index 000000000..03da45c08 --- /dev/null +++ b/sw-block/.private/phase/phase-16e-review.md @@ -0,0 +1,156 @@ +# Phase 16E Review + +Date: 2026-04-04 +Status: ready for review + +## Review Object + +Review the current bounded `Phase 16E` working state as: + +1. `Phase 15` delivered +2. `16A` delivered +3. `16B` delivered +4. `16C` delivered +5. `16D` delivered +6. `16E` bounded catch-up recovery-task startup ownership on the + single-replica primary path + +## What Is In Scope + +### Previously delivered closure + +Please treat these as already accepted background: + +1. `Phase 15` surface/store/outward consume-chain rebinding +2. `16A` command-driven adapter ownership +3. `16B` live recovery execution ownership +4. `16C` rebuilding-assignment entry ownership +5. `16D` rebuild recovery-task startup ownership + +### `16E` current bounded refinement + +Review only this new bounded step: + +1. primary assignment with one replica now marks `RecoveryTarget=SessionCatchUp` + in the core assignment event +2. the core emits `start_recovery_task` for that bounded catch-up startup path +3. the adapter starts the recovery goroutine from that command, not from + orchestrator `SessionsCreated` / `SessionsSuperseded` +4. the bounded command sequence for this path is now: + - `apply_role` + - `configure_shipper` + - `start_recovery_task` + - `start_catchup` +5. assignment change resets the startup dedupe key, so endpoint/version change + still emits a fresh task-start command +6. legacy `P4` remains preserved only as a compatibility guard + +Expected judgment: + +1. this is still a bounded runtime-ownership refinement +2. catch-up task startup is now core-command-driven on the bounded single-replica + primary path +3. this is not yet full recovery-loop ownership + +## What Is Explicitly Out Of Scope + +Do NOT review `16E` as claiming: + +1. multi-replica catch-up startup ownership +2. full recovery-loop closure +3. broad end-to-end failover/recovery/publication closure +4. launch / rollout readiness + +## Primary Files + +Phase tracking: + +1. `sw-block/.private/phase/phase-16.md` +2. `sw-block/.private/phase/phase-16-log.md` + +Core/runtime code: + +1. `sw-block/engine/replication/command.go` +2. `sw-block/engine/replication/state.go` +3. `sw-block/engine/replication/engine.go` +4. `sw-block/engine/replication/phase14_command_test.go` +5. `weed/server/block_recovery.go` +6. `weed/server/volume_server_block.go` +7. `weed/server/volume_server_block_test.go` + +## Evidence Summary + +### Engine command proof + +1. `go test ./sw-block/engine/replication/...` +2. result: `PASS` +3. key proof: + - `TestPhase14_CommandSequence_PrimaryAssignmentIsBounded` + - proves primary assignment now emits: + - `apply_role` + - `configure_shipper` + - `start_recovery_task` + - `publish_projection` +4. supporting proof: + - `TestPhase14_CommandSequence_AssignmentChangeAllowsFreshRecoveryStart` + - proves assignment change re-emits fresh `start_recovery_task` + +### Focused integrated proof + +1. `go test ./weed/server -run "TestBlockService_ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart|RebuildingRole_PreservesLegacyFallbackWithoutCore)"` +2. result: `PASS` +3. key new catch-up proof: + - `TestBlockService_ApplyAssignments_PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp` + - proves executed command sequence: + - `apply_role` + - `configure_shipper` + - `start_recovery_task` + - `start_catchup` + - proves sender reaches `StateInSync` + - proves projection returns to `RecoveryIdle` + +### Compatibility and aggregate proof + +1. `go test ./weed/server -run "TestP4_(LivePath_RealVol_ReachesPlan|SerializedReplacement_DrainsBeforeStart|ShutdownDrain)"` +2. result: `PASS` +3. `legacy P4` is still preserved as compatibility guard only +4. `go test ./weed/server -run "Test(P4_|P16B_|BlockService_(ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|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)"` +5. result: `PASS` + +## Review Questions + +### For `tester` + +Please challenge the proof posture: + +1. Does `16E` really prove catch-up task startup is command-driven on the bounded + core-present primary path? +2. Are the proofs behavioral enough, rather than just proving command plumbing? +3. Is assignment-change reissue of `start_recovery_task` bounded and correct? +4. Are there any remaining bounded single-replica catch-up startup paths that + still bypass the core when the core is present? + +### For `manager` + +Please challenge boundaries and overclaim: + +1. Is `16E` still a bounded refinement rather than a disguised move toward full + recovery-loop closure? +2. Is the claim narrow enough: + - single-replica primary catch-up startup only + - not multi-replica startup ownership + - not full runtime-loop ownership +3. Is `legacy P4` positioning now disciplined enough: + - compatibility guard + - not semantic authority proof for the core-present path +4. Is this a reasonable review/commit boundary? + +## Requested Output Shape + +Please reply with one of: + +1. `ACCEPT` +2. `ACCEPT WITH MINOR FIXES` +3. `REJECT` + +If not `ACCEPT`, keep findings bounded to this `16E` claim only. diff --git a/sw-block/bridge/blockvol/bridge_test.go b/sw-block/bridge/blockvol/bridge_test.go index 53a6a89c9..7961f4d46 100644 --- a/sw-block/bridge/blockvol/bridge_test.go +++ b/sw-block/bridge/blockvol/bridge_test.go @@ -177,4 +177,25 @@ func TestContract_BlockVolReaderInterface(t *testing.T) { // Verify the contract interface is implementable. var _ BlockVolReader = &pushReader{psa: NewPushStorageAdapter()} var _ BlockVolPinner = &pushPinner{psa: NewPushStorageAdapter()} + var _ BlockVolCatchUpIO = fakeExecutor{} + var _ BlockVolRebuildIO = fakeExecutor{} + var _ BlockVolExecutor = fakeExecutor{} +} + +type fakeExecutor struct{} + +func (fakeExecutor) StreamWALEntries(startExclusive, endInclusive uint64) (uint64, error) { + return endInclusive, nil +} + +func (fakeExecutor) TruncateWAL(truncateLSN uint64) error { + return nil +} + +func (fakeExecutor) TransferSnapshot(snapshotLSN uint64) error { + return nil +} + +func (fakeExecutor) TransferFullBase(committedLSN uint64) (uint64, error) { + return committedLSN, nil } diff --git a/sw-block/bridge/blockvol/contract.go b/sw-block/bridge/blockvol/contract.go index ef41b6465..09d80a002 100644 --- a/sw-block/bridge/blockvol/contract.go +++ b/sw-block/bridge/blockvol/contract.go @@ -1,5 +1,7 @@ package blockvol +import engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" + // === Phase 07 P1: Handoff contract === // // This file defines the interface boundary between: @@ -19,11 +21,11 @@ package blockvol // BlockVolState represents the real storage state from a blockvol instance. // Each field maps to a specific blockvol source (current P1 implementation): // -// WALHeadLSN ← vol.nextLSN - 1 (last written LSN) -// WALTailLSN ← vol.super.WALCheckpointLSN (LSN boundary, not byte offset) -// CommittedLSN ← vol.flusher.CheckpointLSN() (V1 interim: committed = checkpointed) -// CheckpointLSN ← vol.super.WALCheckpointLSN (durable base image) -// CheckpointTrusted ← vol.super.Validate() == nil (superblock integrity) +// WALHeadLSN ← vol.nextLSN - 1 (last written LSN) +// WALTailLSN ← vol.super.WALCheckpointLSN (LSN boundary, not byte offset) +// CommittedLSN ← vol.flusher.CheckpointLSN() (V1 interim: committed = checkpointed) +// CheckpointLSN ← vol.super.WALCheckpointLSN (durable base image) +// CheckpointTrusted ← vol.super.Validate() == nil (superblock integrity) type BlockVolState struct { WALHeadLSN uint64 WALTailLSN uint64 @@ -63,20 +65,24 @@ type BlockVolPinner interface { HoldFullBase(committedLSN uint64) (release func(), err error) } -// BlockVolExecutor performs actual recovery I/O. Implemented by the -// weed-side bridge. It does NOT decide recovery policy — it only -// executes what the engine tells it to do. -type BlockVolExecutor interface { - // StreamWALEntries streams WAL entries from startExclusive+1 to endInclusive - // to the replica. Returns the highest LSN successfully transferred. - StreamWALEntries(startExclusive, endInclusive uint64) (transferredTo uint64, err error) +// BlockVolCatchUpIO is the weed-free catch-up execution port. It intentionally +// matches engine.CatchUpIO so executor implementations can plug directly into +// the V2 runtime without importing weed/ into sw-block. +type BlockVolCatchUpIO interface { + engine.CatchUpIO +} - // TransferSnapshot transfers a checkpoint/snapshot at snapshotLSN to the replica. - TransferSnapshot(snapshotLSN uint64) error +// BlockVolRebuildIO is the weed-free rebuild execution port. It intentionally +// matches engine.RebuildIO so rebuild mechanics can move behind sw-block-owned +// contracts while real blockvol calls remain in thin adapter implementations. +type BlockVolRebuildIO interface { + engine.RebuildIO +} - // TransferFullBase transfers the full extent image to the replica. - TransferFullBase(committedLSN uint64) error - - // TruncateWAL removes entries beyond truncateLSN from the replica. - TruncateWAL(truncateLSN uint64) error +// BlockVolExecutor is the combined execution-muscle surface for the current +// bounded runtime path. Implementations execute I/O only; they do not own +// recovery policy, lifecycle meaning, or publication semantics. +type BlockVolExecutor interface { + BlockVolCatchUpIO + BlockVolRebuildIO } diff --git a/sw-block/bridge/blockvol/control_adapter.go b/sw-block/bridge/blockvol/control_adapter.go index c0a3a6d16..6b12b8e0d 100644 --- a/sw-block/bridge/blockvol/control_adapter.go +++ b/sw-block/bridge/blockvol/control_adapter.go @@ -37,6 +37,30 @@ func MakeReplicaID(volumeName, serverID string) string { return fmt.Sprintf("%s/%s", volumeName, serverID) } +// ReplicaAssignmentForServer builds one engine replica assignment from stable +// volume/server identity plus endpoint. This is the canonical identity mapping +// shared by control ingestion and adapter-side assignment rebinding. +func ReplicaAssignmentForServer(volumeName, serverID string, endpoint engine.Endpoint) engine.ReplicaAssignment { + return engine.ReplicaAssignment{ + ReplicaID: MakeReplicaID(volumeName, serverID), + Endpoint: endpoint, + } +} + +// RecoveryTargetForRole maps a role-shaped control input to the bounded engine +// recovery target. This is a pure translation rule, not a recovery policy +// decision. +func RecoveryTargetForRole(role string) engine.SessionKind { + switch role { + case "replica": + return engine.SessionCatchUp + case "rebuilding": + return engine.SessionRebuild + default: + return "" + } +} + // ToAssignmentIntent converts a master assignment into an engine intent. // The adapter maps role transitions to SessionKind but does NOT decide // the actual recovery outcome (that's the engine's job). @@ -46,38 +70,22 @@ func (ca *ControlAdapter) ToAssignmentIntent(primary MasterAssignment, replicas } for _, r := range replicas { - replicaID := MakeReplicaID(r.VolumeName, r.ReplicaServerID) - intent.Replicas = append(intent.Replicas, engine.ReplicaAssignment{ - ReplicaID: replicaID, - Endpoint: engine.Endpoint{ - DataAddr: r.DataAddr, - CtrlAddr: r.CtrlAddr, - Version: r.AddrVersion, - }, + replica := ReplicaAssignmentForServer(r.VolumeName, r.ReplicaServerID, engine.Endpoint{ + DataAddr: r.DataAddr, + CtrlAddr: r.CtrlAddr, + Version: r.AddrVersion, }) + intent.Replicas = append(intent.Replicas, replica) // Map role to recovery intent (if needed). - kind := mapRoleToSessionKind(r.Role) + kind := RecoveryTargetForRole(r.Role) if kind != "" { if intent.RecoveryTargets == nil { intent.RecoveryTargets = map[string]engine.SessionKind{} } - intent.RecoveryTargets[replicaID] = kind + intent.RecoveryTargets[replica.ReplicaID] = kind } } return intent } - -// mapRoleToSessionKind maps a master-assigned role to an engine SessionKind. -// This is a pure translation — NO policy decision. -func mapRoleToSessionKind(role string) engine.SessionKind { - switch role { - case "replica": - return engine.SessionCatchUp // default recovery for reconnecting replicas - case "rebuilding": - return engine.SessionRebuild - default: - return "" // no recovery needed (primary, or unknown) - } -} diff --git a/sw-block/bridge/blockvol/doc.go b/sw-block/bridge/blockvol/doc.go index 95b803ecd..95a0a756e 100644 --- a/sw-block/bridge/blockvol/doc.go +++ b/sw-block/bridge/blockvol/doc.go @@ -1,9 +1,14 @@ -// Package blockvol bridges the V2 engine to real blockvol storage and -// control-plane state. +// Package blockvol defines the weed-free bridge contracts that connect the V2 +// engine to blockvol-backed control and execution mechanics. // -// This package implements the adapter interfaces defined in -// sw-block/engine/replication/ using real blockvol internals as the -// source of truth. +// This package owns: +// - stable control translation helpers +// - storage/execution port contracts +// - thin adapters that consume those contracts without importing weed/ +// +// Real blockvol-backed implementations live outside this package (today under +// weed/storage/blockvol/v2bridge/). This package must remain reusable from +// sw-block without directly depending on weed/. // // Hard rules (Phase 07): // - ReplicaID = / (not address-derived) @@ -12,8 +17,9 @@ // - Bridge translates engine decisions into blockvol actions // // Adapter replacement order: -// P0: control_adapter (assignment → engine intent) -// P0: storage_adapter (blockvol state → RetainedHistory) -// P1: executor_bridge (engine executor → blockvol I/O) -// P1: observe_adapter (engine status → service diagnostics) +// +// P0: control_adapter (assignment → engine intent) +// P0: storage_adapter (blockvol state → RetainedHistory) +// P1: executor_bridge (engine executor → blockvol I/O) +// P1: observe_adapter (engine status → service diagnostics) package blockvol diff --git a/sw-block/design/v2-assignment-translation-unification.md b/sw-block/design/v2-assignment-translation-unification.md new file mode 100644 index 000000000..dc7619598 --- /dev/null +++ b/sw-block/design/v2-assignment-translation-unification.md @@ -0,0 +1,108 @@ +# V2 Assignment Translation Unification + +Date: 2026-04-04 +Status: active + +## Purpose + +This note defines how assignment translation should be unified so that: + +1. `weed/storage/blockvol/v2bridge/control.go` +2. `weed/server/volume_server_block.go` + +do not drift on identity, role, or recovery-target meaning. + +## Current Drift Risk + +Today there are two live translation sites: + +1. `ControlBridge.ConvertAssignment()` produces `engine.AssignmentIntent` +2. `BlockService.coreAssignmentEvent()` produces `engine.AssignmentDelivered` + +They do not translate the same source type, but they do share semantic rules: + +1. how to build stable `ReplicaID` +2. how to map role-shaped inputs to recovery target +3. how to represent one local replica endpoint in engine types + +If those rules stay duplicated, later migration batches will reintroduce split +truth. + +## Canonical Rule Placement + +The canonical reusable rules belong in: + +1. `sw-block/bridge/blockvol` + +Why: + +1. the rules are semantic translation, not product integration +2. both `weed/storage/blockvol/v2bridge` and `weed/server` can import this + package +3. `sw-block` remains weed-free + +## Rules That Must Be Canonical + +### 1. Stable identity + +Canonical helper: + +1. `MakeReplicaID()` +2. `ReplicaAssignmentForServer()` + +Rule: + +1. `ReplicaID = /` +2. never derive identity from transport address + +### 2. Recovery-target mapping + +Canonical helper: + +1. `RecoveryTargetForRole()` + +Rule: + +1. `replica -> catchup` +2. `rebuilding -> rebuild` +3. all other roles -> no recovery target + +### 3. Endpoint packaging + +Canonical helper: + +1. `ReplicaAssignmentForServer()` + +Rule: + +1. adapter code may still source endpoint fields from different wire/runtime + inputs +2. but once packaged into `engine.ReplicaAssignment`, the shape must be uniform + +## What Still Stays Local + +These parts remain adapter-local and should NOT be forced into one helper yet: + +1. reading `blockvol.BlockVolumeAssignment` +2. deciding whether the local VS is primary/replica/rebuilding in a given + runtime context +3. multi-replica traversal over heartbeat/master wire structures + +Those are source-format adaptation concerns, not canonical translation rules. + +## Implemented First Step + +The first unification step is already applied: + +1. `sw-block/bridge/blockvol/control_adapter.go` + - now exports the canonical helpers +2. `weed/server/volume_server_block.go` + - now consumes the same helper layer for local assignment rebinding + +## Next Step + +The next step after this document is: + +1. reduce `weed/storage/blockvol/v2bridge/control.go` to source-format + extraction only +2. keep all shared semantic mapping rules in `sw-block/bridge/blockvol` diff --git a/sw-block/design/v2-execution-muscles-inventory.md b/sw-block/design/v2-execution-muscles-inventory.md new file mode 100644 index 000000000..0e21b83b6 --- /dev/null +++ b/sw-block/design/v2-execution-muscles-inventory.md @@ -0,0 +1,149 @@ +# V2 Execution Muscles Inventory + +Date: 2026-04-04 +Status: active + +## Purpose + +This note inventories the current V1/weed code that should be treated as +execution muscles for separation, rather than semantic authority. + +## Movable Execution-Muscle Clusters + +### 1. Recovery executor mechanics + +Files: + +1. `weed/storage/blockvol/v2bridge/executor.go` +2. related tests under `weed/storage/blockvol/v2bridge/*transfer*` +3. related tests under `weed/storage/blockvol/v2bridge/*snapshot*` +4. related tests under `weed/storage/blockvol/v2bridge/*truncate*` + +Why this is a muscle: + +1. it performs TCP rebuild/catch-up I/O +2. it applies WAL entries, snapshots, and full-base transfer +3. it does not decide recovery policy; the engine already decides the plan + +### 2. Storage state read mechanics + +File: + +1. `weed/storage/blockvol/v2bridge/reader.go` + +Why this is a muscle: + +1. it only reads real `BlockVol` state +2. it populates the engine-facing retained-history view +3. it carries no semantic authority + +### 3. Pin / retention hold mechanics + +File: + +1. `weed/storage/blockvol/v2bridge/pinner.go` + +Why this is a muscle: + +1. it only implements hold/release against real flusher retention +2. pin meaning belongs to the engine; this file only executes the hold + +### 4. Recovery-side blockvol shims + +Current location: + +1. `weed/server/block_recovery.go` + +Candidate functions: + +1. `readerShimForRecovery` +2. `pinnerShimForRecovery` + +Why they are muscles: + +1. they are thin glue from runtime host code into bridge contracts +2. they should not remain mixed into recovery lifecycle ownership + +## Code That Should Stay In weed/ + +### 1. Product adapter shell + +Files: + +1. `weed/server/volume_server_block.go` +2. `weed/server/block_recovery.go` + +Why they stay: + +1. process lifecycle and goroutine hosting +2. command dispatch into real servers/backends +3. heartbeat, iSCSI, NVMe, and server integration + +### 2. Raw blockvol backend + +Files: + +1. `weed/storage/blockvol/*` + +Why they stay: + +1. they are the actual backend implementation +2. the separation goal is not to move `BlockVol` itself into `sw-block` + +### 3. Real master/heartbeat wire adaptation + +File: + +1. `weed/storage/blockvol/v2bridge/control.go` + +Why it stays for now: + +1. it consumes `weed/storage/blockvol.BlockVolumeAssignment` +2. it is still closest to master / heartbeat wire shape +3. canonical mapping logic can move into `sw-block`, but source-format + adaptation still belongs in `weed/` + +## Main Migration Risks + +### 1. Hidden weed dependency + +Risk: + +1. `executor.go`, `reader.go`, and `pinner.go` directly depend on + `weed/storage/blockvol.BlockVol` + +Implication: + +1. they cannot be moved into `sw-block` unchanged +2. ports/interfaces must be stabilized first + +### 2. Assignment translation drift + +Risk files: + +1. `weed/storage/blockvol/v2bridge/control.go` +2. `weed/server/volume_server_block.go` + +Implication: + +1. if migration happens before canonical helper extraction, identity and + recovery-target rules can drift + +### 3. Runtime host coupling + +Risk file: + +1. `weed/server/block_recovery.go` + +Implication: + +1. goroutine lifecycle, logging, sender invalidation, and core event emission + are still mixed together +2. only the execution muscles should move first, not the whole runtime host + +## Recommended Move Order + +1. stabilize contracts in `sw-block/bridge/blockvol` +2. extract canonical mapping helpers into `sw-block` +3. migrate reader/pinner/executor logic behind those ports +4. shrink `weed/server` to adapter shell afterward diff --git a/sw-block/design/v2-first-migration-batch.md b/sw-block/design/v2-first-migration-batch.md new file mode 100644 index 000000000..14b235ae8 --- /dev/null +++ b/sw-block/design/v2-first-migration-batch.md @@ -0,0 +1,109 @@ +# V2 First Migration Batch + +Date: 2026-04-04 +Status: delivered + +## Purpose + +This note defines the first migration batch for the `sw-block` separation work. + +The batch must: + +1. move code toward `sw-block` +2. keep `sw-block` free of direct `weed/` imports +3. avoid moving `BlockService` or `RecoveryManager` whole + +## Batch Goal + +Establish one clean execution-muscle layer behind `sw-block` ports, while +keeping `weed/` as a thin adapter shell. + +## Batch Scope + +### In scope + +1. `sw-block/bridge/blockvol` contract cleanup +2. canonical helper extraction for identity and recovery-target mapping +3. reader / pinner / executor migration target design +4. tests that prove those contracts and helpers + +### Out of scope + +1. moving `weed/server/volume_server_block.go` +2. moving `weed/server/block_recovery.go` +3. moving the full `blockvol` backend +4. broad master/heartbeat refactor + +## Target Package Shape + +### Keep as long-term owner + +1. `sw-block/engine/replication` +2. `sw-block/bridge/blockvol` + +### Future landing zone for execution muscles + +Recommended target inside `sw-block`: + +1. keep contracts in `sw-block/bridge/blockvol` +2. add a future execution-oriented package only after ports are stable, for + example: + - `sw-block/bridge/blockvol/runtime` + - or `sw-block/runtime/blockvol` + +For the first batch, do NOT create that new package yet unless the existing +contracts prove insufficient. + +### Keep as thin adapter implementations + +1. `weed/storage/blockvol/v2bridge` +2. `weed/server/*` + +## Concrete Batch Steps + +1. normalize `sw-block/bridge/blockvol` contracts so they match the engine's + real IO surfaces +2. make canonical helper functions in `sw-block` for: + - replica identity + - recovery-target mapping +3. switch duplicate adapter-side mapping sites to consume those helpers +4. leave real `BlockVol`-backed implementations in `weed/` for now +5. only after steps 1-4 are stable, start moving implementation files + +## Execution Form + +This batch is executed through the validate-able tasks in: + +1. `sw-block/design/v2-first-migration-task-pack.md` + +That task pack turns the batch into four parallelizable work items: + +1. canonical assignment translation +2. reader port separation +3. pinner port separation +4. executor muscle separation + +## Why This Batch Is First + +This batch is first because it creates a safe migration destination: + +1. without stable ports, code movement just relocates coupling +2. without canonical helpers, control translation will drift during migration +3. moving execution muscles before shrinking `weed/server` keeps product risk low + +## Exit Condition + +This batch is complete when: + +1. `sw-block` owns the canonical contract layer +2. `weed/` implements that layer without redefining semantics +3. future code moves become mechanical implementation relocation, not + architecture redesign + +## Delivery Note + +This batch is now delivered: + +1. Task A was completed by code change in `a38e04c03` +2. Tasks B/C/D were confirmed already clean by review against the task-pack + acceptance bar diff --git a/sw-block/design/v2-first-migration-task-pack.md b/sw-block/design/v2-first-migration-task-pack.md new file mode 100644 index 000000000..ec5674164 --- /dev/null +++ b/sw-block/design/v2-first-migration-task-pack.md @@ -0,0 +1,308 @@ +# V2 First Migration Task Pack + +Date: 2026-04-04 +Status: delivered + +## Purpose + +This note turns the first separation batch into validate-able engineering tasks. + +Each task must name: + +1. source +2. destination +3. authority rule +4. adapter boundary +5. acceptance criteria +6. validation proof + +The goal is to make separation work parallelizable without letting `V1` +runtime-owner behavior silently leak back in. + +## Shared Rules + +All tasks in this pack inherit these rules: + +1. `sw-block` must not directly import `weed/storage/blockvol` +2. `weed/` may implement ports, but must not redefine semantic truth +3. each task must move one boundary, not redesign the whole runtime +4. compatibility guards may stay, but must not be treated as semantic-authority + proof + +Existing landing zones already exist: + +1. `sw-block/bridge/blockvol` +2. `sw-block/bridge/blockvol/control_adapter.go` +3. `sw-block/bridge/blockvol/contract.go` + +So Task A does not begin with package creation. It begins with canonical-rule +consolidation into the existing `sw-block` bridge layer. + +## Task A: Canonical Assignment Translation + +### Goal + +Make `sw-block` own the canonical helper rules for: + +1. replica identity +2. recovery-target mapping +3. engine replica-assignment packaging + +### Source + +1. `weed/storage/blockvol/v2bridge/control.go` +2. `weed/server/volume_server_block.go` + +### Destination + +1. `sw-block/bridge/blockvol/control_adapter.go` + +### Authority Rule + +This is semantic translation logic, so the canonical rule belongs in +`sw-block`, not in product adapters. + +### Adapter Boundary + +`weed/` may still: + +1. parse `BlockVolumeAssignment` +2. decide which source fields exist on the wire/runtime side + +`weed/` must not separately redefine: + +1. `ReplicaID = /` +2. `replica -> catchup` +3. `rebuilding -> rebuild` + +### Acceptance + +1. `sw-block` exports canonical helpers for identity and recovery-target mapping +2. `weed/storage/blockvol/v2bridge/control.go` and + `weed/server/volume_server_block.go` both use those helpers +3. no direct address-derived identity logic remains in adapter code + +### Validation + +1. `go test ./sw-block/bridge/blockvol` +2. `go test ./weed/storage/blockvol/v2bridge -run "TestControl_|TestBridge_"` +3. focused server path still passes: + - `go test ./weed/server -run "TestBlockService_ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart)"` + +### Current proof anchors + +1. `TestControlAdapter_StableIdentity` +2. `TestControlAdapter_RebuildRoleMapping` +3. `TestControl_PrimaryAssignment_StableServerID` +4. `TestControl_RebuildAssignment` + +## Task B: Reader Port Separation + +### Goal + +Separate retained-history state reading as a pure execution muscle behind a +stable `sw-block` port. + +### Source + +1. `weed/storage/blockvol/v2bridge/reader.go` + +### Destination + +1. contract remains in `sw-block/bridge/blockvol/contract.go` +2. implementation stays thin in `weed/storage/blockvol/v2bridge/reader.go` +3. future code landing zone, if needed: + - `sw-block/bridge/blockvol/runtime` + - or equivalent execution package under `sw-block` + +### Authority Rule + +Reader logic is not semantic authority. It must only read backend facts and +project them into the engine-facing retained-history shape. + +### Adapter Boundary + +`weed/` may: + +1. read `BlockVol.StatusSnapshot()` +2. map backend fields into the contract shape + +`weed/` must not: + +1. reinterpret durability meaning +2. patch semantic fallbacks into the reader + +### Acceptance + +1. `BlockVolReader` contract stays complete and stable in `sw-block` +2. `Reader` remains a thin adapter over real `BlockVol` +3. `StorageAdapter.GetRetainedHistory()` depends only on the contract, not on + weed internals + +### Validation + +1. `go test ./sw-block/bridge/blockvol` +2. `go test ./weed/storage/blockvol/v2bridge -run "TestReader_"` + +### Current proof anchors + +1. `TestStorageAdapter_RetainedHistoryFromReader` +2. `TestReader_RealBlockVol_StatusSnapshot` +3. `TestReader_RealBlockVol_HeadAdvancesWithWrites` + +## Task C: Pinner Port Separation + +### Goal + +Separate WAL/snapshot/full-base hold mechanics as execution muscles behind a +stable `sw-block` pinning port. + +### Source + +1. `weed/storage/blockvol/v2bridge/pinner.go` + +### Destination + +1. contract remains in `sw-block/bridge/blockvol/contract.go` +2. implementation stays thin in `weed/storage/blockvol/v2bridge/pinner.go` +3. future migration target is a `sw-block`-owned execution-muscle package, with + weed-side `BlockVol` binding left thin + +### Authority Rule + +Hold/release mechanics are execution detail. Recovery policy decides *when* to +hold; pinner only decides *how* to pin in the backend. + +### Adapter Boundary + +`weed/` may: + +1. wire retention floor into `BlockVol` +2. validate concrete hold positions against backend state + +`weed/` must not: + +1. decide recovery target +2. redefine which boundary is authoritative + +### Acceptance + +1. `BlockVolPinner` is the sole engine-facing pin contract +2. pinner implementation remains backend-thin and side-effect-local +3. pin lifecycle symmetry is covered in `sw-block` contract tests + +### Validation + +1. `go test ./sw-block/bridge/blockvol` +2. `go test ./weed/storage/blockvol/v2bridge -run "TestPinner_|TestBridge_"` + +### Current proof anchors + +1. `TestStorageAdapter_WALPinRejectsRecycled` +2. `TestStorageAdapter_SnapshotPinRejectsUntrusted` +3. `TestStorageAdapter_PinReleaseSymmetry` +4. `TestPinner_RealBlockVol_HoldWALRetention` +5. `TestPinner_RealBlockVol_HoldRejectsRecycled` + +## Task D: Executor Muscle Separation + +### Goal + +Separate catch-up / rebuild execution mechanics from `weed/` runtime ownership +so that executor behavior is treated as a reusable muscle behind +`sw-block`-owned ports. + +### Source + +1. `weed/storage/blockvol/v2bridge/executor.go` +2. related tests in `weed/storage/blockvol/v2bridge/*transfer*` +3. related tests in `weed/storage/blockvol/v2bridge/*snapshot*` +4. related tests in `weed/storage/blockvol/v2bridge/*truncate*` + +### Destination + +1. contract shape in `sw-block/bridge/blockvol/contract.go` +2. engine-facing use through: + - `engine.CatchUpIO` + - `engine.RebuildIO` +3. implementation remains thin in `weed/` until backend-binding interfaces are + fully extracted + +### Authority Rule + +Executor code is allowed to: + +1. transfer bytes +2. apply WAL entries +3. install snapshots/full base +4. truncate local WAL + +Executor code is not allowed to: + +1. classify recovery outcome +2. decide whether rebuild vs catch-up is needed +3. own publication or health meaning + +### Adapter Boundary + +`weed/` may: + +1. call real `BlockVol` APIs +2. speak TCP rebuild/catch-up protocol +3. update local backend runtime state during execution + +`weed/` must not: + +1. redefine engine recovery phases +2. redefine target/achieved boundary meaning + +### Acceptance + +1. `BlockVolExecutor` aligns exactly with engine execution port expectations +2. engine/executor integration is possible without `sw-block` importing weed +3. executor logic is documented as reusable execution muscle, not semantic + authority + +### Validation + +1. `go test ./sw-block/bridge/blockvol` +2. `go test ./weed/storage/blockvol/v2bridge -run "TestExecutor_|TestBridge_"` +3. focused integrated runtime tests remain green: + - `go test ./weed/server -run "TestBlockService_ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart)"` + +### Current proof anchors + +1. `TestContract_BlockVolReaderInterface` +2. `TestExecutor_RealBlockVol_StreamWALEntries` +3. `TestExecutor_RealBlockVol_StreamPartialRange` +4. `TestExecutor_ErrorPaths` + +## Parallel Execution Recommendation + +These four tasks are safe to run in parallel if ownership stays clear: + +1. Task A: canonical translation rules +2. Task B: reader port hardening +3. Task C: pinner port hardening +4. Task D: executor contract alignment + +Recommended order for merge: + +1. Task A +2. Task B +3. Task C +4. Task D + +Reason: + +1. Task A removes semantic drift first +2. Tasks B/C/D then migrate pure muscles behind that stable rule layer + +## Delivery Note + +Final outcome: + +1. Task A required code change and is now delivered +2. Tasks B/C/D were reviewed and confirmed already at the acceptance bar +3. the next migration frontier is backend-binding extraction, not more contract + cleanup diff --git a/sw-block/design/v2-legacy-runtime-exit-criteria.md b/sw-block/design/v2-legacy-runtime-exit-criteria.md new file mode 100644 index 000000000..110244865 --- /dev/null +++ b/sw-block/design/v2-legacy-runtime-exit-criteria.md @@ -0,0 +1,99 @@ +# V2 Legacy Runtime Exit Criteria + +Date: 2026-04-04 +Status: active + +## Purpose + +This note defines when legacy runtime-owner paths may be downgraded from +required compatibility coverage to removable implementation history. + +Current legacy examples: + +1. `legacy P4` live-path proofs +2. no-core startup paths +3. `HandleAssignmentResult()`-driven recovery startup kept for compatibility + +## Current Position + +For the current phase, legacy paths must remain: + +1. as compatibility guards +2. as regression protection for no-core behavior +3. but NOT as semantic authority proof for the core-present path + +## Exit Criteria + +A legacy runtime-owner path may be downgraded or removed only when all of the +following are true. + +### 1. V2-native proof replacement exists + +There must be core-present proofs covering the same behavior category: + +1. assignment entry ownership +2. task startup ownership +3. execution ownership +4. observation return ownership +5. outward surface consistency + +### 2. Compatibility mode is no longer required operationally + +At least one of these must be true: + +1. production startup always wires `v2Core` +2. no-core path is explicitly declared unsupported +3. no remaining product surface depends on no-core runtime startup + +### 3. The legacy path is no longer the only guard for a runtime mechanic + +Examples: + +1. serialized replacement/drain behavior +2. shutdown drain behavior +3. live plan-to-execute behavior + +These must have equivalent core-present coverage before legacy deletion. + +### 4. No semantic truth still depends on legacy behavior + +Specifically, removing the legacy path must not change: + +1. identity meaning +2. recovery classification +3. publication meaning +4. durable-boundary meaning + +If removal changes any of those, the legacy path was still hiding semantic +authority and cannot be retired yet. + +## Downgrade Stages + +Legacy paths should retire in stages: + +### Stage 1: authority downgrade + +1. keep tests +2. explicitly classify them as compatibility-only + +### Stage 2: runtime fallback downgrade + +1. keep fallback code only where product startup still needs it +2. stop expanding proof claims from those paths + +### Stage 3: deletion candidate + +1. delete tests or move them to legacy-only coverage +2. remove runtime fallback code only after the new path is already the sole + supported owner + +## Current Judgment + +As of the current separation work: + +1. `legacy P4` stays +2. it is already downgraded to compatibility guard +3. it is not yet removable because: + - no-core behavior still exists + - full runtime-loop closure is not yet complete + - not every old ownership proof has a complete core-present replacement diff --git a/sw-block/design/v2-second-migration-batch.md b/sw-block/design/v2-second-migration-batch.md new file mode 100644 index 000000000..9899be9ba --- /dev/null +++ b/sw-block/design/v2-second-migration-batch.md @@ -0,0 +1,109 @@ +# V2 Second Migration Batch + +Date: 2026-04-04 +Status: delivered + +## Purpose + +This note defines the second migration batch for the `sw-block` separation +work. + +The first batch established contract ownership and canonical translation in +`sw-block`. The second batch starts the next frontier: backend-binding +extraction. + +## Batch Goal + +Separate reusable execution-muscle logic from concrete `BlockVol` bindings so +that more code can physically move toward `sw-block` without importing +`weed/storage/blockvol`. + +## Batch Scope + +### In scope + +1. reader backend-binding extraction +2. pinner backend-binding extraction +3. executor backend capability extraction +4. recovery-side shim reduction where those bindings are still copied manually + +### Out of scope + +1. moving raw `BlockVol` backend code into `sw-block` +2. moving `weed/server/block_recovery.go` whole +3. redesigning the rebuild TCP protocol +4. changing engine semantics or recovery policy + +## Current Boundary Problem + +After the first batch, the ownership split is better, but the reusable logic is +still physically stuck next to `BlockVol` because: + +1. `weed/storage/blockvol/v2bridge/reader.go` reads `BlockVol` directly +2. `weed/storage/blockvol/v2bridge/pinner.go` mixes hold bookkeeping with + concrete retention-floor wiring +3. `weed/storage/blockvol/v2bridge/executor.go` mixes reusable recovery steps + with concrete backend calls +4. `weed/server/block_recovery.go` still contains reader/pinner shims that copy + contract shapes manually + +## Target Package Shape + +Recommended landing zone inside `sw-block`: + +1. keep pure contracts in `sw-block/bridge/blockvol` +2. allow a new execution-oriented package for reusable muscle logic: + `sw-block/bridge/blockvol/runtime` + +Weed-side code should shrink toward: + +1. thin `BlockVol` binding +2. runtime hosting +3. network/wire adaptation + +## Concrete Batch Steps + +1. extract reader logic so `weed/` only fetches backend snapshot data +2. extract pinner hold bookkeeping so `weed/` only performs concrete retention + binding and state checks +3. extract executor-facing backend capabilities so reusable orchestration no + longer depends on direct `BlockVol` imports +4. remove redundant reader/pinner contract-shape shims from + `weed/server/block_recovery.go` where the new extracted layer makes them + unnecessary + +## Execution Form + +This batch is executed through the validate-able tasks in: + +1. `sw-block/design/v2-second-migration-task-pack.md` + +## Why This Batch Is Second + +This batch comes second because the first batch had to finish first: + +1. backend-binding extraction is unsafe until contracts and canonical rules are + stable +2. after Batch 1, the remaining coupling is mostly physical implementation + coupling, not semantic drift +3. shrinking `weed/server` only becomes meaningful once `weed/storage/...` + stops owning reusable muscle logic + +## Exit Condition + +This batch is complete when: + +1. reusable reader/pinner/executor logic can live in `sw-block` without direct + `weed/storage/blockvol` imports +2. weed-side files are reduced to thin backend bindings and runtime hosting +3. recovery-side manual shims are either removed or reduced to trivial wiring + +## Delivery Note + +This batch is now delivered: + +1. Task E removed reader contract-shape shimming and made `v2bridge.Reader` + return the bridge contract directly +2. Task F removed the pinner shim from `weed/server/block_recovery.go` +3. Task G was reviewed and confirmed already clean because `v2bridge.Executor` + already satisfies the engine IO interfaces directly diff --git a/sw-block/design/v2-second-migration-task-pack.md b/sw-block/design/v2-second-migration-task-pack.md new file mode 100644 index 000000000..df1c999f2 --- /dev/null +++ b/sw-block/design/v2-second-migration-task-pack.md @@ -0,0 +1,230 @@ +# V2 Second Migration Task Pack + +Date: 2026-04-04 +Status: delivered + +## Purpose + +This note turns the second separation batch into validate-able engineering +tasks. + +The first batch proved that contract ownership and translation authority already +belong in `sw-block`. The second batch now targets the remaining physical +coupling: backend bindings. + +## Shared Rules + +All tasks in this pack inherit these rules: + +1. `sw-block` must not directly import `weed/storage/blockvol` +2. reusable execution-muscle logic should move toward `sw-block` +3. weed-side code should shrink toward thin concrete bindings +4. no task in this pack may redefine engine semantics or recovery policy + +## Task E: Reader Backend-Binding Extraction + +### Goal + +Extract reusable reader logic from direct `BlockVol` coupling so the +`BlockVol`-specific part becomes a thin snapshot binding. + +### Source + +1. `weed/storage/blockvol/v2bridge/reader.go` +2. `weed/server/block_recovery.go` reader shim + +### Destination + +1. reusable reader logic in `sw-block/bridge/blockvol/runtime` +2. thin `BlockVol` snapshot binding in `weed/storage/blockvol/v2bridge` + +### Authority Rule + +The reusable logic that shapes backend snapshot data into +`bridge.BlockVolState` belongs with `sw-block` execution muscles. + +The weed-side binder may fetch snapshot fields from real `BlockVol`, but it +must not own the reusable state-shaping layer. + +### Adapter Boundary + +`weed/` may: + +1. call `StatusSnapshot()` on real `BlockVol` +2. expose raw backend snapshot data to the extracted layer + +`weed/` must not: + +1. keep a second contract-shape mapping layer in `block_recovery.go` +2. reinterpret retained-history meaning + +### Acceptance + +1. reusable reader logic no longer depends on direct `BlockVol` import +2. `weed/storage/blockvol/v2bridge/reader.go` is reduced to thin binding code +3. `readerShimForRecovery` is removed or reduced to trivial wiring + +### Validation + +1. `go test ./sw-block/bridge/blockvol` +2. `go test ./weed/storage/blockvol/v2bridge -run "TestReader_"` +3. if the recovery shim changes, run: + - `go test ./weed/server -run "TestP4_|TestP16B_"` + +### Current proof anchors + +1. `TestStorageAdapter_RetainedHistoryFromReader` +2. `TestReader_RealBlockVol_StatusSnapshot` +3. `TestReader_RealBlockVol_HeadAdvancesWithWrites` + +## Task F: Pinner Backend-Binding Extraction + +### Goal + +Extract hold bookkeeping and release lifecycle from direct `BlockVol` coupling +so weed-side code only performs concrete retention-floor binding and state +validation. + +### Source + +1. `weed/storage/blockvol/v2bridge/pinner.go` +2. `weed/server/block_recovery.go` pinner shim + +### Destination + +1. reusable hold bookkeeping in `sw-block/bridge/blockvol/runtime` +2. thin `BlockVol` retention binding in `weed/storage/blockvol/v2bridge` + +### Authority Rule + +Hold bookkeeping is reusable execution-muscle logic. Concrete interaction with +the flusher and `StatusSnapshot()` stays in `weed/`, but ID tracking and release +symmetry should not require direct `BlockVol` imports. + +### Adapter Boundary + +`weed/` may: + +1. install retention-floor callbacks on real `BlockVol` +2. validate requested hold positions against live backend snapshot state + +`weed/` must not: + +1. keep reusable hold lifecycle ownership trapped in `weed/` +2. force recovery policy knowledge into the pinner binding + +### Acceptance + +1. reusable hold bookkeeping can live in `sw-block` without `BlockVol` imports +2. weed-side pinner code shrinks toward concrete callback/state binding +3. `pinnerShimForRecovery` is removed or reduced to trivial wiring + +### Validation + +1. `go test ./sw-block/bridge/blockvol` +2. `go test ./weed/storage/blockvol/v2bridge -run "TestPinner_|TestBridge_"` +3. if the recovery shim changes, run: + - `go test ./weed/server -run "TestP4_|TestP16B_"` + +### Current proof anchors + +1. `TestStorageAdapter_WALPinRejectsRecycled` +2. `TestStorageAdapter_SnapshotPinRejectsUntrusted` +3. `TestStorageAdapter_PinReleaseSymmetry` +4. `TestPinner_RealBlockVol_HoldWALRetention` +5. `TestPinner_RealBlockVol_HoldRejectsRecycled` + +## Task G: Executor Backend-Capability Extraction + +### Goal + +Split executor logic into: + +1. reusable orchestration that belongs with `sw-block` execution muscles +2. concrete backend capabilities and wire operations that remain in `weed/` + +### Source + +1. `weed/storage/blockvol/v2bridge/executor.go` +2. related tests in: + - `weed/storage/blockvol/v2bridge/*transfer*` + - `weed/storage/blockvol/v2bridge/*snapshot*` + - `weed/storage/blockvol/v2bridge/*truncate*` + +### Destination + +1. reusable executor orchestration in `sw-block/bridge/blockvol/runtime` +2. thin backend capability bindings in `weed/storage/blockvol/v2bridge` + +### Authority Rule + +The engine still owns recovery policy. This task does not move policy. + +The reusable execution sequence for: + +1. bounded WAL replay +2. full-base install plus second catch-up +3. snapshot transfer verification +4. truncate escalation boundary + +should no longer be inseparable from direct `BlockVol` imports. + +### Adapter Boundary + +`weed/` may: + +1. implement concrete backend operations on real `BlockVol` +2. own rebuild TCP framing and network transport while it still depends on + `blockvol` protocol types + +`weed/` must not: + +1. keep the whole recovery step orchestration trapped behind direct + `BlockVol` imports when capability interfaces can be extracted +2. redefine engine-visible boundary meaning + +### Acceptance + +1. executor reusable logic depends on extracted capability interfaces, not + direct `BlockVol` imports +2. weed-side executor code is reduced to concrete backend/network bindings +3. outcome classification still remains outside the executor layer + +### Validation + +1. `go test ./sw-block/bridge/blockvol` +2. `go test ./weed/storage/blockvol/v2bridge -run "TestExecutor_|TestBridge_"` +3. focused runtime integration still passes: + - `go test ./weed/server -run "TestBlockService_ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart)"` + +### Current proof anchors + +1. `TestContract_BlockVolReaderInterface` +2. `TestExecutor_RealBlockVol_StreamWALEntries` +3. `TestExecutor_RealBlockVol_StreamPartialRange` +4. `TestExecutor_ErrorPaths` + +## Recommended Execution Order + +Recommended order: + +1. Task E +2. Task F +3. Task G + +Reason: + +1. reader extraction is lowest risk and pure read-path +2. pinner extraction adds lifecycle but still avoids policy +3. executor extraction is the largest surface and should build on the previous + two cuts + +## Delivery Note + +Final outcome: + +1. Task E was completed by code change +2. Task F was completed by code change +3. Task G was reviewed and confirmed already clean +4. after Batch 2, `weed/server/block_recovery.go` no longer carries + reader/pinner shim types diff --git a/sw-block/design/v2-separation-port-layer-audit.md b/sw-block/design/v2-separation-port-layer-audit.md new file mode 100644 index 000000000..c988829e0 --- /dev/null +++ b/sw-block/design/v2-separation-port-layer-audit.md @@ -0,0 +1,135 @@ +# V2 Separation Port Layer Audit + +Date: 2026-04-04 +Status: active + +## Purpose + +This note audits the current `sw-block` port layer for the separation effort: + +1. define which contracts already belong in `sw-block` +2. identify what was still underspecified or mismatched +3. record the normalized boundary for future migration batches + +## Current Port Layer + +The current reusable boundary inside `sw-block` is: + +1. `sw-block/bridge/blockvol/contract.go` +2. `sw-block/bridge/blockvol/storage_adapter.go` +3. `sw-block/bridge/blockvol/control_adapter.go` + +These files are the intended weed-free bridge between: + +1. `sw-block/engine/replication` +2. `weed/storage/blockvol/v2bridge` +3. `weed/server/*` adapter code + +## Audited Contracts + +### Storage state port + +File: + +1. `sw-block/bridge/blockvol/contract.go` + +Stable contract: + +1. `BlockVolReader` +2. `BlockVolState` + +This is already the right ownership: + +1. `sw-block` owns the shape of retained-history inputs +2. `weed/` only implements how to read those facts from real `BlockVol` + +### Retention / snapshot pinning port + +File: + +1. `sw-block/bridge/blockvol/contract.go` + +Stable contract: + +1. `BlockVolPinner` + +This remains correct because: + +1. pin lifecycle meaning belongs to the V2 recovery driver +2. actual hold/release mechanics remain weed-side implementation detail + +### Recovery execution port + +Previous issue: + +1. `BlockVolExecutor` in `contract.go` did not match the real engine execution + interfaces precisely +2. in particular, rebuild full-base transfer in the engine returns achieved LSN, + but the contract only returned `error` + +Normalized decision: + +1. `sw-block` now names: + - `BlockVolCatchUpIO` + - `BlockVolRebuildIO` + - `BlockVolExecutor` +2. these contracts intentionally match: + - `engine.CatchUpIO` + - `engine.RebuildIO` + +This is the right long-term boundary because: + +1. `sw-block` owns the execution port shape +2. `weed/storage/blockvol/v2bridge.Executor` remains only one implementation +3. future migration can move execution code without changing engine contracts + +### Assignment translation helper port + +Normalized helper layer: + +1. `ReplicaAssignmentForServer()` +2. `RecoveryTargetForRole()` + +These are now the canonical helper rules in: + +1. `sw-block/bridge/blockvol/control_adapter.go` + +They exist to stop identity / recovery-target mapping from drifting between: + +1. `weed/storage/blockvol/v2bridge/control.go` +2. `weed/server/volume_server_block.go` + +## Code Normalization Completed + +Implemented in this batch: + +1. `sw-block/bridge/blockvol/doc.go` + - clarified that the package owns weed-free contracts and thin adapters, + not real blockvol implementations +2. `sw-block/bridge/blockvol/contract.go` + - aligned execution contracts with engine IO interfaces +3. `sw-block/bridge/blockvol/control_adapter.go` + - extracted canonical helper functions for identity and recovery-target + mapping +4. `sw-block/bridge/blockvol/bridge_test.go` + - added interface-compatibility proof for the normalized execution contracts + +## Resulting Boundary Rule + +After this audit, the port layer rule is: + +1. `sw-block` defines contracts and canonical mapping helpers +2. `weed/` implements real storage, transport, and runtime bindings +3. no `sw-block` package in this layer should import `weed/` + +## What Still Does Not Move Yet + +This audit does NOT move: + +1. `weed/storage/blockvol/v2bridge.Executor` +2. `weed/storage/blockvol/v2bridge.Reader` +3. `weed/storage/blockvol/v2bridge.Pinner` +4. `weed/server/BlockService` +5. `weed/server/RecoveryManager` + +It only stabilizes the port layer those migrations will target. diff --git a/sw-block/design/v2-third-migration-batch.md b/sw-block/design/v2-third-migration-batch.md new file mode 100644 index 000000000..f74d36a33 --- /dev/null +++ b/sw-block/design/v2-third-migration-batch.md @@ -0,0 +1,103 @@ +# V2 Third Migration Batch + +Date: 2026-04-04 +Status: active + +## Purpose + +This note defines the third migration batch for the `sw-block` separation work. + +Batch 1 stabilized contract ownership and canonical translation. +Batch 2 removed backend-binding shims and confirmed thin `v2bridge` +implementations. +Batch 3 now targets the remaining runtime-owner concentration in +`weed/server/block_recovery.go`. + +## Batch Goal + +Reduce `weed/server/block_recovery.go` to a host shell that: + +1. owns goroutine lifecycle +2. owns concrete server/block-store access +3. delegates reusable recovery coordination to `sw-block`-owned helpers + +## Batch Scope + +### In scope + +1. pending recovery execution coordination +2. catch-up/rebuild plan execution helper extraction +3. rebuild completion observation shaping +4. explicit isolation of legacy no-core startup behavior + +### Out of scope + +1. moving the full `RecoveryManager` out of `weed/server` +2. changing core command semantics +3. removing `legacy P4` or no-core paths prematurely +4. redesigning block-store access or sender registry ownership + +## Current Boundary Problem + +After Batch 2, `Reader`, `Pinner`, and `Executor` are thinner, but +`weed/server/block_recovery.go` still owns several reusable layers at once: + +1. task host lifecycle +2. pending execution cache and mismatch cancellation +3. catch-up/rebuild execution helper wiring +4. rebuild completion shaping into core events +5. legacy no-core startup compatibility + +That keeps too much reusable coordination trapped in the product adapter shell. + +## Target Package Shape + +Recommended split: + +1. keep host lifecycle in `weed/server` +2. allow reusable recovery coordination helpers in + `sw-block/engine/replication/runtime` +3. keep concrete `BlockVol` access and server integration in `weed/` + +Reason: + +1. pending execution and plan completion shaping are engine-oriented, not + backend-specific +2. those helpers should not require `weed/server` ownership just to exist + +## Concrete Batch Steps + +1. extract pending execution coordination into reusable runtime helpers +2. extract catch-up/rebuild execution helper logic so `weed/server` only + supplies IO bindings and host callbacks +3. extract rebuild completion observation shaping so `weed/server` only reads + backend facts and forwards them +4. isolate no-core startup compatibility behind explicit legacy-only entry + points + +## Execution Form + +This batch is executed through the validate-able tasks in: + +1. `sw-block/design/v2-third-migration-task-pack.md` + +## Why This Batch Is Third + +This batch comes third because: + +1. runtime-host thinning only becomes clear after the backend-binding layer is + already reduced +2. otherwise `block_recovery.go` would still be compensating for low-level shim + coupling +3. the remaining work is now primarily coordination extraction, not contract + cleanup + +## Exit Condition + +This batch is complete when: + +1. `weed/server/block_recovery.go` is mostly host wiring and concrete backend + access +2. reusable pending-execution and completion-shaping logic no longer requires + product adapter ownership +3. legacy no-core startup behavior is clearly isolated as compatibility-only diff --git a/sw-block/design/v2-third-migration-task-pack.md b/sw-block/design/v2-third-migration-task-pack.md new file mode 100644 index 000000000..705dc8a18 --- /dev/null +++ b/sw-block/design/v2-third-migration-task-pack.md @@ -0,0 +1,218 @@ +# V2 Third Migration Task Pack + +Date: 2026-04-04 +Status: active + +## Purpose + +This note turns the third separation batch into validate-able engineering +tasks. + +The key remaining concentration is no longer in `v2bridge`, but in +`weed/server/block_recovery.go`, where host wiring and reusable recovery +coordination still live together. + +## Shared Rules + +All tasks in this pack inherit these rules: + +1. `weed/server` should remain the runtime host shell +2. reusable coordination should move toward `sw-block` +3. legacy no-core support may stay, but only as compatibility-only logic +4. no task in this pack may redefine recovery policy or core semantics + +## Task H: Pending Execution Coordinator Extraction + +### Goal + +Extract pending-execution caching and fail-closed command matching from +`weed/server/block_recovery.go` into reusable runtime helpers. + +### Source + +1. `weed/server/block_recovery.go` +2. `pendingRecoveryExecution` +3. `storePendingExecution` +4. `takePendingExecution` +5. `peekPendingExecution` +6. `hasPendingExecution` +7. `cancelPendingExecution` +8. `ExecutePendingCatchUp` +9. `ExecutePendingRebuild` + +### Destination + +1. reusable coordinator helpers in `sw-block/engine/replication/runtime` +2. thin host-side wiring in `weed/server/block_recovery.go` + +### Authority Rule + +Pending execution ownership is runtime coordination logic. It belongs closer to +the engine/runtime boundary than to the product adapter shell. + +The host shell may store concrete handles, but it should not own the reusable +matching/cancelation semantics. + +### Adapter Boundary + +`weed/` may: + +1. supply concrete volume IDs, replica IDs, and IO bindings +2. trigger coordinator actions from host callbacks + +`weed/` must not: + +1. keep the only implementation of fail-closed pending command matching +2. duplicate target-mismatch cancellation logic in multiple host sites + +### Acceptance + +1. pending execution matching/cancelation logic is reusable outside + `weed/server` +2. `weed/server/block_recovery.go` shrinks to host-side calls into that helper +3. fail-closed mismatch behavior remains explicit and covered + +### Validation + +1. `go test ./sw-block/engine/replication/...` +2. `go test ./weed/server -run "TestP16B_|TestP4_"` + +### Current proof anchors + +1. `TestP16B_RunCatchUp_EscalatesNeedsRebuildIntoCoreProjection` +2. `TestP16B_RunRebuild_FailClosedWithoutFreshStartRebuildCommand` +3. `TestP4_LivePath_RealVol_ReachesPlan` + +## Task I: Recovery Execution Helper Extraction + +### Goal + +Extract reusable catch-up/rebuild plan execution helpers so `weed/server` +supplies only: + +1. concrete IO bindings +2. host callbacks +3. logging/context shell + +### Source + +1. `weed/server/block_recovery.go` +2. `runCatchUp` +3. `runRebuild` +4. `executeCatchUpPlan` +5. `executeRebuildPlan` + +### Destination + +1. reusable execution helpers in `sw-block/engine/replication/runtime` +2. thin host-side volume/session access in `weed/server` + +### Authority Rule + +The engine still decides plan outcome. This task does not move policy. + +What moves is the reusable execution-path coordination that applies an existing +plan using supplied IO and emits the corresponding completion callbacks. + +### Adapter Boundary + +`weed/` may: + +1. fetch real `BlockVol` and build concrete `Reader` / `Pinner` / `Executor` +2. look up sender/session state +3. host goroutines and cancellation contexts + +`weed/` must not: + +1. remain the sole owner of reusable catch-up/rebuild execution wiring +2. mix host concerns and execution-helper concerns in one large function + +### Acceptance + +1. reusable execution helper logic no longer requires `weed/server` ownership +2. `runCatchUp` and `runRebuild` become noticeably smaller host-shell methods +3. catch-up and rebuild still preserve the current bounded command-driven path + +### Validation + +1. `go test ./sw-block/engine/replication/...` +2. `go test ./weed/server -run "TestP16B_|TestP4_"` +3. `go test ./weed/server -run "TestBlockService_ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart)"` + +### Current proof anchors + +1. `TestP16B_RunCatchUp_UpdatesCoreProjectionFromLiveRecovery` +2. `TestP16B_RunRebuild_UsesCoreStartRebuildCommandOnLivePath` +3. `TestP4_SerializedReplacement_DrainsBeforeStart` +4. `TestP4_ShutdownDrain` + +## Task J: Legacy No-Core Isolation + +### Goal + +Make no-core startup behavior explicitly legacy-scoped so the core-present path +and the compatibility path are structurally separate. + +### Source + +1. `weed/server/block_recovery.go` +2. `HandleAssignmentResult` +3. no-core branches inside `runCatchUp` and `runRebuild` +4. `sw-block/design/v2-legacy-runtime-exit-criteria.md` + +### Destination + +1. explicit legacy-only entry points or helper section in `weed/server` +2. updated design note if the isolation shape needs to be recorded + +### Authority Rule + +Legacy compatibility may remain, but it must stop looking like part of the +mainline runtime owner path. + +### Adapter Boundary + +`weed/` may: + +1. keep no-core compatibility while the product still needs it +2. retain `legacy P4` coverage as compatibility guard + +`weed/` must not: + +1. hide compatibility startup inside the same mainline path used for + core-present ownership +2. let no-core behavior continue to blur the supported owner model + +### Acceptance + +1. no-core startup paths are clearly labeled and structurally separated +2. core-present runtime ownership remains the obvious default path +3. legacy proofs remain compatibility-only and are not strengthened into + semantic-authority claims + +### Validation + +1. `go test ./weed/server -run "TestP4_"` +2. `go test ./weed/server -run "TestP16B_|TestBlockService_ApplyAssignments_"` + +### Current proof anchors + +1. `TestP4_LivePath_RealVol_ReachesPlan` +2. `TestP4_SerializedReplacement_DrainsBeforeStart` +3. `TestP4_ShutdownDrain` +4. `TestBlockService_ApplyAssignments_PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp` +5. `TestBlockService_ApplyAssignments_RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart` + +## Recommended Execution Order + +Recommended order: + +1. Task H +2. Task I +3. Task J + +Reason: + +1. the pending coordinator is the narrowest reusable slice +2. execution helper extraction should build on that coordinator boundary +3. legacy isolation should happen after the mainline path is already cleaner diff --git a/sw-block/engine/replication/command.go b/sw-block/engine/replication/command.go index 2651ac1db..6290e0c91 100644 --- a/sw-block/engine/replication/command.go +++ b/sw-block/engine/replication/command.go @@ -27,6 +27,14 @@ type ConfigureShipperCommand struct { func (ConfigureShipperCommand) commandName() string { return "configure_shipper" } +type StartRecoveryTaskCommand struct { + VolumeID string + ReplicaID string + Kind SessionKind +} + +func (StartRecoveryTaskCommand) commandName() string { return "start_recovery_task" } + type StartCatchUpCommand struct { VolumeID string TargetLSN uint64 diff --git a/sw-block/engine/replication/engine.go b/sw-block/engine/replication/engine.go index 755bfb3f6..7ed9385c5 100644 --- a/sw-block/engine/replication/engine.go +++ b/sw-block/engine/replication/engine.go @@ -270,14 +270,16 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) [] roleChanged := st.Role != ev.Role epochChanged := st.Epoch != ev.Epoch replicasChanged := !sameReplicaAssignments(st.DesiredReplicas, ev.Replicas) + recoveryTargetChanged := st.recoveryTarget != ev.RecoveryTarget st.Epoch = ev.Epoch st.Role = ev.Role st.DesiredReplicas = append([]ReplicaAssignment(nil), ev.Replicas...) + st.recoveryTarget = ev.RecoveryTarget st.Readiness.Assigned = true st.Mode.Authority = RuntimeAuthorityConstrainedV1 - if epochChanged || roleChanged { + if epochChanged || roleChanged || recoveryTargetChanged { st.Readiness.RoleApplied = false } @@ -300,13 +302,16 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) [] st.Readiness.ShipperConnected = false } - if epochChanged || roleChanged || replicasChanged { + if epochChanged || roleChanged || replicasChanged || recoveryTargetChanged { st.degraded = false st.degradeReason = "" st.resetInvalidation() st.Recovery = RecoveryView{Phase: RecoveryIdle} 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 } @@ -333,6 +338,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 + 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 + } return cmds } @@ -368,10 +384,21 @@ func (st *VolumeState) shouldApplyRole() bool { func (st *VolumeState) shouldStartReceiver() bool { return st.Role == RoleReplica && + st.recoveryTarget != SessionRebuild && !st.Readiness.ReceiverReady && st.commands.ReceiverStartEpoch != st.Epoch } +func (st *VolumeState) shouldStartRecoveryTask() bool { + if st.recoveryTarget == "" || len(st.DesiredReplicas) != 1 { + return false + } + replicaID := st.DesiredReplicas[0].ReplicaID + return st.commands.RecoveryTaskEpoch != st.Epoch || + st.commands.RecoveryTaskReplicaID != replicaID || + st.commands.RecoveryTaskKind != st.recoveryTarget +} + func (st *VolumeState) shouldConfigureShipper() bool { return st.Role == RolePrimary && st.hasReplicas() && @@ -405,6 +432,8 @@ func (st *VolumeState) bootstrapReason() string { switch { case !st.Readiness.RoleApplied: return "awaiting_role_apply" + case st.Role == RoleReplica && st.recoveryTarget == SessionRebuild: + return "awaiting_rebuild_start" case st.Role == RoleReplica && !st.Readiness.ReceiverReady: return "awaiting_receiver_ready" case st.Role == RolePrimary && !st.Readiness.ShipperConfigured: diff --git a/sw-block/engine/replication/event.go b/sw-block/engine/replication/event.go index ff4ca2aae..e9b35fecf 100644 --- a/sw-block/engine/replication/event.go +++ b/sw-block/engine/replication/event.go @@ -8,10 +8,11 @@ type Event interface { // AssignmentDelivered carries the desired local role and replica set. type AssignmentDelivered struct { - ID string - Epoch uint64 - Role VolumeRole - Replicas []ReplicaAssignment + ID string + Epoch uint64 + Role VolumeRole + Replicas []ReplicaAssignment + RecoveryTarget SessionKind } func (e AssignmentDelivered) VolumeID() string { return e.ID } diff --git a/sw-block/engine/replication/phase14_command_test.go b/sw-block/engine/replication/phase14_command_test.go index 483699965..033b65394 100644 --- a/sw-block/engine/replication/phase14_command_test.go +++ b/sw-block/engine/replication/phase14_command_test.go @@ -8,9 +8,10 @@ import ( func TestPhase14_CommandSequence_PrimaryAssignmentIsBounded(t *testing.T) { core := NewCoreEngine() ev := AssignmentDelivered{ - ID: "vol-cmd-primary", - Epoch: 1, - Role: RolePrimary, + ID: "vol-cmd-primary", + Epoch: 1, + Role: RolePrimary, + RecoveryTarget: SessionCatchUp, Replicas: []ReplicaAssignment{ {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.10:9333", CtrlAddr: "10.0.0.10:9334", Version: 1}}, }, @@ -20,6 +21,7 @@ func TestPhase14_CommandSequence_PrimaryAssignmentIsBounded(t *testing.T) { assertCommandNames(t, result.Commands, []string{ "apply_role", "configure_shipper", + "start_recovery_task", "publish_projection", }) @@ -131,15 +133,17 @@ func TestPhase14_CommandSequence_CatchUpStartIsBounded(t *testing.T) { core := NewCoreEngine() core.ApplyEvent(AssignmentDelivered{ - ID: "vol-cmd-catchup", - Epoch: 6, - Role: RoleReplica, + ID: "vol-cmd-catchup", + Epoch: 6, + Role: RolePrimary, + RecoveryTarget: SessionCatchUp, Replicas: []ReplicaAssignment{ {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.16:9333", CtrlAddr: "10.0.0.16:9334", Version: 1}}, }, }) core.ApplyEvent(RoleApplied{ID: "vol-cmd-catchup"}) - core.ApplyEvent(ReceiverReadyObserved{ID: "vol-cmd-catchup"}) + core.ApplyEvent(ShipperConfiguredObserved{ID: "vol-cmd-catchup"}) + core.ApplyEvent(ShipperConnectedObserved{ID: "vol-cmd-catchup"}) result := core.ApplyEvent(CatchUpPlanned{ID: "vol-cmd-catchup", TargetLSN: 55}) assertCommandNames(t, result.Commands, []string{ @@ -155,15 +159,14 @@ func TestPhase14_CommandSequence_RebuildStartIsBounded(t *testing.T) { core := NewCoreEngine() core.ApplyEvent(AssignmentDelivered{ - ID: "vol-cmd-rebuild", - Epoch: 7, - Role: RoleReplica, + ID: "vol-cmd-rebuild", + Epoch: 7, + Role: RoleReplica, + RecoveryTarget: SessionRebuild, Replicas: []ReplicaAssignment{ {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.17:9333", CtrlAddr: "10.0.0.17:9334", Version: 1}}, }, }) - core.ApplyEvent(RoleApplied{ID: "vol-cmd-rebuild"}) - core.ApplyEvent(ReceiverReadyObserved{ID: "vol-cmd-rebuild"}) core.ApplyEvent(NeedsRebuildObserved{ID: "vol-cmd-rebuild", Reason: "gap_too_large"}) result := core.ApplyEvent(RebuildStarted{ID: "vol-cmd-rebuild", TargetLSN: 80}) @@ -176,19 +179,40 @@ func TestPhase14_CommandSequence_RebuildStartIsBounded(t *testing.T) { assertCommandNames(t, result.Commands, nil) } +func TestPhase14_CommandSequence_RebuildingAssignmentStartsRecoveryTaskWithoutReceiver(t *testing.T) { + core := NewCoreEngine() + + result := core.ApplyEvent(AssignmentDelivered{ + ID: "vol-cmd-rebuild-assign", + Epoch: 8, + Role: RoleReplica, + RecoveryTarget: SessionRebuild, + Replicas: []ReplicaAssignment{ + {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.20:9333", CtrlAddr: "10.0.0.20:9334", Version: 1}}, + }, + }) + assertCommandNames(t, result.Commands, []string{ + "apply_role", + "start_recovery_task", + "publish_projection", + }) +} + func TestPhase14_CommandSequence_AssignmentChangeAllowsFreshRecoveryStart(t *testing.T) { core := NewCoreEngine() core.ApplyEvent(AssignmentDelivered{ - ID: "vol-cmd-reassign", - Epoch: 1, - Role: RoleReplica, + ID: "vol-cmd-reassign", + Epoch: 1, + Role: RolePrimary, + RecoveryTarget: SessionCatchUp, Replicas: []ReplicaAssignment{ {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.18:9333", CtrlAddr: "10.0.0.18:9334", Version: 1}}, }, }) core.ApplyEvent(RoleApplied{ID: "vol-cmd-reassign"}) - core.ApplyEvent(ReceiverReadyObserved{ID: "vol-cmd-reassign"}) + core.ApplyEvent(ShipperConfiguredObserved{ID: "vol-cmd-reassign"}) + core.ApplyEvent(ShipperConnectedObserved{ID: "vol-cmd-reassign"}) result := core.ApplyEvent(CatchUpPlanned{ID: "vol-cmd-reassign", TargetLSN: 90}) assertCommandNames(t, result.Commands, []string{ @@ -196,14 +220,21 @@ func TestPhase14_CommandSequence_AssignmentChangeAllowsFreshRecoveryStart(t *tes "publish_projection", }) - core.ApplyEvent(AssignmentDelivered{ - ID: "vol-cmd-reassign", - Epoch: 2, - Role: RoleReplica, + result = core.ApplyEvent(AssignmentDelivered{ + ID: "vol-cmd-reassign", + Epoch: 2, + Role: RolePrimary, + RecoveryTarget: SessionCatchUp, Replicas: []ReplicaAssignment{ {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.19:9333", CtrlAddr: "10.0.0.19:9334", Version: 2}}, }, }) + assertCommandNames(t, result.Commands, []string{ + "apply_role", + "configure_shipper", + "start_recovery_task", + "publish_projection", + }) result = core.ApplyEvent(CatchUpPlanned{ID: "vol-cmd-reassign", TargetLSN: 90}) assertCommandNames(t, result.Commands, []string{ diff --git a/sw-block/engine/replication/state.go b/sw-block/engine/replication/state.go index 50a89bff2..7904c2647 100644 --- a/sw-block/engine/replication/state.go +++ b/sw-block/engine/replication/state.go @@ -92,6 +92,9 @@ type commandState struct { ReceiverStartEpoch uint64 ShipperConfigEpoch uint64 ShipperConfigReplicas []ReplicaAssignment + RecoveryTaskEpoch uint64 + RecoveryTaskReplicaID string + RecoveryTaskKind SessionKind CatchUpTargetLSN uint64 RebuildTargetLSN uint64 InvalidationIssued bool @@ -112,11 +115,12 @@ type VolumeState struct { Publication PublicationView Recovery RecoveryView - degraded bool - degradeReason string - needsRebuild bool - rebuildReason string - commands commandState + degraded bool + degradeReason string + needsRebuild bool + rebuildReason string + recoveryTarget SessionKind + commands commandState } func newVolumeState(volumeID string) *VolumeState { diff --git a/weed/server/blockcmd/dispatch.go b/weed/server/blockcmd/dispatch.go new file mode 100644 index 000000000..b39ced611 --- /dev/null +++ b/weed/server/blockcmd/dispatch.go @@ -0,0 +1,139 @@ +package blockcmd + +import ( + "fmt" + + engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +// Ops executes concrete backend-integrated command operations. +// It does not own command dispatch or host-side event semantics. +type Ops interface { + ApplyRole(assignment blockvol.BlockVolumeAssignment) (bool, error) + StartReceiver(assignment blockvol.BlockVolumeAssignment) (bool, error) + 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) +} + +// HostEffects applies server-adapter side effects after concrete command +// execution succeeds. These effects stay out of the backend-binding layer. +type HostEffects interface { + RecordCommand(volumeID, name string) + EmitCoreEvent(ev engine.Event) + PublishProjection(volumeID string, projection engine.PublicationProjection) error +} + +// Dispatcher runs engine commands against concrete ops and host effects. +type Dispatcher struct { + ops Ops + effects HostEffects +} + +func NewDispatcher(ops Ops, effects HostEffects) *Dispatcher { + return &Dispatcher{ops: ops, effects: effects} +} + +func (d *Dispatcher) Run(cmds []engine.Command, assignment *blockvol.BlockVolumeAssignment) error { + for _, cmd := range cmds { + switch v := cmd.(type) { + case engine.ApplyRoleCommand: + a, ok, err := requireAssignment("apply_role", v.VolumeID, assignment) + if err != nil || !ok { + return err + } + executed, err := d.ops.ApplyRole(a) + if err != nil { + return err + } + if executed { + d.effects.RecordCommand(v.VolumeID, "apply_role") + } + case engine.StartReceiverCommand: + a, ok, err := requireAssignment("start_receiver", v.VolumeID, assignment) + if err != nil || !ok { + return err + } + if a.ReplicaDataAddr == "" || a.ReplicaCtrlAddr == "" { + continue + } + executed, err := d.ops.StartReceiver(a) + if err != nil { + return err + } + if executed { + d.effects.RecordCommand(v.VolumeID, "start_receiver") + d.effects.EmitCoreEvent(engine.ReceiverReadyObserved{ID: v.VolumeID}) + } + case engine.ConfigureShipperCommand: + executed, connected, err := d.ops.ConfigureShipper(v.VolumeID, v.Replicas) + if err != nil { + return err + } + if executed { + d.effects.RecordCommand(v.VolumeID, "configure_shipper") + d.effects.EmitCoreEvent(engine.ShipperConfiguredObserved{ID: v.VolumeID}) + if connected { + d.effects.EmitCoreEvent(engine.ShipperConnectedObserved{ID: v.VolumeID}) + } + } + case engine.StartRecoveryTaskCommand: + a, ok, err := requireAssignment("start_recovery_task", v.VolumeID, assignment) + if err != nil || !ok { + return err + } + if v.ReplicaID == "" { + continue + } + executed, err := d.ops.StartRecoveryTask(v.ReplicaID, a) + if err != nil { + return err + } + if executed { + d.effects.RecordCommand(v.VolumeID, "start_recovery_task") + } + case engine.InvalidateSessionCommand: + executed, err := d.ops.InvalidateSession(v.VolumeID, v.Reason) + if err != nil { + return err + } + if executed { + d.effects.RecordCommand(v.VolumeID, "invalidate_session") + } + case engine.StartCatchUpCommand: + executed, err := d.ops.StartCatchUp(v.VolumeID, v.TargetLSN) + if err != nil { + return err + } + if executed { + d.effects.RecordCommand(v.VolumeID, "start_catchup") + } + case engine.StartRebuildCommand: + executed, err := d.ops.StartRebuild(v.VolumeID, v.TargetLSN) + if err != nil { + return err + } + if executed { + d.effects.RecordCommand(v.VolumeID, "start_rebuild") + } + case engine.PublishProjectionCommand: + if err := d.effects.PublishProjection(v.VolumeID, v.Projection); err != nil { + return fmt.Errorf("publish projection %s: %w", v.VolumeID, err) + } + } + } + return nil +} + +func requireAssignment(action, volumeID string, assignment *blockvol.BlockVolumeAssignment) (blockvol.BlockVolumeAssignment, bool, error) { + if assignment == nil { + return blockvol.BlockVolumeAssignment{}, false, nil + } + if assignment.Path != volumeID { + return blockvol.BlockVolumeAssignment{}, false, fmt.Errorf("block service: core %s path mismatch %q != %q", action, assignment.Path, volumeID) + } + return *assignment, true, nil +} diff --git a/weed/server/blockcmd/dispatch_test.go b/weed/server/blockcmd/dispatch_test.go new file mode 100644 index 000000000..9465810c7 --- /dev/null +++ b/weed/server/blockcmd/dispatch_test.go @@ -0,0 +1,186 @@ +package blockcmd + +import ( + "errors" + "reflect" + "testing" + + engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +type fakeOps struct { + applyRoleFn func(blockvol.BlockVolumeAssignment) (bool, error) + startReceiverFn func(blockvol.BlockVolumeAssignment) (bool, error) + configureShipperFn func(string, []engine.ReplicaAssignment) (bool, bool, error) + startRecoveryTaskFn func(string, blockvol.BlockVolumeAssignment) (bool, error) + invalidateSessionFn func(string, string) (bool, error) + startCatchUpFn func(string, uint64) (bool, error) + startRebuildFn func(string, uint64) (bool, error) +} + +func (f fakeOps) ApplyRole(a blockvol.BlockVolumeAssignment) (bool, error) { + if f.applyRoleFn == nil { + return false, nil + } + return f.applyRoleFn(a) +} + +func (f fakeOps) StartReceiver(a blockvol.BlockVolumeAssignment) (bool, error) { + if f.startReceiverFn == nil { + return false, nil + } + return f.startReceiverFn(a) +} + +func (f fakeOps) ConfigureShipper(volumeID string, replicas []engine.ReplicaAssignment) (bool, bool, error) { + if f.configureShipperFn == nil { + return false, false, nil + } + return f.configureShipperFn(volumeID, replicas) +} + +func (f fakeOps) StartRecoveryTask(replicaID string, assignment blockvol.BlockVolumeAssignment) (bool, error) { + if f.startRecoveryTaskFn == nil { + return false, nil + } + return f.startRecoveryTaskFn(replicaID, assignment) +} + +func (f fakeOps) InvalidateSession(volumeID, reason string) (bool, error) { + if f.invalidateSessionFn == nil { + return false, nil + } + return f.invalidateSessionFn(volumeID, reason) +} + +func (f fakeOps) StartCatchUp(volumeID string, targetLSN uint64) (bool, error) { + if f.startCatchUpFn == nil { + return false, nil + } + return f.startCatchUpFn(volumeID, targetLSN) +} + +func (f fakeOps) StartRebuild(volumeID string, targetLSN uint64) (bool, error) { + if f.startRebuildFn == nil { + return false, nil + } + return f.startRebuildFn(volumeID, targetLSN) +} + +type fakeEffects struct { + recorded []string + events []engine.Event + published map[string]engine.PublicationProjection +} + +func (f *fakeEffects) RecordCommand(volumeID, name string) { + f.recorded = append(f.recorded, volumeID+":"+name) +} + +func (f *fakeEffects) EmitCoreEvent(ev engine.Event) { + f.events = append(f.events, ev) +} + +func (f *fakeEffects) PublishProjection(volumeID string, projection engine.PublicationProjection) error { + if f.published == nil { + f.published = make(map[string]engine.PublicationProjection) + } + f.published[volumeID] = projection + return nil +} + +func TestDispatcher_ApplyRoleRequiresMatchingAssignment(t *testing.T) { + d := NewDispatcher(fakeOps{}, &fakeEffects{}) + err := d.Run([]engine.Command{engine.ApplyRoleCommand{VolumeID: "vol1"}}, &blockvol.BlockVolumeAssignment{Path: "other"}) + if err == nil { + t.Fatal("expected path mismatch error") + } +} + +func TestDispatcher_StartReceiverRecordsAndEmitsObserved(t *testing.T) { + effects := &fakeEffects{} + d := NewDispatcher(fakeOps{ + startReceiverFn: func(a blockvol.BlockVolumeAssignment) (bool, error) { + return true, nil + }, + }, effects) + assignment := &blockvol.BlockVolumeAssignment{ + Path: "vol1", + ReplicaDataAddr: "10.0.0.1:9333", + ReplicaCtrlAddr: "10.0.0.1:9334", + } + if err := d.Run([]engine.Command{engine.StartReceiverCommand{VolumeID: "vol1"}}, assignment); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(effects.recorded, []string{"vol1:start_receiver"}) { + t.Fatalf("recorded=%v", effects.recorded) + } + if len(effects.events) != 1 { + t.Fatalf("events=%d", len(effects.events)) + } + if _, ok := effects.events[0].(engine.ReceiverReadyObserved); !ok { + t.Fatalf("event=%T", effects.events[0]) + } +} + +func TestDispatcher_ConfigureShipperKeepsHostEffectsServerSide(t *testing.T) { + effects := &fakeEffects{} + d := NewDispatcher(fakeOps{ + configureShipperFn: func(volumeID string, replicas []engine.ReplicaAssignment) (bool, bool, error) { + return true, true, nil + }, + }, effects) + err := d.Run([]engine.Command{engine.ConfigureShipperCommand{ + VolumeID: "vol1", + Replicas: []engine.ReplicaAssignment{{ReplicaID: "vol1/vs2", Endpoint: engine.Endpoint{DataAddr: "data", CtrlAddr: "ctrl"}}}, + }}, nil) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(effects.recorded, []string{"vol1:configure_shipper"}) { + t.Fatalf("recorded=%v", effects.recorded) + } + if len(effects.events) != 2 { + t.Fatalf("events=%d", len(effects.events)) + } + if _, ok := effects.events[0].(engine.ShipperConfiguredObserved); !ok { + t.Fatalf("event0=%T", effects.events[0]) + } + if _, ok := effects.events[1].(engine.ShipperConnectedObserved); !ok { + t.Fatalf("event1=%T", effects.events[1]) + } +} + +func TestDispatcher_PublishProjectionUsesHostEffect(t *testing.T) { + effects := &fakeEffects{} + d := NewDispatcher(fakeOps{}, effects) + proj := engine.PublicationProjection{VolumeID: "vol1", Role: engine.RolePrimary} + if err := d.Run([]engine.Command{engine.PublishProjectionCommand{VolumeID: "vol1", Projection: proj}}, nil); err != nil { + t.Fatal(err) + } + got, ok := effects.published["vol1"] + if !ok || got.VolumeID != "vol1" || got.Role != engine.RolePrimary { + t.Fatalf("published=%v", effects.published) + } +} + +func TestDispatcher_StopsOnFirstError(t *testing.T) { + effects := &fakeEffects{} + d := NewDispatcher(fakeOps{ + startCatchUpFn: func(volumeID string, targetLSN uint64) (bool, error) { + return false, errors.New("boom") + }, + startRebuildFn: func(volumeID 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}, + }, nil) + if err == nil { + t.Fatal("expected error") + } +} diff --git a/weed/server/volume_server_block.go b/weed/server/volume_server_block.go index 7fca2286c..db7afb806 100644 --- a/weed/server/volume_server_block.go +++ b/weed/server/volume_server_block.go @@ -9,7 +9,9 @@ import ( "strings" "sync" + bridgeblockvol "github.com/seaweedfs/seaweedfs/sw-block/bridge/blockvol" engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" + "github.com/seaweedfs/seaweedfs/weed/server/blockcmd" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" @@ -95,6 +97,9 @@ type BlockService struct { // routable host:port. This is the -ip value (IP or resolvable hostname), // never an opaque server identity from -id. advertisedHost string + + // TestHook: if set, invoked when the legacy direct rebuild starter is used. + onLegacyStartRebuild func(path, rebuildAddr string, epoch uint64) } // V2Orchestrator returns the V2 engine orchestrator for inspection/testing. @@ -467,6 +472,8 @@ func (bs *BlockService) ProcessAssignments(assignments []blockvol.BlockVolumeAss // readiness bookkeeping. Returns per-assignment errors parallel to the input. func (bs *BlockService) ApplyAssignments(assignments []blockvol.BlockVolumeAssignment) []error { errs := make([]error, len(assignments)) + var legacyRecoveryResults []engine.AssignmentResult + var removedRecoveryResults []engine.AssignmentResult // V2 bridge: convert and deliver to engine orchestrator (Phase 08 P1). // P3: skip V2 processing for repeated unchanged assignments. @@ -487,8 +494,14 @@ func (bs *BlockService) ApplyAssignments(assignments []blockvol.BlockVolumeAssig len(result.SessionsCreated)+len(result.SessionsSuperseded)) // P4: drive live recovery execution based on engine result. - if bs.v2Recovery != nil && (len(result.SessionsCreated) > 0 || len(result.SessionsSuperseded) > 0 || len(result.Removed) > 0) { - bs.v2Recovery.HandleAssignmentResult(result, assignments) + if bs.v2Recovery != nil { + if bs.v2Core == nil { + if len(result.SessionsCreated) > 0 || len(result.SessionsSuperseded) > 0 || len(result.Removed) > 0 { + legacyRecoveryResults = append(legacyRecoveryResults, result) + } + } else if len(result.Removed) > 0 { + removedRecoveryResults = append(removedRecoveryResults, result) + } } } } @@ -511,11 +524,19 @@ func (bs *BlockService) ApplyAssignments(assignments []blockvol.BlockVolumeAssig case blockvol.RolePrimary: case blockvol.RoleReplica: case blockvol.RoleRebuilding: - if a.RebuildAddr != "" { + if a.RebuildAddr != "" && bs.v2Core == nil { bs.startRebuild(a.Path, a.RebuildAddr, a.Epoch) } } } + if bs.v2Recovery != nil { + for _, result := range legacyRecoveryResults { + bs.v2Recovery.HandleAssignmentResult(result, assignments) + } + for _, result := range removedRecoveryResults { + bs.v2Recovery.HandleRemovedAssignments(result) + } + } return errs } @@ -544,83 +565,49 @@ func (bs *BlockService) applyCoreCommands(cmds []engine.Command) { } func (bs *BlockService) applyCoreCommandsWithAssignment(cmds []engine.Command, assignment *blockvol.BlockVolumeAssignment) error { - for _, cmd := range cmds { - switch v := cmd.(type) { - case engine.ApplyRoleCommand: - if err := bs.executeApplyRoleCommand(v, assignment); err != nil { - return err - } - case engine.StartReceiverCommand: - if err := bs.executeStartReceiverCommand(v, assignment); err != nil { - return err - } - case engine.ConfigureShipperCommand: - if err := bs.executeConfigureShipperCommand(v); err != nil { - return err - } - case engine.InvalidateSessionCommand: - if err := bs.executeInvalidateSessionCommand(v); err != nil { - return err - } - case engine.StartCatchUpCommand: - if err := bs.executeStartCatchUpCommand(v); err != nil { - return err - } - case engine.StartRebuildCommand: - if err := bs.executeStartRebuildCommand(v); err != nil { - return err - } - case engine.PublishProjectionCommand: - proj := v.Projection - if latest, ok := bs.V2Core().Projection(v.VolumeID); ok { - proj = latest - } - bs.coreProjMu.Lock() - if bs.coreProj == nil { - bs.coreProj = make(map[string]engine.PublicationProjection) - } - bs.coreProj[v.VolumeID] = proj - bs.coreProjMu.Unlock() - } - } - return nil -} - -func (bs *BlockService) executeApplyRoleCommand(cmd engine.ApplyRoleCommand, assignment *blockvol.BlockVolumeAssignment) error { - if assignment == nil { + if bs == nil { return nil } - if assignment.Path != cmd.VolumeID { - return fmt.Errorf("block service: core apply_role path mismatch %q != %q", assignment.Path, cmd.VolumeID) - } - if err := bs.applyRoleAssignment(*assignment); err != nil { - return err - } - bs.recordExecutedCoreCommand(cmd.VolumeID, "apply_role") - return nil + return bs.coreCommandDispatcher().Run(cmds, assignment) } -func (bs *BlockService) executeStartReceiverCommand(cmd engine.StartReceiverCommand, assignment *blockvol.BlockVolumeAssignment) error { - if assignment == nil { - return nil +func (bs *BlockService) coreCommandDispatcher() *blockcmd.Dispatcher { + return blockcmd.NewDispatcher(coreCommandOps{bs: bs}, coreCommandEffects{bs: bs}) +} + +type coreCommandOps struct { + bs *BlockService +} + +func (ops coreCommandOps) ApplyRole(assignment blockvol.BlockVolumeAssignment) (bool, error) { + if ops.bs == nil { + return false, nil } - if assignment.Path != cmd.VolumeID { - return fmt.Errorf("block service: core start_receiver path mismatch %q != %q", assignment.Path, cmd.VolumeID) + if err := ops.bs.applyRoleAssignment(assignment); err != nil { + return false, err + } + return true, nil +} + +func (ops coreCommandOps) StartReceiver(assignment blockvol.BlockVolumeAssignment) (bool, error) { + if ops.bs == nil { + return false, nil } if assignment.ReplicaDataAddr == "" || assignment.ReplicaCtrlAddr == "" { - return nil + return false, nil } - if err := bs.setupReplicaReceiver(assignment.Path, assignment.ReplicaDataAddr, assignment.ReplicaCtrlAddr); err != nil { - return err + if err := ops.bs.setupReplicaReceiver(assignment.Path, assignment.ReplicaDataAddr, assignment.ReplicaCtrlAddr); err != nil { + return false, err } - bs.recordExecutedCoreCommand(cmd.VolumeID, "start_receiver") - bs.applyCoreEvent(engine.ReceiverReadyObserved{ID: cmd.VolumeID}) - return nil + return true, nil } -func (bs *BlockService) executeConfigureShipperCommand(cmd engine.ConfigureShipperCommand) error { - addrs := make([]blockvol.ReplicaAddr, 0, len(cmd.Replicas)) - for _, replica := range cmd.Replicas { +func (ops coreCommandOps) ConfigureShipper(volumeID string, replicas []engine.ReplicaAssignment) (bool, bool, error) { + if ops.bs == nil { + return false, false, nil + } + addrs := make([]blockvol.ReplicaAddr, 0, len(replicas)) + for _, replica := range replicas { if replica.Endpoint.DataAddr == "" || replica.Endpoint.CtrlAddr == "" { continue } @@ -631,63 +618,100 @@ func (bs *BlockService) executeConfigureShipperCommand(cmd engine.ConfigureShipp }) } if len(addrs) == 0 { - return nil + return false, false, nil } if len(addrs) == 1 { - if err := bs.setupPrimaryReplication(cmd.VolumeID, addrs[0].DataAddr, addrs[0].CtrlAddr); err != nil { - return err + if err := ops.bs.setupPrimaryReplication(volumeID, addrs[0].DataAddr, addrs[0].CtrlAddr); err != nil { + return false, false, err } } else { - if err := bs.setupPrimaryReplicationMulti(cmd.VolumeID, addrs); err != nil { - return err + if err := ops.bs.setupPrimaryReplicationMulti(volumeID, addrs); err != nil { + return false, false, err } } - bs.recordExecutedCoreCommand(cmd.VolumeID, "configure_shipper") - bs.applyCoreEvent(engine.ShipperConfiguredObserved{ID: cmd.VolumeID}) - if bs.isPrimaryShipperConnected(cmd.VolumeID) { - bs.applyCoreEvent(engine.ShipperConnectedObserved{ID: cmd.VolumeID}) - } - return nil + return true, ops.bs.isPrimaryShipperConnected(volumeID), nil } -func (bs *BlockService) executeInvalidateSessionCommand(cmd engine.InvalidateSessionCommand) error { - if bs == nil || bs.v2Orchestrator == nil || bs.v2Core == nil { - return nil +func (ops coreCommandOps) StartRecoveryTask(replicaID string, assignment blockvol.BlockVolumeAssignment) (bool, error) { + if ops.bs == nil || ops.bs.v2Recovery == nil { + return false, nil } - proj, ok := bs.v2Core.Projection(cmd.VolumeID) + ops.bs.v2Recovery.StartRecoveryTask(replicaID, []blockvol.BlockVolumeAssignment{assignment}) + return true, nil +} + +func (ops coreCommandOps) InvalidateSession(volumeID, reason string) (bool, error) { + if ops.bs == nil || ops.bs.v2Orchestrator == nil || ops.bs.v2Core == nil { + return false, nil + } + proj, ok := ops.bs.v2Core.Projection(volumeID) if !ok { - return nil + return false, nil } for _, replicaID := range proj.ReplicaIDs { - sender := bs.v2Orchestrator.Registry.Sender(replicaID) + sender := ops.bs.v2Orchestrator.Registry.Sender(replicaID) if sender == nil { continue } - sender.InvalidateSession(cmd.Reason, engine.StateDisconnected) + sender.InvalidateSession(reason, engine.StateDisconnected) } - bs.recordExecutedCoreCommand(cmd.VolumeID, "invalidate_session") - return nil + return true, nil } -func (bs *BlockService) executeStartCatchUpCommand(cmd engine.StartCatchUpCommand) error { - if bs == nil || bs.v2Recovery == nil { - return nil +func (ops coreCommandOps) StartCatchUp(volumeID string, targetLSN uint64) (bool, error) { + if ops.bs == nil || ops.bs.v2Recovery == nil { + return false, nil } - if err := bs.v2Recovery.ExecutePendingCatchUp(cmd.VolumeID, cmd.TargetLSN); err != nil { - return err + if err := ops.bs.v2Recovery.ExecutePendingCatchUp(volumeID, targetLSN); err != nil { + return false, err } - bs.recordExecutedCoreCommand(cmd.VolumeID, "start_catchup") - return nil + return true, nil } -func (bs *BlockService) executeStartRebuildCommand(cmd engine.StartRebuildCommand) error { - if bs == nil || bs.v2Recovery == nil { +func (ops coreCommandOps) StartRebuild(volumeID string, targetLSN uint64) (bool, error) { + if ops.bs == nil || ops.bs.v2Recovery == nil { + return false, nil + } + if err := ops.bs.v2Recovery.ExecutePendingRebuild(volumeID, targetLSN); err != nil { + return false, err + } + return true, nil +} + +type coreCommandEffects struct { + bs *BlockService +} + +func (effects coreCommandEffects) RecordCommand(volumeID, name string) { + if effects.bs == nil { + return + } + effects.bs.recordExecutedCoreCommand(volumeID, name) +} + +func (effects coreCommandEffects) EmitCoreEvent(ev engine.Event) { + if effects.bs == nil { + return + } + effects.bs.applyCoreEvent(ev) +} + +func (effects coreCommandEffects) PublishProjection(volumeID string, projection engine.PublicationProjection) error { + if effects.bs == nil { return nil } - if err := bs.v2Recovery.ExecutePendingRebuild(cmd.VolumeID, cmd.TargetLSN); err != nil { - return err + proj := projection + if core := effects.bs.V2Core(); core != nil { + if latest, ok := core.Projection(volumeID); ok { + proj = latest + } } - bs.recordExecutedCoreCommand(cmd.VolumeID, "start_rebuild") + effects.bs.coreProjMu.Lock() + if effects.bs.coreProj == nil { + effects.bs.coreProj = make(map[string]engine.PublicationProjection) + } + effects.bs.coreProj[volumeID] = proj + effects.bs.coreProjMu.Unlock() return nil } @@ -731,33 +755,42 @@ func (bs *BlockService) coreAssignmentEvent(a blockvol.BlockVolumeAssignment) (e if ra.ServerID == "" { continue } - ev.Replicas = append(ev.Replicas, engine.ReplicaAssignment{ - ReplicaID: fmt.Sprintf("%s/%s", a.Path, ra.ServerID), - Endpoint: engine.Endpoint{ - DataAddr: ra.DataAddr, - CtrlAddr: ra.CtrlAddr, - }, - }) + ev.Replicas = append(ev.Replicas, bridgeblockvol.ReplicaAssignmentForServer(a.Path, ra.ServerID, engine.Endpoint{ + DataAddr: ra.DataAddr, + CtrlAddr: ra.CtrlAddr, + })) + } + if len(ev.Replicas) == 1 { + ev.RecoveryTarget = engine.SessionCatchUp } } else if a.ReplicaServerID != "" && a.ReplicaDataAddr != "" { - ev.Replicas = []engine.ReplicaAssignment{{ - ReplicaID: fmt.Sprintf("%s/%s", a.Path, a.ReplicaServerID), - Endpoint: engine.Endpoint{ + ev.Replicas = []engine.ReplicaAssignment{ + bridgeblockvol.ReplicaAssignmentForServer(a.Path, a.ReplicaServerID, engine.Endpoint{ DataAddr: a.ReplicaDataAddr, CtrlAddr: a.ReplicaCtrlAddr, - }, - }} + }), + } + ev.RecoveryTarget = engine.SessionCatchUp } return ev, true case blockvol.RoleReplica: ev.Role = engine.RoleReplica - ev.Replicas = []engine.ReplicaAssignment{{ - ReplicaID: fmt.Sprintf("%s/%s", a.Path, bs.localServerID), - Endpoint: engine.Endpoint{ + ev.Replicas = []engine.ReplicaAssignment{ + bridgeblockvol.ReplicaAssignmentForServer(a.Path, bs.localServerID, engine.Endpoint{ DataAddr: a.ReplicaDataAddr, CtrlAddr: a.ReplicaCtrlAddr, - }, - }} + }), + } + return ev, true + case blockvol.RoleRebuilding: + ev.Role = engine.RoleReplica + ev.RecoveryTarget = bridgeblockvol.RecoveryTargetForRole("rebuilding") + ev.Replicas = []engine.ReplicaAssignment{ + bridgeblockvol.ReplicaAssignmentForServer(a.Path, bs.localServerID, engine.Endpoint{ + DataAddr: a.ReplicaDataAddr, + CtrlAddr: a.ReplicaCtrlAddr, + }), + } return ev, true default: return engine.AssignmentDelivered{}, false @@ -901,6 +934,9 @@ func (bs *BlockService) setupReplicaReceiver(path, dataAddr, ctrlAddr string) er // Future work: VS could report rebuild completion via heartbeat so master // can update registry state (e.g., promote from Rebuilding to Replica). func (bs *BlockService) startRebuild(path, rebuildAddr string, epoch uint64) { + if bs != nil && bs.onLegacyStartRebuild != nil { + bs.onLegacyStartRebuild(path, rebuildAddr, epoch) + } go func() { vol, ok := bs.blockStore.GetBlockVolume(path) if !ok {