fix(recovery): roll discard intents forward and fence per-session conflicts

discard roll-forward required byte equality with the pristine replacement, so a failed cleanup on an already-live session bricked startup permanently. It now checks identity only. Recover fences a session whose own durable state is self-inconsistent instead of aborting startup for every session; infrastructure failures still abort.

commit: 48ce43781515a0e3c09ec6de261a54698c127f43
author: brice · 2mo ago · 2026-07-26 04:54 UTC
parent: 9f082df
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +234 -16

M internal/usecase/recovery/coordinator.go +105 -8


@@ -83,12 +83,54 @@ func compactStrings(values []string) []string {
83	83		return out
84	84		}
85	85
	86	+	var (
	87	+	// ErrDeletionTombstoneInvalid marks a tombstone whose own identity cannot be
	88	+	// used. ErrDeletionTombstoneConflict marks a tombstone that disagrees with a
	89	+	// live catalogue record. Both are session-scoped: they fence one session
	90	+	// instead of failing startup.
	91	+	ErrDeletionTombstoneInvalid = errors.New("recovery: deletion tombstone is unusable")
	92	+	ErrDeletionTombstoneConflict = errors.New("recovery: deletion tombstone conflicts with catalogue")
	93	+	// ErrSessionRecordInvalid marks a catalogue record whose own contents cannot
	94	+	// be advanced deterministically.
	95	+	ErrSessionRecordInvalid = errors.New("recovery: catalogue record is unusable")
	96	+	)
	97	+
	98	+	// Degraded reasons are content-free by construction: they never carry paths,
	99	+	// terminal content, or operator input.
	100	+	const (
	101	+	degradedReasonDiscardIntent = "discard intent conflict"
	102	+	degradedReasonTombstone = "deletion tombstone conflict"
	103	+	degradedReasonRecord = "catalogue record conflict"
	104	+	)
	105	+
