diff --git a/sw-block/.private/phase/phase-16-log.md b/sw-block/.private/phase/phase-16-log.md index 26df4b0c2..0f7babc37 100644 --- a/sw-block/.private/phase/phase-16-log.md +++ b/sw-block/.private/phase/phase-16-log.md @@ -912,3 +912,56 @@ Conclusion: replica-scoped on the bounded path 3. this slice still does not claim broad multi-replica startup ownership or full recovery-loop closure + +--- + +#### `16H` Start Note Rev 1 + +Date: 2026-04-04 +Scope: conservative multi-replica catch-up observation aggregation on the +bounded core-present path + +Why this slice exists: + +1. `16F` made bounded recovery command addressing replica-scoped +2. `16G` made bounded recovery observation events replica-scoped +3. but the volume-level recovery view can still return to `idle` too early if + more than one replica is catching up and the first one finishes + +Chosen implementation rule: + +1. track bounded catch-up observation internally per replica +2. aggregate the volume-level recovery view conservatively +3. stay in `catching_up` until all bounded catch-up replicas complete +4. do not yet claim broad multi-replica startup ownership + +--- + +#### `16H` Delivery Note Rev 1 + +Date: 2026-04-04 +Scope: conservative multi-replica catch-up observation aggregation on the +bounded core-present path + +What changed: + +1. `sw-block/engine/replication/state.go` + - added internal per-replica catch-up observation tracking +2. `sw-block/engine/replication/engine.go` + - catch-up planning / progress / completion now aggregate volume-level + recovery state conservatively across bounded replicas +3. `sw-block/engine/replication/phase14_boundary_test.go` + - added focused proof that one completed replica does not return the volume + to `idle` or `publish_healthy` while another replica is still catching up + +Proof / evidence: + +1. `go test ./...` from `sw-block/engine/replication` +2. result: `PASS` + +Conclusion: + +1. the bounded multi-replica catch-up observation path no longer overclaims + completion after the first replica finishes +2. this slice is still only an enabling aggregation step, not broad + multi-replica startup ownership diff --git a/sw-block/.private/phase/phase-16.md b/sw-block/.private/phase/phase-16.md index e83f57f8b..2e27e67a8 100644 --- a/sw-block/.private/phase/phase-16.md +++ b/sw-block/.private/phase/phase-16.md @@ -283,6 +283,47 @@ Evidence: 1. focused working-tree change after `b304b8e21` +### `16H`: Multi-Replica Catch-Up Observation Aggregation + +Goal: + +1. keep the volume-level recovery view on the bounded core path from returning + to `idle` too early when more than one replica is still catching up +2. make the bounded recovery projection aggregate multi-replica catch-up + progress conservatively enough for later startup-ownership widening + +Acceptance object: + +1. when multiple replica-scoped catch-up observations exist for the same + volume, the recovery phase remains `catching_up` until all bounded replicas + complete +2. bounded durable/progress fields do not overclaim completion after only one + replica finishes +3. current single-replica catch-up and rebuilding proofs remain green +4. this slice still does not yet claim broad multi-replica startup ownership + +Current chosen path: + +1. bounded catch-up observation state is tracked per replica internally +2. volume-level recovery projection aggregates that state conservatively +3. aggregate recovery idles only after all bounded catch-up replicas complete + +Status: + +1. delivered + +Delivered result: + +1. bounded catch-up observation is tracked internally per replica +2. volume-level recovery stays `catching_up` until all bounded catch-up + replicas complete +3. bounded durable/progress fields no longer overclaim completion after the + first replica finishes on the multi-replica catch-up path + +Evidence: + +1. focused working-tree change after `16ba70f85` + ## Current Checkpoint Review Target The current review target is the current widened bounded runtime checkpoint @@ -306,6 +347,8 @@ after `Phase 15` closeout: - bounded catch-up recovery-task startup ownership on the single-replica primary path - replica-scoped recovery command addressing on those same bounded paths + - conservative multi-replica catch-up observation aggregation on those same + bounded paths This checkpoint is intentionally still bounded: @@ -338,12 +381,15 @@ boundary: same bounded paths 8. `16G` delivered: - recovery observation events are replica-scoped on those same bounded paths +9. `16H` delivered: + - multi-replica catch-up observation is aggregated conservatively at the + volume projection layer After this checkpoint: 1. keep `legacy P4` only as a compatibility guard 2. the next bounded semantic/runtime decision is whether to widen startup - ownership beyond the single-replica catch-up path now that both command and - observation recovery seams are replica-scoped + ownership beyond the single-replica catch-up path now that multi-replica + catch-up aggregation no longer overclaims completion 3. do not yet claim full recovery-loop closure 4. do not broaden into launch claims diff --git a/sw-block/engine/replication/engine.go b/sw-block/engine/replication/engine.go index 3ba806849..ddb1b97e5 100644 --- a/sw-block/engine/replication/engine.go +++ b/sw-block/engine/replication/engine.go @@ -82,16 +82,27 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { } 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 + replicaID, ok := st.recoveryCommandReplicaIDFromEvent(v.ReplicaID) + if ok { + st.recordCatchUpPlan(replicaID, v.TargetLSN) + targetLSN, achievedLSN, _ := st.catchUpAggregate() + st.Recovery.Phase = RecoveryCatchingUp + st.Recovery.AchievedLSN = achievedLSN + st.Recovery.TargetLSN = targetLSN + st.Boundary.TargetLSN = targetLSN + st.Boundary.AchievedLSN = achievedLSN + } else { + 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 replicaID, ok := st.recoveryCommandReplicaIDFromEvent(v.ReplicaID); ok && st.shouldStartCatchUp(replicaID, v.TargetLSN) { + if ok && st.shouldStartCatchUp(replicaID, v.TargetLSN) { cmds = append(cmds, StartCatchUpCommand{ VolumeID: st.VolumeID, ReplicaID: replicaID, @@ -104,28 +115,56 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult { } case RecoveryProgressObserved: - if v.AchievedLSN > st.Recovery.AchievedLSN { - st.Recovery.AchievedLSN = v.AchievedLSN - } - if v.AchievedLSN > st.Boundary.AchievedLSN { - st.Boundary.AchievedLSN = v.AchievedLSN + if replicaID, ok := st.recoveryCommandReplicaIDFromEvent(v.ReplicaID); ok && st.observeCatchUpProgress(replicaID, v.AchievedLSN) { + _, achievedLSN, _ := st.catchUpAggregate() + st.Recovery.AchievedLSN = achievedLSN + st.Boundary.AchievedLSN = achievedLSN + } else { + 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 replicaID, ok := st.recoveryCommandReplicaIDFromEvent(v.ReplicaID); ok && st.completeCatchUp(replicaID, v.AchievedLSN) { + targetLSN, achievedLSN, allDone := st.catchUpAggregate() + st.Recovery.TargetLSN = targetLSN + st.Recovery.AchievedLSN = achievedLSN + st.Boundary.TargetLSN = targetLSN + st.Boundary.AchievedLSN = achievedLSN + if allDone { + if achievedLSN > st.Boundary.DurableLSN { + st.Boundary.DurableLSN = achievedLSN + } + st.Recovery.Phase = RecoveryIdle + st.Recovery.Reason = "" + st.commands.CatchUpTargets = nil + st.catchUps = nil + } else { + st.Recovery.Phase = RecoveryCatchingUp + st.Recovery.Reason = "" + } + } else { + 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.CatchUpTargets = nil } - 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.CatchUpTargets = nil case NeedsRebuildObserved: + st.catchUps = nil + st.commands.CatchUpTargets = nil st.needsRebuild = true st.rebuildReason = v.Reason st.degraded = false @@ -317,6 +356,7 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) [] st.Recovery = RecoveryView{Phase: RecoveryIdle} st.Boundary.TargetLSN = 0 st.Boundary.AchievedLSN = 0 + st.catchUps = nil st.commands.RecoveryTaskEpoch = 0 st.commands.RecoveryTaskTargets = nil st.commands.CatchUpTargets = nil @@ -470,6 +510,77 @@ func (st *VolumeState) recoveryCommandReplicaIDFromEvent(replicaID string) (stri return st.recoveryCommandReplicaID() } +func (st *VolumeState) recordCatchUpPlan(replicaID string, targetLSN uint64) { + if replicaID == "" || targetLSN == 0 { + return + } + if st.catchUps == nil { + st.catchUps = make(map[string]catchUpObservation) + } + obs := st.catchUps[replicaID] + if obs.TargetLSN != targetLSN { + obs.AchievedLSN = 0 + } + obs.TargetLSN = targetLSN + obs.Completed = false + st.catchUps[replicaID] = obs +} + +func (st *VolumeState) observeCatchUpProgress(replicaID string, achievedLSN uint64) bool { + if replicaID == "" || st.catchUps == nil { + return false + } + obs, ok := st.catchUps[replicaID] + if !ok { + return false + } + if achievedLSN > obs.AchievedLSN { + obs.AchievedLSN = achievedLSN + } + st.catchUps[replicaID] = obs + return true +} + +func (st *VolumeState) completeCatchUp(replicaID string, achievedLSN uint64) bool { + if replicaID == "" || st.catchUps == nil { + return false + } + obs, ok := st.catchUps[replicaID] + if !ok { + return false + } + if achievedLSN > obs.AchievedLSN { + obs.AchievedLSN = achievedLSN + } + if obs.TargetLSN > obs.AchievedLSN { + obs.AchievedLSN = obs.TargetLSN + } + obs.Completed = true + st.catchUps[replicaID] = obs + return true +} + +func (st *VolumeState) catchUpAggregate() (targetLSN uint64, achievedLSN uint64, allDone bool) { + if len(st.catchUps) == 0 { + return 0, 0, true + } + allDone = true + first := true + for _, obs := range st.catchUps { + if obs.TargetLSN > targetLSN { + targetLSN = obs.TargetLSN + } + if first || obs.AchievedLSN < achievedLSN { + achievedLSN = obs.AchievedLSN + first = false + } + if !obs.Completed { + allDone = false + } + } + return targetLSN, achievedLSN, allDone +} + func (st *VolumeState) bootstrapReason() string { switch { case !st.Readiness.RoleApplied: diff --git a/sw-block/engine/replication/phase14_boundary_test.go b/sw-block/engine/replication/phase14_boundary_test.go index a954b86ec..351ec8a54 100644 --- a/sw-block/engine/replication/phase14_boundary_test.go +++ b/sw-block/engine/replication/phase14_boundary_test.go @@ -190,3 +190,60 @@ func TestPhase14_AssignmentChangeClearsStaleRecoveryTruth(t *testing.T) { t.Fatalf("publication_reason=%q", result.Projection.Publication.Reason) } } + +func TestPhase14_MultiReplicaCatchUpAggregatesUntilAllComplete(t *testing.T) { + core := NewCoreEngine() + + core.ApplyEvent(AssignmentDelivered{ + ID: "vol-multi-catchup", + Epoch: 1, + Role: RolePrimary, + RecoveryTarget: SessionCatchUp, + Replicas: []ReplicaAssignment{ + {ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.40:9333", CtrlAddr: "10.0.0.40:9334", Version: 1}}, + {ReplicaID: "replica-2", Endpoint: Endpoint{DataAddr: "10.0.0.41:9333", CtrlAddr: "10.0.0.41:9334", Version: 1}}, + }, + }) + core.ApplyEvent(RoleApplied{ID: "vol-multi-catchup"}) + core.ApplyEvent(ShipperConfiguredObserved{ID: "vol-multi-catchup"}) + core.ApplyEvent(ShipperConnectedObserved{ID: "vol-multi-catchup"}) + + result := core.ApplyEvent(CatchUpPlanned{ID: "vol-multi-catchup", ReplicaID: "replica-1", TargetLSN: 100}) + if result.State.Recovery.Phase != RecoveryCatchingUp { + t.Fatalf("phase_after_first_plan=%s", result.State.Recovery.Phase) + } + result = core.ApplyEvent(CatchUpPlanned{ID: "vol-multi-catchup", ReplicaID: "replica-2", TargetLSN: 100}) + if result.State.Recovery.Phase != RecoveryCatchingUp { + t.Fatalf("phase_after_second_plan=%s", result.State.Recovery.Phase) + } + + result = core.ApplyEvent(RecoveryProgressObserved{ID: "vol-multi-catchup", ReplicaID: "replica-1", AchievedLSN: 100}) + if result.Projection.Recovery.AchievedLSN != 0 { + t.Fatalf("aggregate_achieved_lsn=%d", result.Projection.Recovery.AchievedLSN) + } + + result = core.ApplyEvent(CatchUpCompleted{ID: "vol-multi-catchup", ReplicaID: "replica-1", AchievedLSN: 100}) + if result.State.Recovery.Phase != RecoveryCatchingUp { + t.Fatalf("phase_after_first_complete=%s", result.State.Recovery.Phase) + } + if result.Projection.Boundary.DurableLSN != 0 { + t.Fatalf("durable_lsn_after_first_complete=%d", result.Projection.Boundary.DurableLSN) + } + if result.Projection.Publication.Healthy { + t.Fatal("publication should not become healthy before all replicas complete") + } + + result = core.ApplyEvent(CatchUpCompleted{ID: "vol-multi-catchup", ReplicaID: "replica-2", AchievedLSN: 100}) + if result.State.Recovery.Phase != RecoveryIdle { + t.Fatalf("phase_after_all_complete=%s", result.State.Recovery.Phase) + } + if result.Projection.Boundary.DurableLSN != 100 { + t.Fatalf("durable_lsn_after_all_complete=%d", result.Projection.Boundary.DurableLSN) + } + if result.Projection.Mode.Name != ModePublishHealthy { + t.Fatalf("mode_after_all_complete=%s", result.Projection.Mode.Name) + } + if !result.Projection.Publication.Healthy { + t.Fatal("publication should become healthy after all replicas complete") + } +} diff --git a/sw-block/engine/replication/state.go b/sw-block/engine/replication/state.go index 6354feac6..aa80325b5 100644 --- a/sw-block/engine/replication/state.go +++ b/sw-block/engine/replication/state.go @@ -100,6 +100,12 @@ type commandState struct { InvalidationReason string } +type catchUpObservation struct { + TargetLSN uint64 + AchievedLSN uint64 + Completed bool +} + // VolumeState is the minimal V2-core-owned state for one volume on the bounded // current path. type VolumeState struct { @@ -120,6 +126,7 @@ type VolumeState struct { rebuildReason string recoveryTarget SessionKind commands commandState + catchUps map[string]catchUpObservation } func newVolumeState(volumeID string) *VolumeState { @@ -163,5 +170,11 @@ func (s *VolumeState) Snapshot() VolumeState { out.commands.RebuildTargets[replicaID] = target } } + if s.catchUps != nil { + out.catchUps = make(map[string]catchUpObservation, len(s.catchUps)) + for replicaID, obs := range s.catchUps { + out.catchUps[replicaID] = obs + } + } return out }