refactor: aggregate bounded multi-replica catch-up conservatively

Track catch-up observations per replica so the volume-level recovery view stays in catching_up until all bounded replicas complete. This preserves the current bounded semantics while removing an overclaim that would block later multi-replica startup ownership work.

Made-with: Cursor

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

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

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:

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")
+	}
+}

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
 }