86	106		type Coordinator struct {
87	107		catalogue ports.Catalogue
88	108		repository ports.SnapshotRepository
89	109		journal ports.RecoveryJournal
90	110		locks *KeyLocks
91	111		random io.Reader
	112	+
	113	+	conflictMu sync.Mutex
	114	+	conflicts []RecoveryConflict
	115	+	}
	116	+
	117	+	// RecoveryConflict is one session-scoped item Recover fenced off. Recover keeps
	118	+	// the durable source (intent or tombstone) so a later operator action can still
	119	+	// resolve it; startup continues for every other session.
	120	+	type RecoveryConflict struct {
	121	+	Session string
	122	+	Kind string
	123	+	Err error
	124	+	}
	125	+
	126	+	// Conflicts returns the session-scoped items fenced by the last Recover calls.
	127	+	func (c *Coordinator) Conflicts() []RecoveryConflict {
	128	+	if c == nil {
	129	+	return nil
	130	+	}
	131	+	c.conflictMu.Lock()
	132	+	defer c.conflictMu.Unlock()
	133	+	return append([]RecoveryConflict(nil), c.conflicts...)
92	134		}
93	135
94	136		func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, journal ports.RecoveryJournal, random io.Reader) *Coordinator {

@@ -180,7 +222,7 @@ func (c *Coordinator) deleteLocked(ctx context.Context, record domain.CatalogueR
180	222		record.RecoveryState = domain.RecoveryDeleting
181	223		record.DegradedReason = ""
182	224		if err := record.Validate(); err != nil {
183		-	return fmt.Errorf("recovery: invalid deleting record: %w", err)
	225	+	return fmt.Errorf("%w: invalid deleting record: %w", ErrSessionRecordInvalid, err)
184	226		}
185	227		if err := c.catalogue.Replace(record.Name, record); err != nil {
186	228		return err

@@ -202,6 +244,12 @@ func (c *Coordinator) deleteLocked(ctx context.Context, record domain.CatalogueR
202	244		// Recover rolls deleting records and strict deletion tombstones forward. The
203	245		// two sources are enumerated independently because catalogue removal precedes
204	246		// tombstone removal in the commit protocol.
	247	+	//
	248	+	// A per-item failure that only proves one session's durable state disagrees
	249	+	// with itself fences that session and lets startup continue: healthy sessions
	250	+	// must still restore next to a corrupt neighbour. Infrastructure failures
	251	+	// (listing, journal, repository, or catalogue IO, and a non-advancing cursor)
	252	+	// still abort startup, because they say nothing about any single session.
205	253		func (c *Coordinator) Recover(ctx context.Context) error {
206	254		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.journal == nil \|\| c.locks == nil {
207	255		return errors.New("recovery: incomplete coordinator dependencies")

@@ -211,8 +259,12 @@ func (c *Coordinator) Recover(ctx context.Context) error {
211	259		return err
212	260		}
213	261		for _, intent := range intents {
	262	+	// The intent is deliberately retained for a fenced session: only a
	263	+	// successful roll-forward may remove it.
214	264		if err := c.recoverDiscard(ctx, intent); err != nil {
215		-	return err
	265	+	if err := c.fenceSession(intent.SessionName, "discard-intent", degradedReasonDiscardIntent, err); err != nil {
	266	+	return err
	267	+	}
216	268		}
217	269		}
218	270		for _, candidate := range c.catalogue.Records() {

@@ -220,13 +272,16 @@ func (c *Coordinator) Recover(ctx context.Context) error {
220	272		continue
221	273		}
222	274		unlock := c.locks.Lock([]string{candidate.Name})
	275	+	var deleteErr error
223	276		current, ok := c.catalogue.Record(candidate.Name)
224	277		if ok && current.IncarnationID == candidate.IncarnationID && current.RecoveryState == domain.RecoveryDeleting {
225		-	err = c.deleteLocked(ctx, current)
	278	+	deleteErr = c.deleteLocked(ctx, current)
226	279		}
227	280		unlock()
228		-	if err != nil {
229		-	return err
	281	+	if deleteErr != nil {
	282	+	if err := c.fenceSession(candidate.Name, "catalogue-record", degradedReasonRecord, deleteErr); err != nil {
	283	+	return err
	284	+	}
230	285		}
231	286		}
232	287

@@ -237,8 +292,12 @@ func (c *Coordinator) Recover(ctx context.Context) error {
237	292		return err
238	293		}
239	294		for _, tombstone := range page.Tombstones {
	295	+	// A fenced tombstone is never deleted here: the next startup or an
	296	+	// explicit operator action must still be able to observe it.
240	297		if err := c.recoverDeletionTombstone(ctx, tombstone); err != nil {
241		-	return err
	298	+	if err := c.fenceSession(tombstone.Name, "deletion-tombstone", degradedReasonTombstone, err); err != nil {
	299	+	return err
	300	+	}
242	301		}
243	302		}
244	303		if page.Done {

@@ -251,9 +310,47 @@ func (c *Coordinator) Recover(ctx context.Context) error {
251	310		}
252	311		}
253	312
	313	+	// fenceSession returns cause unchanged when it is not session-scoped, so the
	314	+	// caller aborts startup. For a session-scoped cause it records the conflict and
	315	+	// durably marks the session degraded when the record can represent that state.
	316	+	func (c *Coordinator) fenceSession(name, kind, reason string, cause error) error {
	317	+	if !isSessionScopedConflict(cause) {
	318	+	return cause
	319	+	}
	320	+	c.conflictMu.Lock()
	321	+	c.conflicts = append(c.conflicts, RecoveryConflict{Session: name, Kind: kind, Err: cause})
	322	+	c.conflictMu.Unlock()
	323	+
	324	+	unlock := c.locks.Lock([]string{name})
	325	+	defer unlock()
	326	+	record, ok := c.catalogue.Record(name)
	327	+	if !ok \|\| record.RecoveryState != domain.RecoveryHealthy {
	328	+	// Fresh records cannot represent degraded state (it requires a committed
	329	+	// checkpoint), and deleting or already degraded records must keep the
	330	+	// state their own protocol owns. Fencing is then the recorded skip.
	331	+	return nil
	332	+	}
	333	+	record.RecoveryState = domain.RecoveryDegraded
	334	+	record.DegradedReason = reason
	335	+	if err := record.Validate(); err != nil {
	336	+	return nil
	337	+	}
	338	+	// A catalogue write failure here is infrastructure, not one session's data.
	339	+	return c.catalogue.Replace(name, record)
	340	+	}
	341	+
	342	+	// isSessionScopedConflict reports whether err proves only that one session's
	343	+	// durable state is self-inconsistent. Everything else - IO, decode, traversal,
	344	+	// context cancellation - is treated as infrastructure and aborts startup.
	345	+	func isSessionScopedConflict(err error) bool {
	346	+	return errors.Is(err, ErrDiscardConflict) \|\| errors.Is(err, ErrDiscardIntentInvalid) \|\|
	347	+	errors.Is(err, ErrDeletionTombstoneConflict) \|\| errors.Is(err, ErrDeletionTombstoneInvalid) \|\|
	348	+	errors.Is(err, ErrSessionRecordInvalid)
	349	+	}
	350	+
254	351		func (c *Coordinator) recoverDeletionTombstone(ctx context.Context, tombstone domain.DeletionTombstone) error {
255	352		if err := tombstone.Validate(); err != nil {
256		-	return err
	353	+	return fmt.Errorf("%w: %w", ErrDeletionTombstoneInvalid, err)
257	354		}
258	355		unlock := c.locks.Lock([]string{tombstone.Name})
259	356		defer unlock()

@@ -264,7 +361,7 @@ func (c *Coordinator) recoverDeletionTombstone(ctx context.Context, tombstone do
264	361		case record.IncarnationID != tombstone.IncarnationID:
265	362		includeLegacyName = false
266	363		case record.RecoveryState != domain.RecoveryDeleting:
267		-	return fmt.Errorf("recovery: deletion tombstone conflicts with non-deleting session %q", tombstone.Name)
	364	+	return fmt.Errorf("%w: non-deleting session %q", ErrDeletionTombstoneConflict, tombstone.Name)
268	365		}
269	366		}
270	367		if err := c.repository.QuarantineDeletionSources(ctx, tombstone, includeLegacyName); err != nil {


@@ -83,12 +83,54 @@ func compactStrings(values []string) []string {
83		return out	83		return out
84		}	84		}
85			85
			86	+	var (
			87	+	// ErrDeletionTombstoneInvalid marks a tombstone whose own identity cannot be
			88	+	// used. ErrDeletionTombstoneConflict marks a tombstone that disagrees with a
			89	+	// live catalogue record. Both are session-scoped: they fence one session
			90	+	// instead of failing startup.
			91	+	ErrDeletionTombstoneInvalid = errors.New("recovery: deletion tombstone is unusable")
			92	+	ErrDeletionTombstoneConflict = errors.New("recovery: deletion tombstone conflicts with catalogue")
			93	+	// ErrSessionRecordInvalid marks a catalogue record whose own contents cannot
			94	+	// be advanced deterministically.
			95	+	ErrSessionRecordInvalid = errors.New("recovery: catalogue record is unusable")
			96	+	)
			97	+
			98	+	// Degraded reasons are content-free by construction: they never carry paths,
			99	+	// terminal content, or operator input.
			100	+	const (
			101	+	degradedReasonDiscardIntent = "discard intent conflict"
			102	+	degradedReasonTombstone = "deletion tombstone conflict"
			103	+	degradedReasonRecord = "catalogue record conflict"
			104	+	)
			105	+
86		type Coordinator struct {	106		type Coordinator struct {
87		catalogue ports.Catalogue	107		catalogue ports.Catalogue
88		repository ports.SnapshotRepository	108		repository ports.SnapshotRepository
89		journal ports.RecoveryJournal	109		journal ports.RecoveryJournal
90		locks *KeyLocks	110		locks *KeyLocks
91		random io.Reader	111		random io.Reader
			112	+
			113	+	conflictMu sync.Mutex
			114	+	conflicts []RecoveryConflict
			115	+	}
			116	+
			117	+	// RecoveryConflict is one session-scoped item Recover fenced off. Recover keeps
			118	+	// the durable source (intent or tombstone) so a later operator action can still
			119	+	// resolve it; startup continues for every other session.
			120	+	type RecoveryConflict struct {
			121	+	Session string
			122	+	Kind string
			123	+	Err error
			124	+	}
			125	+
			126	+	// Conflicts returns the session-scoped items fenced by the last Recover calls.
			127	+	func (c *Coordinator) Conflicts() []RecoveryConflict {
			128	+	if c == nil {
			129	+	return nil
			130	+	}
			131	+	c.conflictMu.Lock()
			132	+	defer c.conflictMu.Unlock()
			133	+	return append([]RecoveryConflict(nil), c.conflicts...)
92		}	134		}
93			135
94		func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, journal ports.RecoveryJournal, random io.Reader) *Coordinator {	136		func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, journal ports.RecoveryJournal, random io.Reader) *Coordinator {

@@ -180,7 +222,7 @@ func (c *Coordinator) deleteLocked(ctx context.Context, record domain.CatalogueR
180		record.RecoveryState = domain.RecoveryDeleting	222		record.RecoveryState = domain.RecoveryDeleting
181		record.DegradedReason = ""	223		record.DegradedReason = ""
182		if err := record.Validate(); err != nil {	224		if err := record.Validate(); err != nil {
183	-	return fmt.Errorf("recovery: invalid deleting record: %w", err)	225	+	return fmt.Errorf("%w: invalid deleting record: %w", ErrSessionRecordInvalid, err)
184		}	226		}
185		if err := c.catalogue.Replace(record.Name, record); err != nil {	227		if err := c.catalogue.Replace(record.Name, record); err != nil {
186		return err	228		return err

@@ -202,6 +244,12 @@ func (c *Coordinator) deleteLocked(ctx context.Context, record domain.CatalogueR
202		// Recover rolls deleting records and strict deletion tombstones forward. The	244		// Recover rolls deleting records and strict deletion tombstones forward. The
203		// two sources are enumerated independently because catalogue removal precedes	245		// two sources are enumerated independently because catalogue removal precedes
204		// tombstone removal in the commit protocol.	246		// tombstone removal in the commit protocol.
			247	+	//
			248	+	// A per-item failure that only proves one session's durable state disagrees
			249	+	// with itself fences that session and lets startup continue: healthy sessions
			250	+	// must still restore next to a corrupt neighbour. Infrastructure failures
			251	+	// (listing, journal, repository, or catalogue IO, and a non-advancing cursor)
			252	+	// still abort startup, because they say nothing about any single session.
205		func (c *Coordinator) Recover(ctx context.Context) error {	253		func (c *Coordinator) Recover(ctx context.Context) error {
206		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.journal == nil \|\| c.locks == nil {	254		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.journal == nil \|\| c.locks == nil {
207		return errors.New("recovery: incomplete coordinator dependencies")	255		return errors.New("recovery: incomplete coordinator dependencies")

@@ -211,8 +259,12 @@ func (c *Coordinator) Recover(ctx context.Context) error {
211		return err	259		return err
212		}	260		}
213		for _, intent := range intents {	261		for _, intent := range intents {
			262	+	// The intent is deliberately retained for a fenced session: only a
			263	+	// successful roll-forward may remove it.
214		if err := c.recoverDiscard(ctx, intent); err != nil {	264		if err := c.recoverDiscard(ctx, intent); err != nil {
215	-	return err	265	+	if err := c.fenceSession(intent.SessionName, "discard-intent", degradedReasonDiscardIntent, err); err != nil {
			266	+	return err
			267	+	}
216		}	268		}
217		}	269		}
218		for _, candidate := range c.catalogue.Records() {	270		for _, candidate := range c.catalogue.Records() {

@@ -220,13 +272,16 @@ func (c *Coordinator) Recover(ctx context.Context) error {
220		continue	272		continue
221		}	273		}
222		unlock := c.locks.Lock([]string{candidate.Name})	274		unlock := c.locks.Lock([]string{candidate.Name})
			275	+	var deleteErr error
223		current, ok := c.catalogue.Record(candidate.Name)	276		current, ok := c.catalogue.Record(candidate.Name)
224		if ok && current.IncarnationID == candidate.IncarnationID && current.RecoveryState == domain.RecoveryDeleting {	277		if ok && current.IncarnationID == candidate.IncarnationID && current.RecoveryState == domain.RecoveryDeleting {
225	-	err = c.deleteLocked(ctx, current)	278	+	deleteErr = c.deleteLocked(ctx, current)
226		}	279		}
227		unlock()	280		unlock()
228	-	if err != nil {	281	+	if deleteErr != nil {
229	-	return err	282	+	if err := c.fenceSession(candidate.Name, "catalogue-record", degradedReasonRecord, deleteErr); err != nil {
			283	+	return err
			284	+	}
230		}	285		}
231		}	286		}
232			287

@@ -237,8 +292,12 @@ func (c *Coordinator) Recover(ctx context.Context) error {
237		return err	292		return err
238		}	293		}
239		for _, tombstone := range page.Tombstones {	294		for _, tombstone := range page.Tombstones {
			295	+	// A fenced tombstone is never deleted here: the next startup or an
			296	+	// explicit operator action must still be able to observe it.
240		if err := c.recoverDeletionTombstone(ctx, tombstone); err != nil {	297		if err := c.recoverDeletionTombstone(ctx, tombstone); err != nil {
241	-	return err	298	+	if err := c.fenceSession(tombstone.Name, "deletion-tombstone", degradedReasonTombstone, err); err != nil {
			299	+	return err
			300	+	}
242		}	301		}
243		}	302		}
244		if page.Done {	303		if page.Done {

@@ -251,9 +310,47 @@ func (c *Coordinator) Recover(ctx context.Context) error {
251		}	310		}
252		}	311		}
253			312
			313	+	// fenceSession returns cause unchanged when it is not session-scoped, so the
			314	+	// caller aborts startup. For a session-scoped cause it records the conflict and
			315	+	// durably marks the session degraded when the record can represent that state.
			316	+	func (c *Coordinator) fenceSession(name, kind, reason string, cause error) error {
			317	+	if !isSessionScopedConflict(cause) {
			318	+	return cause
			319	+	}
			320	+	c.conflictMu.Lock()
			321	+	c.conflicts = append(c.conflicts, RecoveryConflict{Session: name, Kind: kind, Err: cause})
			322	+	c.conflictMu.Unlock()
			323	+
			324	+	unlock := c.locks.Lock([]string{name})
			325	+	defer unlock()
			326	+	record, ok := c.catalogue.Record(name)
			327	+	if !ok \|\| record.RecoveryState != domain.RecoveryHealthy {
			328	+	// Fresh records cannot represent degraded state (it requires a committed
			329	+	// checkpoint), and deleting or already degraded records must keep the
			330	+	// state their own protocol owns. Fencing is then the recorded skip.
			331	+	return nil
			332	+	}
			333	+	record.RecoveryState = domain.RecoveryDegraded
			334	+	record.DegradedReason = reason
			335	+	if err := record.Validate(); err != nil {
			336	+	return nil
			337	+	}
			338	+	// A catalogue write failure here is infrastructure, not one session's data.
			339	+	return c.catalogue.Replace(name, record)
			340	+	}
			341	+
			342	+	// isSessionScopedConflict reports whether err proves only that one session's
			343	+	// durable state is self-inconsistent. Everything else - IO, decode, traversal,
			344	+	// context cancellation - is treated as infrastructure and aborts startup.
			345	+	func isSessionScopedConflict(err error) bool {
			346	+	return errors.Is(err, ErrDiscardConflict) \|\| errors.Is(err, ErrDiscardIntentInvalid) \|\|
			347	+	errors.Is(err, ErrDeletionTombstoneConflict) \|\| errors.Is(err, ErrDeletionTombstoneInvalid) \|\|
			348	+	errors.Is(err, ErrSessionRecordInvalid)
			349	+	}
			350	+
254		func (c *Coordinator) recoverDeletionTombstone(ctx context.Context, tombstone domain.DeletionTombstone) error {	351		func (c *Coordinator) recoverDeletionTombstone(ctx context.Context, tombstone domain.DeletionTombstone) error {
255		if err := tombstone.Validate(); err != nil {	352		if err := tombstone.Validate(); err != nil {
256	-	return err	353	+	return fmt.Errorf("%w: %w", ErrDeletionTombstoneInvalid, err)
257		}	354		}
258		unlock := c.locks.Lock([]string{tombstone.Name})	355		unlock := c.locks.Lock([]string{tombstone.Name})
259		defer unlock()	356		defer unlock()

@@ -264,7 +361,7 @@ func (c *Coordinator) recoverDeletionTombstone(ctx context.Context, tombstone do
264		case record.IncarnationID != tombstone.IncarnationID:	361		case record.IncarnationID != tombstone.IncarnationID:
265		includeLegacyName = false	362		includeLegacyName = false
266		case record.RecoveryState != domain.RecoveryDeleting:	363		case record.RecoveryState != domain.RecoveryDeleting:
267	-	return fmt.Errorf("recovery: deletion tombstone conflicts with non-deleting session %q", tombstone.Name)	364	+	return fmt.Errorf("%w: non-deleting session %q", ErrDeletionTombstoneConflict, tombstone.Name)
268		}	365		}
269		}	366		}
270		if err := c.repository.QuarantineDeletionSources(ctx, tombstone, includeLegacyName); err != nil {	367		if err := c.repository.QuarantineDeletionSources(ctx, tombstone, includeLegacyName); err != nil {

M internal/usecase/recovery/discard.go +12 -5


@@ -28,6 +28,7 @@ var (
28	28		ErrRecoveryRecordNotFound = errors.New("recovery: session not found")
29	29		ErrSessionNotDegraded = errors.New("recovery: session is not degraded")
30	30		ErrDiscardConflict = errors.New("recovery: discard intent conflicts with catalogue")
	31	+	ErrDiscardIntentInvalid = errors.New("recovery: discard intent is unusable")
31	32		)
32	33
33	34		// Retry directly validates the committed checkpoint. A failed validation is

@@ -212,7 +213,7 @@ func (c *Coordinator) recoverDiscardLocked(ctx context.Context, intent domain.Di
212	213		}
213	214		next := freshReplacement(intent)
214	215		if err := next.Validate(); err != nil {
215		-	return fmt.Errorf("recovery: invalid discard replacement: %w", err)
	216	+	return fmt.Errorf("%w: invalid discard replacement: %w", ErrDiscardIntentInvalid, err)
216	217		}
217	218		switch current.IncarnationID {
218	219		case intent.OldIncarnation:

@@ -220,8 +221,14 @@ func (c *Coordinator) recoverDiscardLocked(ctx context.Context, intent domain.Di
220	221		return fmt.Errorf("%w: old record changed", ErrDiscardConflict)
221	222		}
222	223		case intent.NewIncarnation:
223		-	if !catalogueRecordsEqual(current, next) {
224		-	return fmt.Errorf("%w: replacement record changed", ErrDiscardConflict)
	224	+	// Step 4 is already durable. The replacement is live from this point on,
	225	+	// so it may legitimately have advanced to a committed checkpoint before a
	226	+	// later step failed. Only its identity may be required here: demanding
	227	+	// byte equality with the pristine replacement would fail closed forever
	228	+	// on a session that is in fact healthy, and destroy committed state if it
	229	+	// were overwritten. Steps 2-3 and 5 below are idempotent.
	230	+	if current.Name != intent.SessionName {
	231	+	return fmt.Errorf("%w: replacement session identity changed", ErrDiscardConflict)
225	232		}
226	233		default:
227	234		return fmt.Errorf("%w: unexpected incarnation", ErrDiscardConflict)

@@ -243,11 +250,11 @@ func (c *Coordinator) recoverDiscardLocked(ctx context.Context, intent domain.Di
243	250
244	251		func validateDiscardIntent(intent domain.DiscardIntent) error {
245	252		if err := intent.OldRecord.Validate(); err != nil {
246		-	return fmt.Errorf("recovery: invalid discard old record: %w", err)
	253	+	return fmt.Errorf("%w: invalid old record: %w", ErrDiscardIntentInvalid, err)
247	254		}
248	255		if intent.SessionName != intent.OldRecord.Name \|\| intent.OldIncarnation != intent.OldRecord.IncarnationID \|\|
249	256		intent.NewIncarnation == (domain.IncarnationID{}) \|\| intent.NewIncarnation == intent.OldIncarnation \|\| intent.Reason == "" {
250		-	return errors.New("recovery: invalid discard intent")
	257	+	return fmt.Errorf("%w: inconsistent intent fields", ErrDiscardIntentInvalid)
251	258		}
252	259		return nil
253	260		}

 	ErrRecoveryRecordNotFound = errors.New("recovery: session not found")
 	ErrSessionNotDegraded     = errors.New("recovery: session is not degraded")
 	ErrDiscardConflict        = errors.New("recovery: discard intent conflicts with catalogue")
+	ErrDiscardIntentInvalid   = errors.New("recovery: discard intent is unusable")
 )
 // Retry directly validates the committed checkpoint. A failed validation is
 	}
 	next := freshReplacement(intent)
 	if err := next.Validate(); err != nil {
-		return fmt.Errorf("recovery: invalid discard replacement: %w", err)
+		return fmt.Errorf("%w: invalid discard replacement: %w", ErrDiscardIntentInvalid, err)
 	}
 	switch current.IncarnationID {
 	case intent.OldIncarnation:
 			return fmt.Errorf("%w: old record changed", ErrDiscardConflict)
 		}
 	case intent.NewIncarnation:
-		if !catalogueRecordsEqual(current, next) {
+		// Step 4 is already durable. The replacement is live from this point on,
-			return fmt.Errorf("%w: replacement record changed", ErrDiscardConflict)
+		// so it may legitimately have advanced to a committed checkpoint before a
+		// later step failed. Only its identity may be required here: demanding
+		// byte equality with the pristine replacement would fail closed forever
+		// on a session that is in fact healthy, and destroy committed state if it
+		// were overwritten. Steps 2-3 and 5 below are idempotent.
+		if current.Name != intent.SessionName {
+			return fmt.Errorf("%w: replacement session identity changed", ErrDiscardConflict)
 		}
 	default:
 		return fmt.Errorf("%w: unexpected incarnation", ErrDiscardConflict)
 func validateDiscardIntent(intent domain.DiscardIntent) error {
 	if err := intent.OldRecord.Validate(); err != nil {
-		return fmt.Errorf("recovery: invalid discard old record: %w", err)
+		return fmt.Errorf("%w: invalid old record: %w", ErrDiscardIntentInvalid, err)
 	}
 	if intent.SessionName != intent.OldRecord.Name || intent.OldIncarnation != intent.OldRecord.IncarnationID ||
 		intent.NewIncarnation == (domain.IncarnationID{}) || intent.NewIncarnation == intent.OldIncarnation || intent.Reason == "" {
-		return errors.New("recovery: invalid discard intent")
+		return fmt.Errorf("%w: inconsistent intent fields", ErrDiscardIntentInvalid)
 	}
 	return nil
 }

M internal/usecase/recovery/transactions_test.go +117 -3


@@ -92,6 +92,7 @@ type transactionRepository struct {
92	92		tombstones map[domain.IncarnationID]domain.DeletionTombstone
93	93		generations map[domain.CheckpointRef]ports.SnapshotGeneration
94	94		loadErr error
	95	+	quarantineErr error
95	96		events []string
96	97		quarantine []bool
97	98		descriptors []domain.QuarantineDescriptor

@@ -163,7 +164,7 @@ func (r *transactionRepository) QuarantineDeletionSources(_ context.Context, _ d
163	164		defer r.mu.Unlock()
164	165		r.events = append(r.events, "quarantine")
165	166		r.quarantine = append(r.quarantine, includeLegacyName)
166		-	return nil
	167	+	return r.quarantineErr
167	168		}
168	169		func (r *transactionRepository) DeleteDeletionTombstone(_ context.Context, id domain.IncarnationID) error {
169	170		r.mu.Lock()

@@ -280,6 +281,8 @@ func TestDegradedExportIsReadOnly(t *testing.T) {
280	281		require.Empty(t, catalogue.events)
281	282		}
282	283
	284	+	// A third incarnation still fails closed for that session: its intent and
	285	+	// record are retained untouched, but startup is no longer aborted.
283	286		func TestDiscardIncarnationConflict(t *testing.T) {
284	287		old := degradedTransactionRecord()
285	288		intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}

@@ -288,11 +291,122 @@ func TestDiscardIncarnationConflict(t *testing.T) {
288	291		catalogue := newTransactionCatalogue(conflict)
289	292		journal := newTransactionJournal()
290	293		journal.intents[old.IncarnationID] = intent
291		-	coordinator := NewCoordinator(catalogue, newTransactionRepository(), journal, nil)
	294	+	repository := newTransactionRepository()
	295	+	coordinator := NewCoordinator(catalogue, repository, journal, nil)
292	296
293		-	require.Error(t, coordinator.Recover(context.Background()))
	297	+	require.NoError(t, coordinator.Recover(context.Background()))
294	298		require.Equal(t, conflict, catalogue.records[old.Name])
295	299		require.Contains(t, journal.intents, old.IncarnationID)
	300	+	require.Empty(t, repository.events)
	301	+	conflicts := coordinator.Conflicts()
	302	+	require.Len(t, conflicts, 1)
	303	+	require.Equal(t, old.Name, conflicts[0].Session)
	304	+	require.Equal(t, "discard-intent", conflicts[0].Kind)
	305	+	require.ErrorIs(t, conflicts[0].Err, ErrDiscardConflict)
	306	+	}
	307	+
	308	+	// A discard whose replacement is already durable, live, and checkpointed must
	309	+	// roll forward. Requiring byte equality with the pristine replacement would
	310	+	// leave the intent unresolvable and abort every future startup.
	311	+	func TestDiscardRecoveryRollsForwardEvolvedReplacement(t *testing.T) {
	312	+	old := degradedTransactionRecord()
	313	+	intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}
	314	+	evolved := freshReplacement(intent)
	315	+	evolved.RecoveryState = domain.RecoveryHealthy
	316	+	evolved.Committed = &domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
	317	+	evolved.UpdatedAt = 1234
	318	+	evolved.LastUsedSeq = 9
	319	+	evolved.TabNames = []string{"one"}
	320	+	require.NoError(t, evolved.Validate())
	321	+
	322	+	catalogue := newTransactionCatalogue(evolved)
	323	+	journal := newTransactionJournal()
	324	+	journal.intents[old.IncarnationID] = intent
	325	+	repository := newTransactionRepository()
	326	+	coordinator := NewCoordinator(catalogue, repository, journal, nil)
	327	+
	328	+	require.NoError(t, coordinator.Recover(context.Background()))
	329	+	require.Empty(t, coordinator.Conflicts())
	330	+	require.Equal(t, evolved, catalogue.records[old.Name], "committed replacement state must survive roll-forward")
	331	+	require.Empty(t, catalogue.events, "the live replacement record must not be rewritten")
	332	+	require.Equal(t, []domain.QuarantineDescriptor{quarantineDescriptor(intent)}, repository.descriptors)
	333	+	require.Equal(t, []domain.IncarnationID{intent.OldIncarnation}, repository.quarantinedIDs)
	334	+	require.NotContains(t, journal.intents, old.IncarnationID)
	335	+	}
	336	+
	337	+	// Goal 3: one session's self-inconsistent durable state must not stop a healthy
	338	+	// neighbour from being recovered, and must not silently resolve itself either.
	339	+	func TestRecoverFencesOneSessionAndContinues(t *testing.T) {
	340	+	ctx := context.Background()
	341	+	broken := degradedTransactionRecord()
	342	+	intent := domain.DiscardIntent{OldRecord: broken, OldIncarnation: broken.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: broken.Name, Reason: "discard"}
	343	+	conflict := broken
	344	+	conflict.IncarnationID = domain.IncarnationID{3}
	345	+	deleting := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{4}, RecoveryState: domain.RecoveryDeleting}
	346	+
	347	+	catalogue := newTransactionCatalogue(conflict, deleting)
	348	+	journal := newTransactionJournal()
	349	+	journal.intents[broken.IncarnationID] = intent
	350	+	repository := newTransactionRepository()
	351	+	coordinator := NewCoordinator(catalogue, repository, journal, nil)
	352	+
	353	+	require.NoError(t, coordinator.Recover(ctx))
	354	+	require.NotContains(t, catalogue.records, deleting.Name, "healthy deletion must complete next to a fenced session")
	355	+	require.Empty(t, repository.tombstones)
	356	+	require.Equal(t, conflict, catalogue.records[broken.Name])
	357	+	require.Contains(t, journal.intents, broken.IncarnationID)
	358	+	require.Len(t, coordinator.Conflicts(), 1)
	359	+	}
	360	+
	361	+	// A tombstone that disagrees with a live healthy record degrades exactly that
	362	+	// record, keeps the tombstone, and leaves every sibling tombstone processed.
	363	+	func TestRecoverDegradesSessionOnTombstoneConflict(t *testing.T) {
	364	+	ctx := context.Background()
	365	+	live := domain.CatalogueRecord{
	366	+	Name: "live", IncarnationID: domain.IncarnationID{5}, RecoveryState: domain.RecoveryHealthy,
	367	+	Committed: &domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{2}},
	368	+	}
	369	+	conflicting := domain.DeletionTombstone{Name: live.Name, IncarnationID: live.IncarnationID}
	370	+	orphan := domain.DeletionTombstone{Name: "gone", IncarnationID: domain.IncarnationID{6}}
	371	+
	372	+	catalogue := newTransactionCatalogue(live)
	373	+	repository := newTransactionRepository()
	374	+	repository.tombstones[conflicting.IncarnationID] = conflicting
	375	+	repository.tombstones[orphan.IncarnationID] = orphan
	376	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	377	+
	378	+	require.NoError(t, coordinator.Recover(ctx))
	379	+	got := catalogue.records[live.Name]
	380	+	require.Equal(t, domain.RecoveryDegraded, got.RecoveryState)
	381	+	require.Equal(t, degradedReasonTombstone, got.DegradedReason)
	382	+	require.Equal(t, live.Committed, got.Committed)
	383	+	require.Contains(t, repository.tombstones, conflicting.IncarnationID, "a fenced tombstone is never removed")
	384	+	require.NotContains(t, repository.tombstones, orphan.IncarnationID)
	385	+	conflicts := coordinator.Conflicts()
	386	+	require.Len(t, conflicts, 1)
	387	+	require.Equal(t, "deletion-tombstone", conflicts[0].Kind)
	388	+	require.ErrorIs(t, conflicts[0].Err, ErrDeletionTombstoneConflict)
	389	+	}
	390	+
	391	+	// Infrastructure failures still abort startup: they say nothing about any one
	392	+	// session, so continuing would mutate durable state on a broken substrate.
	393	+	func TestRecoverFailsOnInfrastructureErrors(t *testing.T) {
	394	+	ctx := context.Background()
	395	+	cause := errors.New("io failure")
	396	+	tombstone := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{4}}
	397	+
	398	+	t.Run("journal listing", func(t *testing.T) {
	399	+	coordinator := NewCoordinator(newTransactionCatalogue(), newTransactionRepository(), journalStub{err: cause}, nil)
	400	+	require.ErrorIs(t, coordinator.Recover(ctx), cause)
	401	+	})
	402	+	t.Run("quarantine write", func(t *testing.T) {
	403	+	repository := newTransactionRepository()
	404	+	repository.tombstones[tombstone.IncarnationID] = tombstone
	405	+	repository.quarantineErr = cause
	406	+	coordinator := NewCoordinator(newTransactionCatalogue(), repository, journalStub{}, nil)
	407	+	require.ErrorIs(t, coordinator.Recover(ctx), cause)
	408	+	require.Empty(t, coordinator.Conflicts())
	409	+	})
296	410		}
297	411
298	412		func TestDeleteRecoveryAfterCatalogueRemoval(t *testing.T) {


@@ -92,6 +92,7 @@ type transactionRepository struct {
92		tombstones map[domain.IncarnationID]domain.DeletionTombstone	92		tombstones map[domain.IncarnationID]domain.DeletionTombstone
93		generations map[domain.CheckpointRef]ports.SnapshotGeneration	93		generations map[domain.CheckpointRef]ports.SnapshotGeneration
94		loadErr error	94		loadErr error
			95	+	quarantineErr error
95		events []string	96		events []string
96		quarantine []bool	97		quarantine []bool
97		descriptors []domain.QuarantineDescriptor	98		descriptors []domain.QuarantineDescriptor

@@ -163,7 +164,7 @@ func (r *transactionRepository) QuarantineDeletionSources(_ context.Context, _ d
163		defer r.mu.Unlock()	164		defer r.mu.Unlock()
164		r.events = append(r.events, "quarantine")	165		r.events = append(r.events, "quarantine")
165		r.quarantine = append(r.quarantine, includeLegacyName)	166		r.quarantine = append(r.quarantine, includeLegacyName)
166	-	return nil	167	+	return r.quarantineErr
167		}	168		}
168		func (r *transactionRepository) DeleteDeletionTombstone(_ context.Context, id domain.IncarnationID) error {	169		func (r *transactionRepository) DeleteDeletionTombstone(_ context.Context, id domain.IncarnationID) error {
169		r.mu.Lock()	170		r.mu.Lock()

@@ -280,6 +281,8 @@ func TestDegradedExportIsReadOnly(t *testing.T) {
280		require.Empty(t, catalogue.events)	281		require.Empty(t, catalogue.events)
281		}	282		}
282			283
			284	+	// A third incarnation still fails closed for that session: its intent and
			285	+	// record are retained untouched, but startup is no longer aborted.
283		func TestDiscardIncarnationConflict(t *testing.T) {	286		func TestDiscardIncarnationConflict(t *testing.T) {
284		old := degradedTransactionRecord()	287		old := degradedTransactionRecord()
285		intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}	288		intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}

@@ -288,11 +291,122 @@ func TestDiscardIncarnationConflict(t *testing.T) {
288		catalogue := newTransactionCatalogue(conflict)	291		catalogue := newTransactionCatalogue(conflict)
289		journal := newTransactionJournal()	292		journal := newTransactionJournal()
290		journal.intents[old.IncarnationID] = intent	293		journal.intents[old.IncarnationID] = intent
291	-	coordinator := NewCoordinator(catalogue, newTransactionRepository(), journal, nil)	294	+	repository := newTransactionRepository()
			295	+	coordinator := NewCoordinator(catalogue, repository, journal, nil)
292			296
293	-	require.Error(t, coordinator.Recover(context.Background()))	297	+	require.NoError(t, coordinator.Recover(context.Background()))
294		require.Equal(t, conflict, catalogue.records[old.Name])	298		require.Equal(t, conflict, catalogue.records[old.Name])
295		require.Contains(t, journal.intents, old.IncarnationID)	299		require.Contains(t, journal.intents, old.IncarnationID)
			300	+	require.Empty(t, repository.events)
			301	+	conflicts := coordinator.Conflicts()
			302	+	require.Len(t, conflicts, 1)
			303	+	require.Equal(t, old.Name, conflicts[0].Session)
			304	+	require.Equal(t, "discard-intent", conflicts[0].Kind)
			305	+	require.ErrorIs(t, conflicts[0].Err, ErrDiscardConflict)
			306	+	}
			307	+
			308	+	// A discard whose replacement is already durable, live, and checkpointed must
			309	+	// roll forward. Requiring byte equality with the pristine replacement would
			310	+	// leave the intent unresolvable and abort every future startup.
			311	+	func TestDiscardRecoveryRollsForwardEvolvedReplacement(t *testing.T) {
			312	+	old := degradedTransactionRecord()
			313	+	intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}
			314	+	evolved := freshReplacement(intent)
			315	+	evolved.RecoveryState = domain.RecoveryHealthy
			316	+	evolved.Committed = &domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
			317	+	evolved.UpdatedAt = 1234
			318	+	evolved.LastUsedSeq = 9
			319	+	evolved.TabNames = []string{"one"}
			320	+	require.NoError(t, evolved.Validate())
			321	+
			322	+	catalogue := newTransactionCatalogue(evolved)
			323	+	journal := newTransactionJournal()
			324	+	journal.intents[old.IncarnationID] = intent
			325	+	repository := newTransactionRepository()
			326	+	coordinator := NewCoordinator(catalogue, repository, journal, nil)
			327	+
			328	+	require.NoError(t, coordinator.Recover(context.Background()))
			329	+	require.Empty(t, coordinator.Conflicts())
			330	+	require.Equal(t, evolved, catalogue.records[old.Name], "committed replacement state must survive roll-forward")
			331	+	require.Empty(t, catalogue.events, "the live replacement record must not be rewritten")
			332	+	require.Equal(t, []domain.QuarantineDescriptor{quarantineDescriptor(intent)}, repository.descriptors)
			333	+	require.Equal(t, []domain.IncarnationID{intent.OldIncarnation}, repository.quarantinedIDs)
			334	+	require.NotContains(t, journal.intents, old.IncarnationID)
			335	+	}
			336	+
			337	+	// Goal 3: one session's self-inconsistent durable state must not stop a healthy
			338	+	// neighbour from being recovered, and must not silently resolve itself either.
			339	+	func TestRecoverFencesOneSessionAndContinues(t *testing.T) {
			340	+	ctx := context.Background()
			341	+	broken := degradedTransactionRecord()
			342	+	intent := domain.DiscardIntent{OldRecord: broken, OldIncarnation: broken.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: broken.Name, Reason: "discard"}
			343	+	conflict := broken
			344	+	conflict.IncarnationID = domain.IncarnationID{3}
			345	+	deleting := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{4}, RecoveryState: domain.RecoveryDeleting}
			346	+
			347	+	catalogue := newTransactionCatalogue(conflict, deleting)
			348	+	journal := newTransactionJournal()
			349	+	journal.intents[broken.IncarnationID] = intent
			350	+	repository := newTransactionRepository()
			351	+	coordinator := NewCoordinator(catalogue, repository, journal, nil)
			352	+
			353	+	require.NoError(t, coordinator.Recover(ctx))
			354	+	require.NotContains(t, catalogue.records, deleting.Name, "healthy deletion must complete next to a fenced session")
			355	+	require.Empty(t, repository.tombstones)
			356	+	require.Equal(t, conflict, catalogue.records[broken.Name])
			357	+	require.Contains(t, journal.intents, broken.IncarnationID)
			358	+	require.Len(t, coordinator.Conflicts(), 1)
			359	+	}
			360	+
			361	+	// A tombstone that disagrees with a live healthy record degrades exactly that
			362	+	// record, keeps the tombstone, and leaves every sibling tombstone processed.
			363	+	func TestRecoverDegradesSessionOnTombstoneConflict(t *testing.T) {
			364	+	ctx := context.Background()
			365	+	live := domain.CatalogueRecord{
			366	+	Name: "live", IncarnationID: domain.IncarnationID{5}, RecoveryState: domain.RecoveryHealthy,
			367	+	Committed: &domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{2}},
			368	+	}
			369	+	conflicting := domain.DeletionTombstone{Name: live.Name, IncarnationID: live.IncarnationID}
			370	+	orphan := domain.DeletionTombstone{Name: "gone", IncarnationID: domain.IncarnationID{6}}
			371	+
			372	+	catalogue := newTransactionCatalogue(live)
			373	+	repository := newTransactionRepository()
			374	+	repository.tombstones[conflicting.IncarnationID] = conflicting
			375	+	repository.tombstones[orphan.IncarnationID] = orphan
			376	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
			377	+
			378	+	require.NoError(t, coordinator.Recover(ctx))
			379	+	got := catalogue.records[live.Name]
			380	+	require.Equal(t, domain.RecoveryDegraded, got.RecoveryState)
			381	+	require.Equal(t, degradedReasonTombstone, got.DegradedReason)
			382	+	require.Equal(t, live.Committed, got.Committed)
			383	+	require.Contains(t, repository.tombstones, conflicting.IncarnationID, "a fenced tombstone is never removed")
			384	+	require.NotContains(t, repository.tombstones, orphan.IncarnationID)
			385	+	conflicts := coordinator.Conflicts()
			386	+	require.Len(t, conflicts, 1)
			387	+	require.Equal(t, "deletion-tombstone", conflicts[0].Kind)
			388	+	require.ErrorIs(t, conflicts[0].Err, ErrDeletionTombstoneConflict)
			389	+	}
			390	+
			391	+	// Infrastructure failures still abort startup: they say nothing about any one
			392	+	// session, so continuing would mutate durable state on a broken substrate.
			393	+	func TestRecoverFailsOnInfrastructureErrors(t *testing.T) {
			394	+	ctx := context.Background()
			395	+	cause := errors.New("io failure")
			396	+	tombstone := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{4}}
			397	+
			398	+	t.Run("journal listing", func(t *testing.T) {
			399	+	coordinator := NewCoordinator(newTransactionCatalogue(), newTransactionRepository(), journalStub{err: cause}, nil)
			400	+	require.ErrorIs(t, coordinator.Recover(ctx), cause)
			401	+	})
			402	+	t.Run("quarantine write", func(t *testing.T) {
			403	+	repository := newTransactionRepository()
			404	+	repository.tombstones[tombstone.IncarnationID] = tombstone
			405	+	repository.quarantineErr = cause
			406	+	coordinator := NewCoordinator(newTransactionCatalogue(), repository, journalStub{}, nil)
			407	+	require.ErrorIs(t, coordinator.Recover(ctx), cause)
			408	+	require.Empty(t, coordinator.Conflicts())
			409	+	})
296		}	410		}
297			411
298		func TestDeleteRecoveryAfterCatalogueRemoval(t *testing.T) {	412		func TestDeleteRecoveryAfterCatalogueRemoval(t *testing.T) {