From a6fc8545b9c862f17325434f116807ec1668f063 Mon Sep 17 00:00:00 2001 From: pingqiu Date: Fri, 3 Apr 2026 16:52:55 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=2014A+14B=20=E2=80=94=20V2=20core?= =?UTF-8?q?=20publication=20ownership=20+=20command=20semantics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 14A: Publication as explicit core-owned state - state.go: PublicationView on VolumeState, explicit gate reasons - engine.go: mode→readiness→publication chain with named gates (awaiting_role_apply, awaiting_shipper_configured, awaiting_barrier_durability) - projection.go: PublicationProjection carries publication truth - RF=1/no-replicas → allocated_only (CP13-9 constraint in core) - phase14_core_test.go: strengthened publication closure + RF=1 proof 14B: Command emission bounded by semantic gap - engine.go: repeated same-assignment skips redundant commands, repeated same-reason BarrierRejected skips duplicate invalidation, command-state tracking on VolumeState - command.go: new command types for bounded emission - event.go: new boundary events - phase14_command_test.go: exact command sequences frozen as proofs (primary/replica repeated assignment, assignment changed, repeated failure) - phase14_boundary_test.go: boundary/recovery structural tests All tests pass in sw-block/engine/replication. Phase 14 docs updated (14A accepted, 14B active→14C planned). Co-Authored-By: Claude Opus 4.6 (1M context) --- sw-block/.private/phase/phase-14-log.md | 357 ++++++++++++++++++ sw-block/.private/phase/phase-14.md | 50 ++- sw-block/engine/replication/command.go | 14 + sw-block/engine/replication/engine.go | 140 ++++++- sw-block/engine/replication/event.go | 45 +++ .../replication/phase14_boundary_test.go | 192 ++++++++++ .../replication/phase14_command_test.go | 115 +++++- .../engine/replication/phase14_core_test.go | 16 +- sw-block/engine/replication/projection.go | 4 +- sw-block/engine/replication/state.go | 33 +- 10 files changed, 919 insertions(+), 47 deletions(-) create mode 100644 sw-block/engine/replication/phase14_boundary_test.go diff --git a/sw-block/.private/phase/phase-14-log.md b/sw-block/.private/phase/phase-14-log.md index ecb3970b7..1c9334906 100644 --- a/sw-block/.private/phase/phase-14-log.md +++ b/sw-block/.private/phase/phase-14-log.md @@ -350,3 +350,360 @@ Constraint / overclaim / proof review: - `14A` publication-gate proofs still hold - barrier durability is still the only path to `publish_healthy` - constrained-`V1` interpretation is still explicit, not broadened + +--- + +### `14B` Delivery Note Rev 2 + +Date: 2026-04-03 +Scope: tighten `publish_projection` so it is also emitted from semantic change, +not from raw event frequency + +What changed: + +1. `PublishProjectionCommand` is now emitted only when the outward projection + actually changes +2. repeated identical events on an already-converged state now become true + no-op command sequences + +Files changed: + +1. `sw-block/engine/replication/engine.go` + - compare previous and new projection + - emit `publish_projection` only on real outward change +2. `sw-block/engine/replication/phase14_command_test.go` + - repeated identical primary assignment now expects no commands + - repeated identical replica assignment now expects no commands + - repeated identical failure now expects no commands after the first + invalidation + - added direct proof that repeated unchanged projection events emit no + `publish_projection` + +Proofs strengthened: + +1. repeated identical assignment is now a true no-op command sequence +2. repeated identical failure is now a true no-op command sequence +3. publish emission is now tied to projection change, not event arrival + +Validation: + +1. `gofmt -w engine.go phase14_command_test.go` +2. `go test ./...` +3. result: `PASS` + +Constraint / overclaim / proof review: + +1. semantic constraint satisfied + - `14B`: command emission is further frozen to semantic deltas only +2. overclaim avoided + - repeated identical events no longer look like fresh publication work + - projection emission no longer overstates outward change when nothing changed +3. proof preserved + - all `14A` and `14B` proofs still pass + - publication remains bounded by the same explicit state owner + +--- + +### `14C` Delivery Note Rev 1 + +Date: 2026-04-03 +Scope: make the first bounded boundary/recovery truths explicit in the core +shell so recovery-in-progress and rebuild closure affect mode/publication +semantics directly + +What changed: + +1. `BoundaryView` now carries more explicit boundary truth: + - `CommittedLSN` + - `TargetLSN` + - `AchievedLSN` + - plus the previously separated durable/checkpoint/diagnostic fields +2. `RecoveryView` is now an explicit core-owned state with bounded phases: + - `idle` + - `catching_up` + - `needs_rebuild` + - `rebuilding` +3. the event vocabulary now includes: + - `CommittedLSNAdvanced` + - `CatchUpPlanned` + - `RecoveryProgressObserved` + - `RebuildStarted` + - extended `RebuildCommitted` with explicit achieved boundary support +4. recovery-in-progress now blocks `replica_ready` / publication overclaim + through mode recompute: + - active catch-up or rebuild forces `bootstrap_pending` + with reason `recovery_in_progress` + - rebuild-required stays `needs_rebuild` + +Files changed: + +1. `sw-block/engine/replication/state.go` + - added explicit `RecoveryView` + - expanded `BoundaryView` +2. `sw-block/engine/replication/event.go` + - added boundary/recovery events +3. `sw-block/engine/replication/engine.go` + - boundary truth is now updated explicitly and monotonically + - recovery state now participates directly in mode/publication recompute + - assignment changes clear stale recovery target/achieved truth +4. `sw-block/engine/replication/phase14_boundary_test.go` + - added structural boundary/recovery proofs + +Proofs added: + +1. boundary-truth separation + - `CommittedLSN`, `CheckpointLSN`, `DurableLSN`, and diagnostic shipped + progress remain distinct truths +2. catch-up blocks ready overclaim + - a replica with role applied + receiver ready still falls back to + `bootstrap_pending` with reason `recovery_in_progress` while catch-up is + active +3. rebuild boundary closure + - `needs_rebuild` -> `rebuilding` -> `idle` is explicit in recovery truth + - rebuild commit aligns achieved/durable/checkpoint boundaries + - rebuild completion on replica returns to `replica_ready`, not + `publish_healthy` + +Validation: + +1. `gofmt -w state.go event.go engine.go phase14_boundary_test.go` +2. `go test ./...` +3. result: `PASS` + +Constraint / overclaim / proof review: + +1. semantic constraint satisfied + - `CP13-3`: durable truth remains distinct from diagnostic sender progress + - `CP13-7`: rebuild is explicit fail-closed truth, not an ambiguous degraded + tail + - `T14`: engine owns recovery policy and meaning, not backend convenience +2. overclaim avoided + - receiver-ready during catch-up no longer looks like final ready state + - rebuild-in-progress no longer risks being interpreted as ordinary + bootstrap/readiness closure + - rebuild completion on replica does not overclaim publication health +3. proof preserved + - all `14A` and `14B` proofs still pass + - publication remains derived from explicit core-owned truth + +--- + +### `14C` Delivery Note Rev 2 + +Date: 2026-04-03 +Scope: close the first bounded recovery-closure gap by making catch-up +completion explicit and projecting recovery truth outward + +What changed: + +1. `PublicationProjection` now carries `RecoveryView`, so recovery truth is part + of outward normalized meaning rather than hidden only in internal state +2. catch-up now has an explicit closure event: + - `CatchUpCompleted` +3. catch-up completion now: + - advances achieved boundary + - advances durable boundary on the bounded replica path + - returns recovery phase to `idle` + - allows mode to return from `bootstrap_pending` to `replica_ready` + +Files changed: + +1. `sw-block/engine/replication/projection.go` + - projection now exposes `RecoveryView` +2. `sw-block/engine/replication/event.go` + - added `CatchUpCompleted` +3. `sw-block/engine/replication/engine.go` + - catch-up planning resets achieved progress for the new plan + - catch-up completion explicitly closes recovery phase and updates boundaries +4. `sw-block/engine/replication/phase14_boundary_test.go` + - strengthened catch-up proof with completion semantics + - strengthened rebuild proof with outward recovery projection checks + +Proofs strengthened: + +1. recovery truth is projection-visible + - `catching_up`, `rebuilding`, and `idle` are now asserted through outward + projection, not only internal state snapshots +2. catch-up completion closure + - replica catch-up returns to `replica_ready` + - achieved and durable boundaries converge to the explicit completed target + - no publication-health overclaim appears + +Validation: + +1. `gofmt -w projection.go event.go engine.go phase14_boundary_test.go` +2. `go test ./...` +3. result: `PASS` + +Constraint / overclaim / proof review: + +1. semantic constraint satisfied + - recovery closure is now expressed as explicit core-owned truth, not timing + intuition +2. overclaim avoided + - catch-up no longer stays indefinitely in an ambiguous in-progress state + - recovery truth no longer disappears from outward projection +3. proof preserved + - `14A`, `14B`, and `14C rev 1` proofs still pass + +--- + +### `14C` Delivery Note Rev 3 + +Date: 2026-04-03 +Scope: turn recovery start into explicit bounded command semantics so `catch-up` +and `rebuild` are not only state/projection truth but also first-class core +decisions + +What changed: + +1. added explicit recovery-start commands: + - `StartCatchUpCommand` + - `StartRebuildCommand` +2. recovery plan/start events now emit bounded commands: + - `CatchUpPlanned(target)` -> `start_catchup` when the target is newly needed + - `RebuildStarted(target)` -> `start_rebuild` when the target is newly needed +3. repeated identical recovery-start events are now true no-op command + sequences + +Files changed: + +1. `sw-block/engine/replication/command.go` + - added explicit recovery-start commands +2. `sw-block/engine/replication/state.go` + - extended private command-state tracking for catch-up/rebuild targets +3. `sw-block/engine/replication/engine.go` + - emits bounded recovery-start commands from recovery events + - deduplicates repeated identical recovery-start requests +4. `sw-block/engine/replication/phase14_command_test.go` + - added bounded catch-up start proof + - added bounded rebuild start proof + +Proofs strengthened: + +1. catch-up start boundedness + - first `CatchUpPlanned(55)` emits: + - `start_catchup` + - `publish_projection` + - repeated identical `CatchUpPlanned(55)` emits no commands +2. rebuild start boundedness + - first `RebuildStarted(80)` emits: + - `start_rebuild` + - `publish_projection` + - repeated identical `RebuildStarted(80)` emits no commands + +Validation: + +1. `gofmt -w command.go state.go engine.go phase14_command_test.go` +2. `go test ./...` +3. result: `PASS` + +Constraint / overclaim / proof review: + +1. semantic constraint satisfied + - recovery policy is now explicit as both state truth and command decision +2. overclaim avoided + - recovery-start intent no longer hides only in state mutation + - repeated planning/start events no longer look like fresh work every time +3. proof preserved + - all `14A`, `14B`, and `14C` proofs still pass + +--- + +### `14C` Delivery Note Rev 4 + +Date: 2026-04-03 +Scope: close the stale-recovery leakage gap so old recovery truth and old +recovery-start intent cannot survive into a new assignment/epoch cycle + +What changed: + +1. added proof that assignment change clears stale recovery truth: + - recovery phase returns to `idle` + - target and achieved boundaries are cleared + - mode/publication fall back to the new assignment bootstrap state +2. added proof that a fresh assignment cycle may legitimately re-emit the same + recovery-start command for the same target + +Files changed: + +1. `sw-block/engine/replication/phase14_boundary_test.go` + - added stale-recovery-reset proof across assignment/epoch change +2. `sw-block/engine/replication/phase14_command_test.go` + - added fresh-cycle recovery-start reissue proof + +Proofs strengthened: + +1. stale recovery does not leak across assignment cycles +2. recovery-start command dedupe is cycle-bounded rather than globally sticky + +Validation: + +1. `gofmt -w phase14_boundary_test.go phase14_command_test.go` +2. `go test ./...` +3. result: `PASS` + +Constraint / overclaim / proof review: + +1. semantic constraint satisfied + - recovery truth and recovery command intent are now scoped to the active + assignment/epoch cycle +2. overclaim avoided + - old target/achieved/recovery phase cannot make a new assignment look + partially recovered + - dedupe state cannot suppress valid fresh-cycle recovery work +3. proof preserved + - all previous `14A/14B/14C` proofs still pass + +#### `Phase 14` first-round closure + +At this point the first bounded `Phase 14` core shell is in place: + +1. `14A` delivered + - explicit mode / readiness / publication ownership +2. `14B` delivered + - bounded command-emission rules +3. `14C` delivered + - explicit boundary / recovery truth, projection visibility, recovery-start + commands, and assignment-cycle reset rules + +Interpretation: + +1. this is a real explicit `V2 core` shell in `sw-block/engine/replication` +2. it is still not a live runtime cutover +3. the best next step is `Phase 15A` adapter ingress/egress rebinding on one + narrow path + +--- + +### Post-Closure Tightening + +Date: 2026-04-03 +Reason: manager review correctly identified two remaining risks: + +1. `14A/14B/14C` slice-boundary blur in top-level phase wording +2. duplicated `publish_healthy` authority in core state/projection + +Actions taken: + +1. `sw-block/.private/phase/phase-14.md` + - tightened `14A` so it owns only mode/readiness/publication shell closure + - made `14B` the explicit owner of command-sequence closure + - made `14C` the explicit owner of durable-boundary and recovery closure +2. `sw-block/engine/replication/state.go` + - removed `ReadinessView.PublishHealthy` + - documented `PublicationView` as the semantic owner for publication truth +3. `sw-block/engine/replication/projection.go` + - removed duplicate top-level `PublishHealthy` convenience field +4. `sw-block/engine/replication/engine.go` + - publication truth is now carried only through `PublicationView` +5. `phase14_*_test.go` + - switched assertions to `Projection.Publication.Healthy` + +Result: + +1. `PublicationView` is now the single semantic owner for publication health +2. `ReadinessView` and `PublicationProjection` no longer carry parallel + publication-health truth +3. top-level `Phase 14` wording now matches the actual `14A/14B/14C` ownership + split more closely diff --git a/sw-block/.private/phase/phase-14.md b/sw-block/.private/phase/phase-14.md index 039d048b0..1929a12ff 100644 --- a/sw-block/.private/phase/phase-14.md +++ b/sw-block/.private/phase/phase-14.md @@ -1,7 +1,7 @@ # Phase 14 Date: 2026-04-03 -Status: active +Status: delivered Purpose: make the `V2 core` explicit inside `sw-block/engine/replication` so accepted semantic constraints become executable ownership, rather than staying only as design and constrained-`V1` interpretation @@ -101,8 +101,8 @@ Goal: Acceptance object: -1. `VolumeState`, event vocabulary, command vocabulary, projection vocabulary, - and deterministic engine loop exist in `sw-block/engine/replication` +1. `VolumeState`, normalized mode/readiness/publication state, and bounded + outward projection exist in `sw-block/engine/replication` 2. `publish_healthy` is derived from named semantic state rather than runtime convenience 3. fail-closed mode distinctions stay explicit: @@ -113,13 +113,23 @@ Acceptance object: - `degraded` - `needs_rebuild` 4. the structural acceptance tests prove: - - stable identity ownership is preserved - - diagnostic shipped progress does not establish durable truth - - `publish_healthy` requires durable boundary truth + - `replica_ready` and `publish_healthy` stay distinct + - no-replica path stays `allocated_only` - `degraded` and `needs_rebuild` remain distinct fail-closed meanings - the current integrated interpretation remains `constrained_v1`, not live `v2_core` cutover +Ownership boundary: + +1. `14A` owns semantic shell closure for: + - mode + - readiness + - publication +2. `14A` does not own: + - command-sequence closure + - durable-boundary closure + - recovery closure + Status: 1. delivered @@ -138,9 +148,15 @@ Acceptance object: 2. one bounded event sequence produces one bounded command sequence 3. command emission does not depend on `weed/` internals +Ownership boundary: + +1. `14B` owns command-sequence closure +2. `14B` does not redefine mode/publication ownership from `14A` +3. `14B` does not absorb durable-boundary or recovery closure from `14C` + Status: -1. active +1. delivered ### `14C`: Boundary / Recovery Semantic Closure @@ -156,9 +172,17 @@ Acceptance object: fail-closed degradation, and rebuild escalation 3. structural tests stay bounded and do not claim live path migration yet +Ownership boundary: + +1. `14C` owns durable-boundary and recovery closure +2. `14C` may affect mode/publication only through explicit boundary/recovery + truth +3. `14C` does not reopen `14A` shell ownership or `14B` command-sequence + closure + Status: -1. planned +1. delivered ## Manager Review Gate @@ -173,8 +197,10 @@ intuition. ## Immediate Next Step -Continue with `14B`. +Phase 14's first bounded core shell is now in place. -Use the explicit shell from `14A` as the semantic substrate, then freeze -gap-driven command emission so repeated assignments and repeated failures do not -silently turn back into runtime-convenience command spam. +The best next step is `Phase 15A`: + +1. connect one narrow adapter ingress into the explicit core +2. connect one bounded command path back out +3. prove the live path does not split semantic truth from the new core owner diff --git a/sw-block/engine/replication/command.go b/sw-block/engine/replication/command.go index fa1340a72..2651ac1db 100644 --- a/sw-block/engine/replication/command.go +++ b/sw-block/engine/replication/command.go @@ -27,6 +27,20 @@ type ConfigureShipperCommand struct { func (ConfigureShipperCommand) commandName() string { return "configure_shipper" } +type StartCatchUpCommand struct { + VolumeID string + TargetLSN uint64 +} + +func (StartCatchUpCommand) commandName() string { return "start_catchup" } + +type StartRebuildCommand struct { + VolumeID string + TargetLSN uint64 +} + +func (StartRebuildCommand) commandName() string { return "start_rebuild" } + type InvalidateSessionCommand struct { VolumeID string Reason string diff --git a/sw-block/engine/replication/engine.go b/sw-block/engine/replication/engine.go index 8602e40f6..755bfb3f6 100644 --- a/sw-block/engine/replication/engine.go +++ b/sw-block/engine/replication/engine.go @@ -1,5 +1,7 @@ package replication +import "reflect" + // CoreEngine is the first explicit Phase 14 V2 core shell. // It is deterministic and side-effect free: one event in, updated state and // commands/projection out. @@ -21,6 +23,7 @@ func NewCoreEngine() *CoreEngine { // ApplyEvent mutates the bounded core state and emits commands/projection. func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { + prevProjection, hadPrevProjection := e.Projection(ev.VolumeID()) st := e.mustState(ev.VolumeID()) var cmds []Command @@ -45,6 +48,11 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { st.Boundary.DiagnosticShippedLSN = v.ShippedLSN } + case CommittedLSNAdvanced: + if v.CommittedLSN > st.Boundary.CommittedLSN { + st.Boundary.CommittedLSN = v.CommittedLSN + } + case BarrierAccepted: if v.FlushedLSN > st.Boundary.DurableLSN { st.Boundary.DurableLSN = v.FlushedLSN @@ -73,11 +81,53 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { st.Boundary.CheckpointLSN = v.CheckpointLSN } + case CatchUpPlanned: + st.Recovery.Phase = RecoveryCatchingUp + st.Recovery.AchievedLSN = 0 + if v.TargetLSN > st.Recovery.TargetLSN { + st.Recovery.TargetLSN = v.TargetLSN + } + if v.TargetLSN > st.Boundary.TargetLSN { + st.Boundary.TargetLSN = v.TargetLSN + } + st.Recovery.Reason = "" + if st.shouldStartCatchUp(v.TargetLSN) { + cmds = append(cmds, StartCatchUpCommand{ + VolumeID: st.VolumeID, + TargetLSN: v.TargetLSN, + }) + st.commands.CatchUpTargetLSN = v.TargetLSN + } + + case RecoveryProgressObserved: + if v.AchievedLSN > st.Recovery.AchievedLSN { + st.Recovery.AchievedLSN = v.AchievedLSN + } + if v.AchievedLSN > st.Boundary.AchievedLSN { + st.Boundary.AchievedLSN = v.AchievedLSN + } + + case CatchUpCompleted: + if v.AchievedLSN > st.Recovery.AchievedLSN { + st.Recovery.AchievedLSN = v.AchievedLSN + } + if v.AchievedLSN > st.Boundary.AchievedLSN { + st.Boundary.AchievedLSN = v.AchievedLSN + } + if v.AchievedLSN > st.Boundary.DurableLSN { + st.Boundary.DurableLSN = v.AchievedLSN + } + st.Recovery.Phase = RecoveryIdle + st.Recovery.Reason = "" + st.commands.CatchUpTargetLSN = 0 + case NeedsRebuildObserved: st.needsRebuild = true st.rebuildReason = v.Reason st.degraded = false st.degradeReason = "" + st.Recovery.Phase = RecoveryNeedsRebuild + st.Recovery.Reason = v.Reason if st.shouldInvalidate(v.Reason) { cmds = append(cmds, InvalidateSessionCommand{ VolumeID: st.VolumeID, @@ -87,26 +137,60 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { st.commands.InvalidationReason = v.Reason } + case RebuildStarted: + st.needsRebuild = true + st.Recovery.Phase = RecoveryRebuilding + st.Recovery.Reason = st.rebuildReason + st.Recovery.AchievedLSN = 0 + if v.TargetLSN > st.Recovery.TargetLSN { + st.Recovery.TargetLSN = v.TargetLSN + } + if v.TargetLSN > st.Boundary.TargetLSN { + st.Boundary.TargetLSN = v.TargetLSN + } + if st.shouldStartRebuild(v.TargetLSN) { + cmds = append(cmds, StartRebuildCommand{ + VolumeID: st.VolumeID, + TargetLSN: v.TargetLSN, + }) + st.commands.RebuildTargetLSN = v.TargetLSN + } + case RebuildCommitted: st.needsRebuild = false st.rebuildReason = "" st.degraded = false st.degradeReason = "" st.resetInvalidation() + st.Recovery.Phase = RecoveryIdle + st.Recovery.Reason = "" if v.FlushedLSN > st.Boundary.DurableLSN { st.Boundary.DurableLSN = v.FlushedLSN } if v.CheckpointLSN > st.Boundary.CheckpointLSN { st.Boundary.CheckpointLSN = v.CheckpointLSN } + achievedLSN := v.AchievedLSN + if achievedLSN == 0 { + achievedLSN = maxUint64(v.FlushedLSN, v.CheckpointLSN) + } + if achievedLSN > st.Recovery.AchievedLSN { + st.Recovery.AchievedLSN = achievedLSN + } + if achievedLSN > st.Boundary.AchievedLSN { + st.Boundary.AchievedLSN = achievedLSN + } + st.commands.RebuildTargetLSN = 0 } e.recompute(st) proj := e.projectionFor(st) - cmds = append(cmds, PublishProjectionCommand{ - VolumeID: st.VolumeID, - Projection: proj, - }) + if !hadPrevProjection || !reflect.DeepEqual(prevProjection, proj) { + cmds = append(cmds, PublishProjectionCommand{ + VolumeID: st.VolumeID, + Projection: proj, + }) + } return ApplyResult{ Commands: cmds, @@ -146,7 +230,6 @@ func (e *CoreEngine) recompute(st *VolumeState) { st.Readiness.ReplicaReady = st.Role == RoleReplica && st.Readiness.RoleApplied && st.Readiness.ReceiverReady - st.Readiness.PublishHealthy = false st.Mode.Authority = RuntimeAuthorityConstrainedV1 st.Mode.Reason = "" st.Publication = PublicationView{} @@ -160,6 +243,10 @@ func (e *CoreEngine) recompute(st *VolumeState) { st.Publication.Reason = defaultReason(st.degradeReason, "degraded") st.Mode.Name = ModeDegraded st.Mode.Reason = st.Publication.Reason + case st.recoveryActive(): + st.Publication.Reason = "recovery_in_progress" + st.Mode.Name = ModeBootstrapPending + st.Mode.Reason = st.Publication.Reason case !st.hasReplicas(): st.Publication.Reason = "allocated_only" st.Mode.Name = ModeAllocatedOnly @@ -169,7 +256,6 @@ func (e *CoreEngine) recompute(st *VolumeState) { case st.Role == RolePrimary && st.primaryEligibleForPublish(): st.Publication.Healthy = true st.Mode.Name = ModePublishHealthy - st.Readiness.PublishHealthy = true case st.Readiness.Assigned || st.Readiness.RoleApplied: st.Publication.Reason = st.bootstrapReason() st.Mode.Name = ModeBootstrapPending @@ -218,6 +304,11 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) [] st.degraded = false st.degradeReason = "" st.resetInvalidation() + st.Recovery = RecoveryView{Phase: RecoveryIdle} + st.Boundary.TargetLSN = 0 + st.Boundary.AchievedLSN = 0 + st.commands.CatchUpTargetLSN = 0 + st.commands.RebuildTargetLSN = 0 } var cmds []Command @@ -251,15 +342,15 @@ func (e *CoreEngine) projectionFor(st *VolumeState) PublicationProjection { replicaIDs = append(replicaIDs, replica.ReplicaID) } return PublicationProjection{ - VolumeID: st.VolumeID, - Epoch: st.Epoch, - Role: st.Role, - Mode: st.Mode, - Publication: st.Publication, - Readiness: st.Readiness, - Boundary: st.Boundary, - ReplicaIDs: replicaIDs, - PublishHealthy: st.Publication.Healthy, + VolumeID: st.VolumeID, + Epoch: st.Epoch, + Role: st.Role, + Mode: st.Mode, + Publication: st.Publication, + Recovery: st.Recovery, + Readiness: st.Readiness, + Boundary: st.Boundary, + ReplicaIDs: replicaIDs, } } @@ -293,6 +384,14 @@ func (st *VolumeState) shouldInvalidate(reason string) bool { return !st.commands.InvalidationIssued || st.commands.InvalidationReason != reason } +func (st *VolumeState) shouldStartCatchUp(targetLSN uint64) bool { + return targetLSN > 0 && st.commands.CatchUpTargetLSN != targetLSN +} + +func (st *VolumeState) shouldStartRebuild(targetLSN uint64) bool { + return targetLSN > 0 && st.commands.RebuildTargetLSN != targetLSN +} + func (st *VolumeState) resetInvalidation() { st.commands.InvalidationIssued = false st.commands.InvalidationReason = "" @@ -319,6 +418,10 @@ func (st *VolumeState) bootstrapReason() string { } } +func (st *VolumeState) recoveryActive() bool { + return st.Recovery.Phase == RecoveryCatchingUp || st.Recovery.Phase == RecoveryRebuilding +} + func defaultReason(reason, fallback string) string { if reason != "" { return reason @@ -340,3 +443,10 @@ func sameReplicaAssignments(left, right []ReplicaAssignment) bool { } return true } + +func maxUint64(left, right uint64) uint64 { + if left > right { + return left + } + return right +} diff --git a/sw-block/engine/replication/event.go b/sw-block/engine/replication/event.go index 6ff1b2ad0..ff4ca2aae 100644 --- a/sw-block/engine/replication/event.go +++ b/sw-block/engine/replication/event.go @@ -52,6 +52,15 @@ type DiagnosticShippedAdvanced struct { func (e DiagnosticShippedAdvanced) VolumeID() string { return e.ID } +// CommittedLSNAdvanced updates the committed boundary truth without implying +// replica durability or checkpoint advancement. +type CommittedLSNAdvanced struct { + ID string + CommittedLSN uint64 +} + +func (e CommittedLSNAdvanced) VolumeID() string { return e.ID } + // BarrierAccepted advances authoritative durable progress. type BarrierAccepted struct { ID string @@ -76,6 +85,32 @@ type CheckpointAdvanced struct { func (e CheckpointAdvanced) VolumeID() string { return e.ID } +// CatchUpPlanned freezes the current replay target as bounded recovery truth. +type CatchUpPlanned struct { + ID string + TargetLSN uint64 +} + +func (e CatchUpPlanned) VolumeID() string { return e.ID } + +// RecoveryProgressObserved updates achieved replay/rebuild progress without +// implying publication or durability closure by itself. +type RecoveryProgressObserved struct { + ID string + AchievedLSN uint64 +} + +func (e RecoveryProgressObserved) VolumeID() string { return e.ID } + +// CatchUpCompleted closes a previously planned catch-up path at an explicit +// achieved boundary. +type CatchUpCompleted struct { + ID string + AchievedLSN uint64 +} + +func (e CatchUpCompleted) VolumeID() string { return e.ID } + // NeedsRebuildObserved is a fail-closed rebuild escalation. type NeedsRebuildObserved struct { ID string @@ -84,9 +119,19 @@ type NeedsRebuildObserved struct { func (e NeedsRebuildObserved) VolumeID() string { return e.ID } +// RebuildStarted moves the recovery truth from blocked-needs-rebuild into an +// explicit rebuilding path with a frozen target. +type RebuildStarted struct { + ID string + TargetLSN uint64 +} + +func (e RebuildStarted) VolumeID() string { return e.ID } + // RebuildCommitted clears the rebuild condition with bounded durable truth. type RebuildCommitted struct { ID string + AchievedLSN uint64 FlushedLSN uint64 CheckpointLSN uint64 } diff --git a/sw-block/engine/replication/phase14_boundary_test.go b/sw-block/engine/replication/phase14_boundary_test.go new file mode 100644 index 000000000..a954b86ec --- /dev/null +++ b/sw-block/engine/replication/phase14_boundary_test.go @@ -0,0 +1,192 @@ +package replication + +import "testing" + +func TestPhase14_BoundaryTruthsStayDistinct(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(CommittedLSNAdvanced{ID: "vol-boundary", CommittedLSN: 20}) + core.ApplyEvent(CheckpointAdvanced{ID: "vol-boundary", CheckpointLSN: 12}) + core.ApplyEvent(DiagnosticShippedAdvanced{ID: "vol-boundary", ShippedLSN: 30}) + result := core.ApplyEvent(BarrierAccepted{ID: "vol-boundary", FlushedLSN: 8}) + + if result.Projection.Boundary.CommittedLSN != 20 { + t.Fatalf("committed_lsn=%d", result.Projection.Boundary.CommittedLSN) + } + if result.Projection.Boundary.CheckpointLSN != 12 { + t.Fatalf("checkpoint_lsn=%d", result.Projection.Boundary.CheckpointLSN) + } + if result.Projection.Boundary.DurableLSN != 8 { + t.Fatalf("durable_lsn=%d", result.Projection.Boundary.DurableLSN) + } + if result.Projection.Boundary.DiagnosticShippedLSN != 30 { + t.Fatalf("diagnostic_shipped_lsn=%d", result.Projection.Boundary.DiagnosticShippedLSN) + } +} + +func TestPhase14_RecoveryCatchUpBlocksReplicaReadyOverclaim(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-recovery", + Epoch: 4, + Role: RoleReplica, + Replicas: []ReplicaAssignment{ + {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.20:9333", CtrlAddr: "10.0.0.20:9334", Version: 1}}, + }, + }) + core.ApplyEvent(RoleApplied{ID: "vol-recovery"}) + core.ApplyEvent(ReceiverReadyObserved{ID: "vol-recovery"}) + + result := core.ApplyEvent(CatchUpPlanned{ID: "vol-recovery", TargetLSN: 40}) + if result.Projection.Mode.Name != ModeBootstrapPending { + t.Fatalf("mode=%s", result.Projection.Mode.Name) + } + if result.Projection.Publication.Reason != "recovery_in_progress" { + t.Fatalf("publication_reason=%q", result.Projection.Publication.Reason) + } + if result.State.Recovery.Phase != RecoveryCatchingUp { + t.Fatalf("recovery_phase=%s", result.State.Recovery.Phase) + } + if result.Projection.Recovery.Phase != RecoveryCatchingUp { + t.Fatalf("projection_recovery_phase=%s", result.Projection.Recovery.Phase) + } + if result.Projection.Boundary.TargetLSN != 40 { + t.Fatalf("target_lsn=%d", result.Projection.Boundary.TargetLSN) + } + + result = core.ApplyEvent(RecoveryProgressObserved{ID: "vol-recovery", AchievedLSN: 25}) + if result.Projection.Mode.Name != ModeBootstrapPending { + t.Fatalf("mode=%s", result.Projection.Mode.Name) + } + if result.Projection.Publication.Healthy { + t.Fatal("catch-up progress must not establish healthy publication") + } + if result.Projection.Boundary.AchievedLSN != 25 { + t.Fatalf("achieved_lsn=%d", result.Projection.Boundary.AchievedLSN) + } + + result = core.ApplyEvent(CatchUpCompleted{ID: "vol-recovery", AchievedLSN: 40}) + if result.State.Recovery.Phase != RecoveryIdle { + t.Fatalf("recovery_phase=%s", result.State.Recovery.Phase) + } + if result.Projection.Recovery.AchievedLSN != 40 { + t.Fatalf("projection_achieved_lsn=%d", result.Projection.Recovery.AchievedLSN) + } + if result.Projection.Boundary.DurableLSN != 40 { + t.Fatalf("durable_lsn=%d", result.Projection.Boundary.DurableLSN) + } + if result.Projection.Mode.Name != ModeReplicaReady { + t.Fatalf("mode=%s", result.Projection.Mode.Name) + } + if result.Projection.Publication.Healthy { + t.Fatal("catch-up completion on replica must not overclaim publish healthy") + } +} + +func TestPhase14_RebuildCommitClosesRecoveryBoundary(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-rebuild", + Epoch: 8, + Role: RoleReplica, + Replicas: []ReplicaAssignment{ + {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.21:9333", CtrlAddr: "10.0.0.21:9334", Version: 1}}, + }, + }) + core.ApplyEvent(RoleApplied{ID: "vol-rebuild"}) + core.ApplyEvent(ReceiverReadyObserved{ID: "vol-rebuild"}) + + result := core.ApplyEvent(NeedsRebuildObserved{ID: "vol-rebuild", Reason: "gap_too_large"}) + if result.State.Recovery.Phase != RecoveryNeedsRebuild { + t.Fatalf("recovery_phase=%s", result.State.Recovery.Phase) + } + if result.Projection.Mode.Name != ModeNeedsRebuild { + t.Fatalf("mode=%s", result.Projection.Mode.Name) + } + + result = core.ApplyEvent(RebuildStarted{ID: "vol-rebuild", TargetLSN: 64}) + if result.State.Recovery.Phase != RecoveryRebuilding { + t.Fatalf("recovery_phase=%s", result.State.Recovery.Phase) + } + if result.Projection.Recovery.Phase != RecoveryRebuilding { + t.Fatalf("projection_recovery_phase=%s", result.Projection.Recovery.Phase) + } + if result.Projection.Boundary.TargetLSN != 64 { + t.Fatalf("target_lsn=%d", result.Projection.Boundary.TargetLSN) + } + if result.Projection.Mode.Name != ModeNeedsRebuild { + t.Fatalf("mode=%s", result.Projection.Mode.Name) + } + + result = core.ApplyEvent(RebuildCommitted{ + ID: "vol-rebuild", + AchievedLSN: 64, + FlushedLSN: 64, + CheckpointLSN: 64, + }) + if result.State.Recovery.Phase != RecoveryIdle { + t.Fatalf("recovery_phase=%s", result.State.Recovery.Phase) + } + if result.Projection.Recovery.Phase != RecoveryIdle { + t.Fatalf("projection_recovery_phase=%s", result.Projection.Recovery.Phase) + } + if result.Projection.Boundary.AchievedLSN != 64 { + t.Fatalf("achieved_lsn=%d", result.Projection.Boundary.AchievedLSN) + } + if result.Projection.Boundary.DurableLSN != 64 { + t.Fatalf("durable_lsn=%d", result.Projection.Boundary.DurableLSN) + } + if result.Projection.Boundary.CheckpointLSN != 64 { + t.Fatalf("checkpoint_lsn=%d", result.Projection.Boundary.CheckpointLSN) + } + if result.Projection.Mode.Name != ModeReplicaReady { + t.Fatalf("mode=%s", result.Projection.Mode.Name) + } + if result.Projection.Publication.Healthy { + t.Fatal("replica rebuild completion must not overclaim healthy publication") + } +} + +func TestPhase14_AssignmentChangeClearsStaleRecoveryTruth(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-reassign", + Epoch: 1, + Role: RoleReplica, + Replicas: []ReplicaAssignment{ + {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.30:9333", CtrlAddr: "10.0.0.30:9334", Version: 1}}, + }, + }) + core.ApplyEvent(RoleApplied{ID: "vol-reassign"}) + core.ApplyEvent(ReceiverReadyObserved{ID: "vol-reassign"}) + core.ApplyEvent(CatchUpPlanned{ID: "vol-reassign", TargetLSN: 70}) + core.ApplyEvent(RecoveryProgressObserved{ID: "vol-reassign", AchievedLSN: 41}) + + result := core.ApplyEvent(AssignmentDelivered{ + ID: "vol-reassign", + Epoch: 2, + Role: RoleReplica, + Replicas: []ReplicaAssignment{ + {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.31:9333", CtrlAddr: "10.0.0.31:9334", Version: 2}}, + }, + }) + + if result.State.Recovery.Phase != RecoveryIdle { + t.Fatalf("recovery_phase=%s", result.State.Recovery.Phase) + } + if result.Projection.Boundary.TargetLSN != 0 { + t.Fatalf("target_lsn=%d", result.Projection.Boundary.TargetLSN) + } + if result.Projection.Boundary.AchievedLSN != 0 { + t.Fatalf("achieved_lsn=%d", result.Projection.Boundary.AchievedLSN) + } + if result.Projection.Mode.Name != ModeBootstrapPending { + t.Fatalf("mode=%s", result.Projection.Mode.Name) + } + if result.Projection.Publication.Reason != "awaiting_role_apply" { + t.Fatalf("publication_reason=%q", result.Projection.Publication.Reason) + } +} diff --git a/sw-block/engine/replication/phase14_command_test.go b/sw-block/engine/replication/phase14_command_test.go index b5f5b66e8..483699965 100644 --- a/sw-block/engine/replication/phase14_command_test.go +++ b/sw-block/engine/replication/phase14_command_test.go @@ -24,9 +24,7 @@ func TestPhase14_CommandSequence_PrimaryAssignmentIsBounded(t *testing.T) { }) result = core.ApplyEvent(ev) - assertCommandNames(t, result.Commands, []string{ - "publish_projection", - }) + assertCommandNames(t, result.Commands, nil) } func TestPhase14_CommandSequence_ReplicaAssignmentIsBounded(t *testing.T) { @@ -48,9 +46,7 @@ func TestPhase14_CommandSequence_ReplicaAssignmentIsBounded(t *testing.T) { }) result = core.ApplyEvent(ev) - assertCommandNames(t, result.Commands, []string{ - "publish_projection", - }) + assertCommandNames(t, result.Commands, nil) } func TestPhase14_CommandSequence_AssignmentChangeReissuesNeededCommand(t *testing.T) { @@ -107,9 +103,113 @@ func TestPhase14_CommandSequence_InvalidateOnlyOnNewFailureTransition(t *testing }) result = core.ApplyEvent(BarrierRejected{ID: "vol-cmd-failure", Reason: "timeout"}) + assertCommandNames(t, result.Commands, nil) +} + +func TestPhase14_CommandSequence_PublishOnlyWhenProjectionChanges(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-cmd-publish", + Epoch: 2, + Role: RolePrimary, + Replicas: []ReplicaAssignment{ + {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.15:9333", CtrlAddr: "10.0.0.15:9334", Version: 1}}, + }, + }) + + result := core.ApplyEvent(RoleApplied{ID: "vol-cmd-publish"}) assertCommandNames(t, result.Commands, []string{ "publish_projection", }) + + result = core.ApplyEvent(RoleApplied{ID: "vol-cmd-publish"}) + assertCommandNames(t, result.Commands, nil) +} + +func TestPhase14_CommandSequence_CatchUpStartIsBounded(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-cmd-catchup", + Epoch: 6, + Role: RoleReplica, + 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"}) + + result := core.ApplyEvent(CatchUpPlanned{ID: "vol-cmd-catchup", TargetLSN: 55}) + assertCommandNames(t, result.Commands, []string{ + "start_catchup", + "publish_projection", + }) + + result = core.ApplyEvent(CatchUpPlanned{ID: "vol-cmd-catchup", TargetLSN: 55}) + assertCommandNames(t, result.Commands, nil) +} + +func TestPhase14_CommandSequence_RebuildStartIsBounded(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-cmd-rebuild", + Epoch: 7, + Role: RoleReplica, + 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}) + assertCommandNames(t, result.Commands, []string{ + "start_rebuild", + "publish_projection", + }) + + result = core.ApplyEvent(RebuildStarted{ID: "vol-cmd-rebuild", TargetLSN: 80}) + assertCommandNames(t, result.Commands, nil) +} + +func TestPhase14_CommandSequence_AssignmentChangeAllowsFreshRecoveryStart(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-cmd-reassign", + Epoch: 1, + Role: RoleReplica, + 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"}) + + result := core.ApplyEvent(CatchUpPlanned{ID: "vol-cmd-reassign", TargetLSN: 90}) + assertCommandNames(t, result.Commands, []string{ + "start_catchup", + "publish_projection", + }) + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-cmd-reassign", + Epoch: 2, + Role: RoleReplica, + Replicas: []ReplicaAssignment{ + {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.19:9333", CtrlAddr: "10.0.0.19:9334", Version: 2}}, + }, + }) + + result = core.ApplyEvent(CatchUpPlanned{ID: "vol-cmd-reassign", TargetLSN: 90}) + assertCommandNames(t, result.Commands, []string{ + "start_catchup", + "publish_projection", + }) } func assertCommandNames(t *testing.T, cmds []Command, want []string) { @@ -118,6 +218,9 @@ func assertCommandNames(t *testing.T, cmds []Command, want []string) { for _, cmd := range cmds { got = append(got, cmd.commandName()) } + if want == nil { + want = []string{} + } if !reflect.DeepEqual(got, want) { t.Fatalf("commands=%v, want %v", got, want) } diff --git a/sw-block/engine/replication/phase14_core_test.go b/sw-block/engine/replication/phase14_core_test.go index 25ed7ff18..e836b8756 100644 --- a/sw-block/engine/replication/phase14_core_test.go +++ b/sw-block/engine/replication/phase14_core_test.go @@ -64,7 +64,7 @@ func TestPhase14_CorePublishHealthyRequiresBarrierDurability(t *testing.T) { if result.Projection.Publication.Reason != "awaiting_role_apply" { t.Fatalf("publication_reason=%q", result.Projection.Publication.Reason) } - if result.Projection.PublishHealthy { + if result.Projection.Publication.Healthy { t.Fatal("fresh assignment must not publish healthy") } @@ -77,7 +77,7 @@ func TestPhase14_CorePublishHealthyRequiresBarrierDurability(t *testing.T) { t.Fatalf("publication_reason=%q", result.Projection.Publication.Reason) } result = core.ApplyEvent(ShipperConnectedObserved{ID: "vol-a"}) - if result.Projection.PublishHealthy { + if result.Projection.Publication.Healthy { t.Fatal("connected shipper without barrier durability must stay non-healthy") } if result.Projection.Mode.Name != ModeBootstrapPending { @@ -88,7 +88,7 @@ func TestPhase14_CorePublishHealthyRequiresBarrierDurability(t *testing.T) { } result = core.ApplyEvent(BarrierAccepted{ID: "vol-a", FlushedLSN: 12}) - if !result.Projection.PublishHealthy { + if !result.Projection.Publication.Healthy { t.Fatal("barrier durability should enable publish healthy on eligible primary path") } if result.Projection.Mode.Name != ModePublishHealthy { @@ -122,7 +122,7 @@ func TestPhase14_CoreDiagnosticShippedDoesNotCreateDurableTruth(t *testing.T) { if result.Projection.Boundary.DiagnosticShippedLSN != 99 { t.Fatalf("diagnostic_shipped=%d", result.Projection.Boundary.DiagnosticShippedLSN) } - if result.Projection.PublishHealthy { + if result.Projection.Publication.Healthy { t.Fatal("diagnostic shipped progress must not establish durable healthy publication") } if result.Projection.Mode.Name != ModeBootstrapPending { @@ -148,7 +148,7 @@ func TestPhase14_CoreFailClosedModesStayDistinct(t *testing.T) { if result.Projection.Mode.Name != ModeDegraded { t.Fatalf("mode=%s", result.Projection.Mode.Name) } - if result.Projection.PublishHealthy { + if result.Projection.Publication.Healthy { t.Fatal("degraded mode must fail closed") } @@ -156,7 +156,7 @@ func TestPhase14_CoreFailClosedModesStayDistinct(t *testing.T) { if result.Projection.Mode.Name != ModeNeedsRebuild { t.Fatalf("mode=%s", result.Projection.Mode.Name) } - if result.Projection.PublishHealthy { + if result.Projection.Publication.Healthy { t.Fatal("needs_rebuild must fail closed") } if result.Projection.Mode.Reason != "gap_too_large" { @@ -188,7 +188,7 @@ func TestPhase14_CoreProjectionMarksConstrainedRuntimeAuthority(t *testing.T) { if result.Projection.Mode.Name != ModeReplicaReady { t.Fatalf("mode=%s", result.Projection.Mode.Name) } - if result.Projection.PublishHealthy { + if result.Projection.Publication.Healthy { t.Fatal("replica-ready is not the same as publish-healthy") } if result.Projection.Publication.Reason != "replica_not_primary" { @@ -208,7 +208,7 @@ func TestPhase14_CoreAllocatedOnlyWithoutReplicas(t *testing.T) { if result.Projection.Mode.Name != ModeAllocatedOnly { t.Fatalf("mode=%s", result.Projection.Mode.Name) } - if result.Projection.PublishHealthy { + if result.Projection.Publication.Healthy { t.Fatal("allocated-only volume must not publish healthy") } if result.Projection.Publication.Reason != "allocated_only" { diff --git a/sw-block/engine/replication/projection.go b/sw-block/engine/replication/projection.go index 1c8ae2b98..2d57afede 100644 --- a/sw-block/engine/replication/projection.go +++ b/sw-block/engine/replication/projection.go @@ -9,9 +9,9 @@ type PublicationProjection struct { Mode ModeView Publication PublicationView + Recovery RecoveryView Readiness ReadinessView Boundary BoundaryView - ReplicaIDs []string - PublishHealthy bool + ReplicaIDs []string } diff --git a/sw-block/engine/replication/state.go b/sw-block/engine/replication/state.go index 1e74871b7..50a89bff2 100644 --- a/sw-block/engine/replication/state.go +++ b/sw-block/engine/replication/state.go @@ -39,14 +39,16 @@ type ReadinessView struct { ShipperConfigured bool ShipperConnected bool ReplicaReady bool - PublishHealthy bool } // BoundaryView captures durable/publication boundary truth. DurableLSN is the // authority for replicated durability; diagnostic shipped progress is separate. type BoundaryView struct { + CommittedLSN uint64 DurableLSN uint64 CheckpointLSN uint64 + TargetLSN uint64 + AchievedLSN uint64 DiagnosticShippedLSN uint64 LastBarrierOK bool LastBarrierReason string @@ -59,20 +61,39 @@ type ModeView struct { Authority RuntimeAuthority } -// PublicationView is the outward publication truth derived from the same core -// state. It is kept separate from readiness so the publication automaton stays -// explicit in Phase 14A. +// PublicationView is the semantic owner for outward publication truth. Other +// projections may derive convenience fields from it, but they must not become +// parallel authorities. type PublicationView struct { Healthy bool Reason string } +type RecoveryPhase string + +const ( + RecoveryIdle RecoveryPhase = "idle" + RecoveryCatchingUp RecoveryPhase = "catching_up" + RecoveryNeedsRebuild RecoveryPhase = "needs_rebuild" + RecoveryRebuilding RecoveryPhase = "rebuilding" +) + +// RecoveryView captures the bounded recovery truth the core owns directly. +type RecoveryView struct { + Phase RecoveryPhase + TargetLSN uint64 + AchievedLSN uint64 + Reason string +} + type commandState struct { RoleEpoch uint64 Role VolumeRole ReceiverStartEpoch uint64 ShipperConfigEpoch uint64 ShipperConfigReplicas []ReplicaAssignment + CatchUpTargetLSN uint64 + RebuildTargetLSN uint64 InvalidationIssued bool InvalidationReason string } @@ -89,6 +110,7 @@ type VolumeState struct { Boundary BoundaryView Mode ModeView Publication PublicationView + Recovery RecoveryView degraded bool degradeReason string @@ -105,6 +127,9 @@ func newVolumeState(volumeID string) *VolumeState { Name: ModeAllocatedOnly, Authority: RuntimeAuthorityConstrainedV1, }, + Recovery: RecoveryView{ + Phase: RecoveryIdle, + }, } }