97f1e81 fix(recovery): fence startup garbage collection · vev

M internal/app/daemon_integration_test.go +8 -9


@@ -548,17 +548,11 @@ func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
548	548		releaseMaintenance := func() { releaseOnce.Do(func() { close(maintenanceRepository.release) }) }
549	549		t.Cleanup(releaseMaintenance)
550	550
551		-	listenerReady := make(chan struct{})
552	551		callbackReturned := make(chan struct{})
553	552		wrapperReturned := make(chan error, 1)
554	553		go func() {
555	554		wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
556	555		defer close(callbackReturned)
557		-	listener, err := ipc.Listen(runtimeDir)
558		-	if err != nil {
559		-	return err
560		-	}
561		-	close(listenerReady)
562	556		d := daemon.New(
563	557		pty.NewFactory(),
564	558		shutdownClock,

@@ -567,14 +561,19 @@ func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
567	561		daemon.WithDurableMaintenance(catalogue, maintenanceRepository),
568	562		daemon.WithCatalogue(catalogue, nil),
569	563		)
	564	+	if err := d.CollectStartupGarbage(ctx); err != nil {
	565	+	return err
	566	+	}
	567	+	listener, err := ipc.Listen(runtimeDir)
	568	+	if err != nil {
	569	+	return err
	570	+	}
570	571		return d.Serve(ctx, listener)
571	572		})
572	573		}()
573	574
574		-	awaitLifecycleStage(t, listenerReady, "daemon listener")
575		-	awaitLifecycleStage(t, maintenanceRepository.entered, "maintenance repository call")
	575	+	awaitLifecycleStage(t, maintenanceRepository.entered, "pre-publication maintenance repository call")
576	576		cancel()
577		-	shutdownClock.nextTimer(t).fire()
578	577		select {
579	578		case <-callbackReturned:
580	579		t.Fatal("Serve callback returned after shutdown deadline while maintenance was blocked")

 	releaseMaintenance := func() { releaseOnce.Do(func() { close(maintenanceRepository.release) }) }
 	t.Cleanup(releaseMaintenance)
-	listenerReady := make(chan struct{})
 	callbackReturned := make(chan struct{})
 	wrapperReturned := make(chan error, 1)
 	go func() {
 		wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
 			defer close(callbackReturned)
-			listener, err := ipc.Listen(runtimeDir)
-			if err != nil {
-				return err
-			}
-			close(listenerReady)
 			d := daemon.New(
 				pty.NewFactory(),
 				shutdownClock,
 				daemon.WithDurableMaintenance(catalogue, maintenanceRepository),
 				daemon.WithCatalogue(catalogue, nil),
 			)
+			if err := d.CollectStartupGarbage(ctx); err != nil {
+				return err
+			}
+			listener, err := ipc.Listen(runtimeDir)
+			if err != nil {
+				return err
+			}
 			return d.Serve(ctx, listener)
 		})
 	}()
-	awaitLifecycleStage(t, listenerReady, "daemon listener")
+	awaitLifecycleStage(t, maintenanceRepository.entered, "pre-publication maintenance repository call")
-	awaitLifecycleStage(t, maintenanceRepository.entered, "maintenance repository call")
 	cancel()
-	shutdownClock.nextTimer(t).fire()
 	select {
 	case <-callbackReturned:
 		t.Fatal("Serve callback returned after shutdown deadline while maintenance was blocked")

M internal/app/run.go +15 -2


@@ -521,9 +521,13 @@ func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord
521	521
522	522		func constructDaemonBeforeSocketPublication(
523	523		construct func() *daemon.Daemon,
	524	+	prepare func(*daemon.Daemon) error,
524	525		listen func() (ports.Listener, error),
525	526		) (*daemon.Daemon, ports.Listener, error) {
526	527		d := construct()
	528	+	if err := prepare(d); err != nil {
	529	+	return d, nil, err
	530	+	}
527	531		ln, err := listen()
528	532		return d, ln, err
529	533		}

@@ -601,12 +605,21 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
601	605		// publication. Phase 3 snapshot restoration remains asynchronous in Serve.
602	606		d, ln, err := constructDaemonBeforeSocketPublication(
603	607		func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
	608	+	func(d *daemon.Daemon) error {
	609	+	if err := d.CollectStartupGarbage(ctx); err != nil {
	610	+	// GC is best-effort, but it is fully finished before socket
	611	+	// publication. A failed pass leaves durable state untouched and
	612	+	// restoration retains its per-session failure isolation.
	613	+	log.Warn("snapshot_garbage_collection_failed", "err", err)
	614	+	}
	615	+	return nil
	616	+	},
604	617		func() (ports.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },
605	618		)
606	619		if err != nil {
607	620		closeErr := opened.Catalogue.Close()
608		-	log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err)
609		-	return errors.Join(fmt.Errorf("vev: daemon listen: %w", err), closeErr)
	621	+	log.Error("daemon startup preparation failed", "socket_dir", ipc.SocketDir(), "err", err)
	622	+	return errors.Join(fmt.Errorf("vev: prepare daemon startup: %w", err), closeErr)
610	623		}
611	624		defer func() { _ = ln.Close() }()
612	625		log.Info("daemon starting", "socket", ln.Addr())

 func constructDaemonBeforeSocketPublication(
 	construct func() *daemon.Daemon,
+	prepare func(*daemon.Daemon) error,
 	listen func() (ports.Listener, error),
 ) (*daemon.Daemon, ports.Listener, error) {
 	d := construct()
+	if err := prepare(d); err != nil {
+		return d, nil, err
+	}
 	ln, err := listen()
 	return d, ln, err
 }
 	// publication. Phase 3 snapshot restoration remains asynchronous in Serve.
 	d, ln, err := constructDaemonBeforeSocketPublication(
 		func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
+		func(d *daemon.Daemon) error {
+			if err := d.CollectStartupGarbage(ctx); err != nil {
+				// GC is best-effort, but it is fully finished before socket
+				// publication. A failed pass leaves durable state untouched and
+				// restoration retains its per-session failure isolation.
+				log.Warn("snapshot_garbage_collection_failed", "err", err)
+			}
+			return nil
+		},
 		func() (ports.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },
 	)
 	if err != nil {
 		closeErr := opened.Catalogue.Close()
-		log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err)
+		log.Error("daemon startup preparation failed", "socket_dir", ipc.SocketDir(), "err", err)
-		return errors.Join(fmt.Errorf("vev: daemon listen: %w", err), closeErr)
+		return errors.Join(fmt.Errorf("vev: prepare daemon startup: %w", err), closeErr)
 	}
 	defer func() { _ = ln.Close() }()
 	log.Info("daemon starting", "socket", ln.Addr())

M internal/app/run_test.go +7 -2


@@ -240,14 +240,19 @@ func TestCatalogueRegistryConstructionPrecedesSocketPublication(t *testing.T) {
240	240		events = append(events, "catalogue-registry")
241	241		return nil
242	242		},
243		-	func() (ports.Listener, error) {
	243	+	func(*daemon.Daemon) error {
244	244		require.Equal(t, []string{"catalogue-registry"}, events)
	245	+	events = append(events, "startup-garbage-collection")
	246	+	return nil
	247	+	},
	248	+	func() (ports.Listener, error) {
	249	+	require.Equal(t, []string{"catalogue-registry", "startup-garbage-collection"}, events)
245	250		events = append(events, "socket-publication")
246	251		return nil, nil
247	252		},
248	253		)
249	254		require.NoError(t, err)
250		-	require.Equal(t, []string{"catalogue-registry", "socket-publication"}, events)
	255	+	require.Equal(t, []string{"catalogue-registry", "startup-garbage-collection", "socket-publication"}, events)
251	256		}
252	257
253	258		func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {

 			events = append(events, "catalogue-registry")
 			return nil
 		},
-		func() (ports.Listener, error) {
+		func(*daemon.Daemon) error {
 			require.Equal(t, []string{"catalogue-registry"}, events)
+			events = append(events, "startup-garbage-collection")
+			return nil
+		},
+		func() (ports.Listener, error) {
+			require.Equal(t, []string{"catalogue-registry", "startup-garbage-collection"}, events)
 			events = append(events, "socket-publication")
 			return nil, nil
 		},
 	)
 	require.NoError(t, err)
-	require.Equal(t, []string{"catalogue-registry", "socket-publication"}, events)
+	require.Equal(t, []string{"catalogue-registry", "startup-garbage-collection", "socket-publication"}, events)
 }
 func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {

M internal/usecase/daemon/daemon.go +2 -6


@@ -302,8 +302,8 @@ func WithNoticeStore(store ports.NoticeStore) Option {
302	302		return func(d *Daemon) { d.noticeStore = store }
303	303		}
304	304
305		-	// WithDurableMaintenance schedules one garbage-collection pass after
306		-	// catalogue-driven restoration completes.
	305	+	// WithDurableMaintenance configures the one pre-publication garbage-collection
	306	+	// pass. The application invokes it before opening the daemon socket.
307	307		func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {
308	308		return func(d *Daemon) {
309	309		d.maintenance = newMaintenanceDependencies(catalogue, repository)

@@ -524,10 +524,6 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
524	524		} else {
525	525		d.closeRestoreDone()
526	526		}
527		-	if d.maintenance.catalogue != nil && d.maintenance.repository != nil {
528		-	d.startDurableMaintenance()
529		-	}
530		-
531	527		// Break the accept loop when either the parent context is cancelled or the
532	528		// registry drains to empty: both close the listener, which fails Accept.
533	529		go func() {

 	return func(d *Daemon) { d.noticeStore = store }
 }
-// WithDurableMaintenance schedules one garbage-collection pass after
+// WithDurableMaintenance configures the one pre-publication garbage-collection
-// catalogue-driven restoration completes.
+// pass. The application invokes it before opening the daemon socket.
 func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {
 	return func(d *Daemon) {
 		d.maintenance = newMaintenanceDependencies(catalogue, repository)
 	} else {
 		d.closeRestoreDone()
 	}
-	if d.maintenance.catalogue != nil && d.maintenance.repository != nil {
-		d.startDurableMaintenance()
-	}
 	// Break the accept loop when either the parent context is cancelled or the
 	// registry drains to empty: both close the listener, which fails Accept.
 	go func() {

M internal/usecase/daemon/maintenance.go +23 -36


@@ -2,6 +2,7 @@ package daemon
2	2
3	3		import (
4	4		"context"
	5	+	"fmt"
5	6
6	7		"github.com/bnema/vev/internal/domain"
7	8		"github.com/bnema/vev/internal/ports"

@@ -16,57 +17,43 @@ func newMaintenanceDependencies(catalogue ports.Catalogue, repository ports.Snap
16	17		return maintenanceDependencies{catalogue: catalogue, repository: repository}
17	18		}
18	19
19		-	func (d *Daemon) startDurableMaintenance() {
	20	+	// CollectStartupGarbage runs the one GC pass before the daemon socket is
	21	+	// published. In production the recovery coordinator fences the catalogue
	22	+	// snapshot and collection against every durable mutation and checkpoint
	23	+	// publication. The fallback keeps standalone daemon tests usable when no
	24	+	// coordinator is installed.
	25	+	func (d *Daemon) CollectStartupGarbage(ctx context.Context) error {
20	26		if d == nil {
21		-	return
	27	+	return nil
22	28		}
23		-	d.snapshotWorkerMu.Lock()
24		-	if d.maintenanceWorkerCancel != nil {
25		-	d.snapshotWorkerMu.Unlock()
26		-	return
	29	+	if d.recovery != nil {
	30	+	incarnations, err := d.recovery.CollectGarbage(ctx)
	31	+	if err != nil {
	32	+	return fmt.Errorf("collect startup snapshots: %w", err)
	33	+	}
	34	+	d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
	35	+	return nil
27	36		}
28		-	ctx, cancel := context.WithCancel(d.serveCtx)
29		-	d.maintenanceWorkerCancel = cancel
30		-	d.maintenanceWorkerDone = make(chan struct{})
31		-	done := d.maintenanceWorkerDone
32		-	d.snapshotWorkerMu.Unlock()
33		-
34		-	go func() {
35		-	defer close(done)
36		-	d.runStartupGarbageCollection(ctx)
37		-	}()
38		-	}
39	37
40		-	// runStartupGarbageCollection waits for catalogue-driven restoration, then
41		-	// takes a fresh catalogue snapshot and applies GC exactly once. A catalogue
42		-	// read failure never produces a keep set, so destructive collection cannot run
43		-	// without known-good catalogue state.
44		-	func (d *Daemon) runStartupGarbageCollection(ctx context.Context) {
45		-	select {
46		-	case <-ctx.Done():
47		-	return
48		-	case <-d.restoreDone:
49		-	}
50		-	deps := &d.maintenance
	38	+	deps := d.maintenance
51	39		if deps.catalogue == nil \|\| deps.repository == nil {
52		-	return
	40	+	return nil
53	41		}
54	42		records, err := deps.catalogue.Records()
55	43		if err != nil {
56		-	d.log.Warn("snapshot_garbage_collection_skipped", "reason", "catalogue-read-failed", "err", err)
57		-	return
	44	+	return fmt.Errorf("read startup catalogue: %w", err)
58	45		}
59	46		keep := make(map[domain.IncarnationID]domain.CheckpointRef, len(records))
60	47		for _, record := range records {
61		-	var committed domain.CheckpointRef
62	48		if record.Committed != nil {
63		-	committed = *record.Committed
	49	+	keep[record.IncarnationID] = *record.Committed
	50	+	continue
64	51		}
65		-	keep[record.IncarnationID] = committed
	52	+	keep[record.IncarnationID] = domain.CheckpointRef{}
66	53		}
67	54		if err := deps.repository.CollectGarbage(ctx, keep); err != nil {
68		-	d.log.Warn("snapshot_garbage_collection_failed", "incarnations", len(keep), "err", err)
69		-	return
	55	+	return fmt.Errorf("collect startup snapshots: %w", err)
70	56		}
71	57		d.log.Info("snapshot_garbage_collection_complete", "incarnations", len(keep))
	58	+	return nil
72	59		}

 import (
 	"context"
+	"fmt"
 	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
 	return maintenanceDependencies{catalogue: catalogue, repository: repository}
 }
-func (d *Daemon) startDurableMaintenance() {
+// CollectStartupGarbage runs the one GC pass before the daemon socket is
+// published. In production the recovery coordinator fences the catalogue
+// snapshot and collection against every durable mutation and checkpoint
+// publication. The fallback keeps standalone daemon tests usable when no
+// coordinator is installed.
+func (d *Daemon) CollectStartupGarbage(ctx context.Context) error {
 	if d == nil {
-		return
+		return nil
 	}
-	d.snapshotWorkerMu.Lock()
+	if d.recovery != nil {
-	if d.maintenanceWorkerCancel != nil {
+		incarnations, err := d.recovery.CollectGarbage(ctx)
-		d.snapshotWorkerMu.Unlock()
+		if err != nil {
-		return
+			return fmt.Errorf("collect startup snapshots: %w", err)
+		}
+		d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
+		return nil
 	}
-	ctx, cancel := context.WithCancel(d.serveCtx)
-	d.maintenanceWorkerCancel = cancel
-	d.maintenanceWorkerDone = make(chan struct{})
-	done := d.maintenanceWorkerDone
-	d.snapshotWorkerMu.Unlock()
-	go func() {
-		defer close(done)
-		d.runStartupGarbageCollection(ctx)
-	}()
-}
-// runStartupGarbageCollection waits for catalogue-driven restoration, then
+	deps := d.maintenance
-// takes a fresh catalogue snapshot and applies GC exactly once. A catalogue
-// read failure never produces a keep set, so destructive collection cannot run
-// without known-good catalogue state.
-func (d *Daemon) runStartupGarbageCollection(ctx context.Context) {
-	select {
-	case <-ctx.Done():
-		return
-	case <-d.restoreDone:
-	}
-	deps := &d.maintenance
 	if deps.catalogue == nil || deps.repository == nil {
-		return
+		return nil
 	}
 	records, err := deps.catalogue.Records()
 	if err != nil {
-		d.log.Warn("snapshot_garbage_collection_skipped", "reason", "catalogue-read-failed", "err", err)
+		return fmt.Errorf("read startup catalogue: %w", err)
-		return
 	}
 	keep := make(map[domain.IncarnationID]domain.CheckpointRef, len(records))
 	for _, record := range records {
-		var committed domain.CheckpointRef
 		if record.Committed != nil {
-			committed = *record.Committed
+			keep[record.IncarnationID] = *record.Committed
+			continue
 		}
-		keep[record.IncarnationID] = committed
+		keep[record.IncarnationID] = domain.CheckpointRef{}
 	}
 	if err := deps.repository.CollectGarbage(ctx, keep); err != nil {
-		d.log.Warn("snapshot_garbage_collection_failed", "incarnations", len(keep), "err", err)
+		return fmt.Errorf("collect startup snapshots: %w", err)
-		return
 	}
 	d.log.Info("snapshot_garbage_collection_complete", "incarnations", len(keep))
+	return nil
 }

M internal/usecase/daemon/maintenance_test.go +2 -4


@@ -25,9 +25,8 @@ func TestStartupGarbageCollectionUsesValidatedCatalogue(t *testing.T) {
25	25		})).Return(nil).Once()
26	26		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
27	27		WithDurableMaintenance(catalogue, repository)(d)
28		-	d.closeRestoreDone()
29	28
30		-	d.runStartupGarbageCollection(context.Background())
	29	+	require.NoError(t, d.CollectStartupGarbage(context.Background()))
31	30		}
32	31
33	32		func TestStartupGarbageCollectionSkipsUnknownCatalogueState(t *testing.T) {

@@ -36,9 +35,8 @@ func TestStartupGarbageCollectionSkipsUnknownCatalogueState(t *testing.T) {
36	35		repository := portsmocks.NewMockSnapshotRepository(t)
37	36		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
38	37		WithDurableMaintenance(catalogue, repository)(d)
39		-	d.closeRestoreDone()
40	38
41		-	d.runStartupGarbageCollection(context.Background())
	39	+	require.Error(t, d.CollectStartupGarbage(context.Background()))
42	40		repository.AssertNotCalled(t, "CollectGarbage", mock.Anything, mock.Anything)
43	41		require.NotNil(t, d.maintenance.catalogue)
44	42		}

 	})).Return(nil).Once()
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	WithDurableMaintenance(catalogue, repository)(d)
-	d.closeRestoreDone()
-	d.runStartupGarbageCollection(context.Background())
+	require.NoError(t, d.CollectStartupGarbage(context.Background()))
 }
 func TestStartupGarbageCollectionSkipsUnknownCatalogueState(t *testing.T) {
 	repository := portsmocks.NewMockSnapshotRepository(t)
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	WithDurableMaintenance(catalogue, repository)(d)
-	d.closeRestoreDone()
-	d.runStartupGarbageCollection(context.Background())
+	require.Error(t, d.CollectStartupGarbage(context.Background()))
 	repository.AssertNotCalled(t, "CollectGarbage", mock.Anything, mock.Anything)
 	require.NotNil(t, d.maintenance.catalogue)
 }

M internal/usecase/recovery/checkpoint.go +5 -0


@@ -22,6 +22,11 @@ func (c *Coordinator) PublishCheckpoint(ctx context.Context, name string, public
22	22		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {
23	23		return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
24	24		}
	25	+	if err := ctx.Err(); err != nil {
	26	+	return domain.CatalogueRecord{}, err
	27	+	}
	28	+	c.mutationMu.Lock()
	29	+	defer c.mutationMu.Unlock()
25	30		unlock := c.locks.Lock([]string{name})
26	31		defer unlock()
27	32


@@ -22,6 +22,11 @@ func (c *Coordinator) PublishCheckpoint(ctx context.Context, name string, public
22	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {	22		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {
23	return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")	23		return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
24	}	24		}
		25	+	if err := ctx.Err(); err != nil {
		26	+	return domain.CatalogueRecord{}, err
		27	+	}
		28	+	c.mutationMu.Lock()
		29	+	defer c.mutationMu.Unlock()
25	unlock := c.locks.Lock([]string{name})	30		unlock := c.locks.Lock([]string{name})
26	defer unlock()	31		defer unlock()
27		32

M internal/usecase/recovery/coordinator.go +43 -0


@@ -85,6 +85,11 @@ type Coordinator struct {
85	85		catalogue ports.Catalogue
86	86		repository ports.SnapshotRepository
87	87		locks *KeyLocks
	88	+	// mutationMu serializes the startup GC snapshot and collection with every
	89	+	// catalogue or checkpoint mutation. Startup normally provides this fence by
	90	+	// running before socket publication; retaining it here also makes the
	91	+	// coordinator safe when used directly by a composition root.
	92	+	mutationMu sync.Mutex
88	93		random io.Reader
89	94		}
90	95

@@ -101,6 +106,8 @@ func (c *Coordinator) Create(ctx context.Context, record domain.CatalogueRecord)
101	106		if err := ctx.Err(); err != nil {
102	107		return domain.CatalogueRecord{}, err
103	108		}
	109	+	c.mutationMu.Lock()
	110	+	defer c.mutationMu.Unlock()
104	111		unlock := c.locks.Lock([]string{record.Name})
105	112		defer unlock()
106	113		if _, exists, err := c.catalogue.Record(record.Name); err != nil {

@@ -133,6 +140,8 @@ func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (doma
133	140		if err := ctx.Err(); err != nil {
134	141		return domain.CatalogueRecord{}, err
135	142		}
	143	+	c.mutationMu.Lock()
	144	+	defer c.mutationMu.Unlock()
136	145		unlock := c.locks.Lock([]string{oldName, newName})
137	146		defer unlock()
138	147		record, ok, err := c.catalogue.Record(oldName)

@@ -168,6 +177,8 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error {
168	177		if err := ctx.Err(); err != nil {
169	178		return err
170	179		}
	180	+	c.mutationMu.Lock()
	181	+	defer c.mutationMu.Unlock()
171	182		unlock := c.locks.Lock([]string{name})
172	183		defer unlock()
173	184		record, ok, err := c.catalogue.Record(name)

@@ -182,3 +193,35 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error {
182	193		}
183	194		return c.repository.DeleteIncarnation(ctx, record.IncarnationID)
184	195		}
	196	+
	197	+	// CollectGarbage takes a catalogue snapshot and applies retention while all
	198	+	// catalogue mutations and checkpoint publications are fenced. Its keep map is
	199	+	// therefore never stale relative to an operation that creates an incarnation
	200	+	// or commits a new generation.
	201	+	func (c *Coordinator) CollectGarbage(ctx context.Context) (int, error) {
	202	+	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil {
	203	+	return 0, errors.New("recovery: incomplete garbage collection dependencies")
	204	+	}
	205	+	if err := ctx.Err(); err != nil {
	206	+	return 0, err
	207	+	}
	208	+	c.mutationMu.Lock()
	209	+	defer c.mutationMu.Unlock()
	210	+
	211	+	records, err := c.catalogue.Records()
	212	+	if err != nil {
	213	+	return 0, err
	214	+	}
	215	+	keep := make(map[domain.IncarnationID]domain.CheckpointRef, len(records))
	216	+	for _, record := range records {
	217	+	if record.Committed != nil {
	218	+	keep[record.IncarnationID] = *record.Committed
	219	+	continue
	220	+	}
	221	+	keep[record.IncarnationID] = domain.CheckpointRef{}
	222	+	}
	223	+	if err := c.repository.CollectGarbage(ctx, keep); err != nil {
	224	+	return len(keep), err
	225	+	}
	226	+	return len(keep), nil
	227	+	}


@@ -85,6 +85,11 @@ type Coordinator struct {
85	catalogue ports.Catalogue	85		catalogue ports.Catalogue
86	repository ports.SnapshotRepository	86		repository ports.SnapshotRepository
87	locks *KeyLocks	87		locks *KeyLocks
		88	+	// mutationMu serializes the startup GC snapshot and collection with every
		89	+	// catalogue or checkpoint mutation. Startup normally provides this fence by
		90	+	// running before socket publication; retaining it here also makes the
		91	+	// coordinator safe when used directly by a composition root.
		92	+	mutationMu sync.Mutex
88	random io.Reader	93		random io.Reader
89	}	94		}
90		95

@@ -101,6 +106,8 @@ func (c *Coordinator) Create(ctx context.Context, record domain.CatalogueRecord)
101	if err := ctx.Err(); err != nil {	106		if err := ctx.Err(); err != nil {
102	return domain.CatalogueRecord{}, err	107		return domain.CatalogueRecord{}, err
103	}	108		}
		109	+	c.mutationMu.Lock()
		110	+	defer c.mutationMu.Unlock()
104	unlock := c.locks.Lock([]string{record.Name})	111		unlock := c.locks.Lock([]string{record.Name})
105	defer unlock()	112		defer unlock()
106	if _, exists, err := c.catalogue.Record(record.Name); err != nil {	113		if _, exists, err := c.catalogue.Record(record.Name); err != nil {

@@ -133,6 +140,8 @@ func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (doma
133	if err := ctx.Err(); err != nil {	140		if err := ctx.Err(); err != nil {
134	return domain.CatalogueRecord{}, err	141		return domain.CatalogueRecord{}, err
135	}	142		}
		143	+	c.mutationMu.Lock()
		144	+	defer c.mutationMu.Unlock()
136	unlock := c.locks.Lock([]string{oldName, newName})	145		unlock := c.locks.Lock([]string{oldName, newName})
137	defer unlock()	146		defer unlock()
138	record, ok, err := c.catalogue.Record(oldName)	147		record, ok, err := c.catalogue.Record(oldName)

@@ -168,6 +177,8 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error {
168	if err := ctx.Err(); err != nil {	177		if err := ctx.Err(); err != nil {
169	return err	178		return err
170	}	179		}
		180	+	c.mutationMu.Lock()
		181	+	defer c.mutationMu.Unlock()
171	unlock := c.locks.Lock([]string{name})	182		unlock := c.locks.Lock([]string{name})
172	defer unlock()	183		defer unlock()
173	record, ok, err := c.catalogue.Record(name)	184		record, ok, err := c.catalogue.Record(name)

@@ -182,3 +193,35 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error {
182	}	193		}
183	return c.repository.DeleteIncarnation(ctx, record.IncarnationID)	194		return c.repository.DeleteIncarnation(ctx, record.IncarnationID)
184	}	195		}
		196	+
		197	+	// CollectGarbage takes a catalogue snapshot and applies retention while all
		198	+	// catalogue mutations and checkpoint publications are fenced. Its keep map is
		199	+	// therefore never stale relative to an operation that creates an incarnation
		200	+	// or commits a new generation.
		201	+	func (c *Coordinator) CollectGarbage(ctx context.Context) (int, error) {
		202	+	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil {
		203	+	return 0, errors.New("recovery: incomplete garbage collection dependencies")
		204	+	}
		205	+	if err := ctx.Err(); err != nil {
		206	+	return 0, err
		207	+	}
		208	+	c.mutationMu.Lock()
		209	+	defer c.mutationMu.Unlock()
		210	+
		211	+	records, err := c.catalogue.Records()
		212	+	if err != nil {
		213	+	return 0, err
		214	+	}
		215	+	keep := make(map[domain.IncarnationID]domain.CheckpointRef, len(records))
		216	+	for _, record := range records {
		217	+	if record.Committed != nil {
		218	+	keep[record.IncarnationID] = *record.Committed
		219	+	continue
		220	+	}
		221	+	keep[record.IncarnationID] = domain.CheckpointRef{}
		222	+	}
		223	+	if err := c.repository.CollectGarbage(ctx, keep); err != nil {
		224	+	return len(keep), err
		225	+	}
		226	+	return len(keep), nil
		227	+	}

M internal/usecase/recovery/discard.go +2 -0


@@ -24,6 +24,8 @@ func (c *Coordinator) Discard(ctx context.Context, name string) error {
24	24		if err := ctx.Err(); err != nil {
25	25		return err
26	26		}
	27	+	c.mutationMu.Lock()
	28	+	defer c.mutationMu.Unlock()
27	29		unlock := c.locks.Lock([]string{name})
28	30		defer unlock()
29	31


@@ -24,6 +24,8 @@ func (c *Coordinator) Discard(ctx context.Context, name string) error {
24	if err := ctx.Err(); err != nil {	24		if err := ctx.Err(); err != nil {
25	return err	25		return err
26	}	26		}
		27	+	c.mutationMu.Lock()
		28	+	defer c.mutationMu.Unlock()
27	unlock := c.locks.Lock([]string{name})	29		unlock := c.locks.Lock([]string{name})
28	defer unlock()	30		defer unlock()
29		31

M internal/usecase/recovery/transactions_test.go +170 -0


@@ -17,6 +17,7 @@ type transactionCatalogue struct {
17	17		records map[string]domain.CatalogueRecord
18	18		events []string
19	19		replaceErrOnce error
	20	+	onCreate func(domain.CatalogueRecord)
20	21		}
21	22
22	23		func newTransactionCatalogue(records ...domain.CatalogueRecord) *transactionCatalogue {

@@ -52,6 +53,9 @@ func (c *transactionCatalogue) Create(record domain.CatalogueRecord) error {
52	53		}
53	54		c.records[record.Name] = record
54	55		c.events = append(c.events, "create")
	56	+	if c.onCreate != nil {
	57	+	c.onCreate(record)
	58	+	}
55	59		return nil
56	60		}
57	61

@@ -105,6 +109,91 @@ type transactionRepository struct {
105	109		deletedIDs []domain.IncarnationID
106	110		}
107	111
	112	+	type garbageCollectionFenceRepository struct {
	113	+	ports.SnapshotRepository
	114	+	mu sync.Mutex
	115	+	events []string
	116	+	collectionEntered chan struct{}
	117	+	releaseCollection chan struct{}
	118	+	collectionFinished bool
	119	+	publishBeforeGC bool
	120	+	incarnations map[domain.IncarnationID]struct{}
	121	+	generations map[domain.IncarnationID]map[uint64]struct{}
	122	+	}
	123	+
	124	+	func (r *garbageCollectionFenceRepository) CollectGarbage(_ context.Context, keep map[domain.IncarnationID]domain.CheckpointRef) error {
	125	+	close(r.collectionEntered)
	126	+	<-r.releaseCollection
	127	+	r.mu.Lock()
	128	+	defer r.mu.Unlock()
	129	+	for id := range r.incarnations {
	130	+	if _, known := keep[id]; !known {
	131	+	delete(r.incarnations, id)
	132	+	}
	133	+	}
	134	+	for id, generations := range r.generations {
	135	+	committed, known := keep[id]
	136	+	if !known {
	137	+	delete(r.generations, id)
	138	+	continue
	139	+	}
	140	+	for generation := range generations {
	141	+	if generation != committed.Generation && generation+1 != committed.Generation {
	142	+	delete(generations, generation)
	143	+	}
	144	+	}
	145	+	}
	146	+	r.collectionFinished = true
	147	+	r.events = append(r.events, "collect")
	148	+	return nil
	149	+	}
	150	+
	151	+	func (r *garbageCollectionFenceRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error {
	152	+	r.mu.Lock()
	153	+	defer r.mu.Unlock()
	154	+	if !r.collectionFinished {
	155	+	r.publishBeforeGC = true
	156	+	}
	157	+	if r.generations == nil {
	158	+	r.generations = make(map[domain.IncarnationID]map[uint64]struct{})
	159	+	}
	160	+	if r.generations[publication.IncarnationID] == nil {
	161	+	r.generations[publication.IncarnationID] = make(map[uint64]struct{})
	162	+	}
	163	+	r.generations[publication.IncarnationID][publication.Generation] = struct{}{}
	164	+	r.events = append(r.events, "publish")
	165	+	return nil
	166	+	}
	167	+
	168	+	func (r *garbageCollectionFenceRepository) addIncarnation(record domain.CatalogueRecord) {
	169	+	r.mu.Lock()
	170	+	defer r.mu.Unlock()
	171	+	if r.incarnations == nil {
	172	+	r.incarnations = make(map[domain.IncarnationID]struct{})
	173	+	}
	174	+	r.incarnations[record.IncarnationID] = struct{}{}
	175	+	}
	176	+
	177	+	func (r *garbageCollectionFenceRepository) hasIncarnation(id domain.IncarnationID) bool {
	178	+	r.mu.Lock()
	179	+	defer r.mu.Unlock()
	180	+	_, ok := r.incarnations[id]
	181	+	return ok
	182	+	}
	183	+
	184	+	func (r *garbageCollectionFenceRepository) hasGeneration(id domain.IncarnationID, generation uint64) bool {
	185	+	r.mu.Lock()
	186	+	defer r.mu.Unlock()
	187	+	_, ok := r.generations[id][generation]
	188	+	return ok
	189	+	}
	190	+
	191	+	func (r *garbageCollectionFenceRepository) eventLog() ([]string, bool) {
	192	+	r.mu.Lock()
	193	+	defer r.mu.Unlock()
	194	+	return append([]string(nil), r.events...), r.publishBeforeGC
	195	+	}
	196	+
108	197		func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
109	198		r.mu.Lock()
110	199		defer r.mu.Unlock()

@@ -124,6 +213,87 @@ func degradedTransactionRecord() domain.CatalogueRecord {
124	213		}
125	214		}
126	215
	216	+	func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *testing.T) {
	217	+	t.Parallel()
	218	+
	219	+	t.Run("a stale keep snapshot cannot remove a newly created incarnation", func(t *testing.T) {
	220	+	t.Parallel()
	221	+	catalogue := newTransactionCatalogue()
	222	+	repository := &garbageCollectionFenceRepository{
	223	+	collectionEntered: make(chan struct{}),
	224	+	releaseCollection: make(chan struct{}),
	225	+	}
	226	+	catalogue.onCreate = repository.addIncarnation
	227	+	coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
	228	+
	229	+	collected := make(chan error, 1)
	230	+	go func() {
	231	+	_, err := coordinator.CollectGarbage(context.Background())
	232	+	collected <- err
	233	+	}()
	234	+	<-repository.collectionEntered
	235	+
	236	+	created := make(chan error, 1)
	237	+	go func() {
	238	+	_, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new"})
	239	+	created <- err
	240	+	}()
	241	+	close(repository.releaseCollection)
	242	+
	243	+	require.NoError(t, <-collected)
	244	+	require.NoError(t, <-created)
	245	+	got, ok, err := catalogue.Record("new")
	246	+	require.NoError(t, err)
	247	+	require.True(t, ok)
	248	+	require.NotZero(t, got.IncarnationID)
	249	+	require.True(t, repository.hasIncarnation(got.IncarnationID), "GC's stale keep snapshot must finish before the new incarnation is published")
	250	+	events, _ := repository.eventLog()
	251	+	require.Equal(t, []string{"collect"}, events)
	252	+	})
	253	+
	254	+	t.Run("a stale keep snapshot cannot prune a newly committed generation", func(t *testing.T) {
	255	+	t.Parallel()
	256	+	committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
	257	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
	258	+	catalogue := newTransactionCatalogue(record)
	259	+	repository := &garbageCollectionFenceRepository{
	260	+	collectionEntered: make(chan struct{}),
	261	+	releaseCollection: make(chan struct{}),
	262	+	generations: map[domain.IncarnationID]map[uint64]struct{}{
	263	+	record.IncarnationID: {1: {}},
	264	+	},
	265	+	}
	266	+	coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
	267	+
	268	+	collected := make(chan error, 1)
	269	+	go func() {
	270	+	_, err := coordinator.CollectGarbage(context.Background())
	271	+	collected <- err
	272	+	}()
	273	+	<-repository.collectionEntered
	274	+
	275	+	published := make(chan error, 1)
	276	+	go func() {
	277	+	_, err := coordinator.PublishCheckpoint(context.Background(), "work", ports.SnapshotPublication{
	278	+	Name: "work",
	279	+	IncarnationID: record.IncarnationID,
	280	+	Generation: 2,
	281	+	ParentCheckpoint: &committed,
	282	+	Manifest: []byte("generation two"),
	283	+	})
	284	+	published <- err
	285	+	}()
	286	+	close(repository.releaseCollection)
	287	+
	288	+	require.NoError(t, <-collected)
	289	+	require.NoError(t, <-published)
	290	+	events, publishedBeforeGC := repository.eventLog()
	291	+	require.Equal(t, []string{"collect", "publish"}, events)
	292	+	require.False(t, publishedBeforeGC)
	293	+	require.True(t, repository.hasGeneration(record.IncarnationID, 2), "GC's stale keep snapshot must finish before generation two is published")
	294	+	})
	295	+	}
	296	+
127	297		func TestDiscardIsRetryIdempotent(t *testing.T) {
128	298		t.Parallel()
129	299		failure := errors.New("injected discard failure")


@@ -17,6 +17,7 @@ type transactionCatalogue struct {
17	records map[string]domain.CatalogueRecord	17		records map[string]domain.CatalogueRecord
18	events []string	18		events []string
19	replaceErrOnce error	19		replaceErrOnce error
		20	+	onCreate func(domain.CatalogueRecord)
20	}	21		}
21		22
22	func newTransactionCatalogue(records ...domain.CatalogueRecord) *transactionCatalogue {	23		func newTransactionCatalogue(records ...domain.CatalogueRecord) *transactionCatalogue {

@@ -52,6 +53,9 @@ func (c *transactionCatalogue) Create(record domain.CatalogueRecord) error {
52	}	53		}
53	c.records[record.Name] = record	54		c.records[record.Name] = record
54	c.events = append(c.events, "create")	55		c.events = append(c.events, "create")
		56	+	if c.onCreate != nil {
		57	+	c.onCreate(record)
		58	+	}
55	return nil	59		return nil
56	}	60		}
57		61

@@ -105,6 +109,91 @@ type transactionRepository struct {
105	deletedIDs []domain.IncarnationID	109		deletedIDs []domain.IncarnationID
106	}	110		}
107		111
		112	+	type garbageCollectionFenceRepository struct {
		113	+	ports.SnapshotRepository
		114	+	mu sync.Mutex
		115	+	events []string
		116	+	collectionEntered chan struct{}
		117	+	releaseCollection chan struct{}
		118	+	collectionFinished bool
		119	+	publishBeforeGC bool
		120	+	incarnations map[domain.IncarnationID]struct{}
		121	+	generations map[domain.IncarnationID]map[uint64]struct{}
		122	+	}
		123	+
		124	+	func (r *garbageCollectionFenceRepository) CollectGarbage(_ context.Context, keep map[domain.IncarnationID]domain.CheckpointRef) error {
		125	+	close(r.collectionEntered)
		126	+	<-r.releaseCollection
		127	+	r.mu.Lock()
		128	+	defer r.mu.Unlock()
		129	+	for id := range r.incarnations {
		130	+	if _, known := keep[id]; !known {
		131	+	delete(r.incarnations, id)
		132	+	}
		133	+	}
		134	+	for id, generations := range r.generations {
		135	+	committed, known := keep[id]
		136	+	if !known {
		137	+	delete(r.generations, id)
		138	+	continue
		139	+	}
		140	+	for generation := range generations {
		141	+	if generation != committed.Generation && generation+1 != committed.Generation {
		142	+	delete(generations, generation)
		143	+	}
		144	+	}
		145	+	}
		146	+	r.collectionFinished = true
		147	+	r.events = append(r.events, "collect")
		148	+	return nil
		149	+	}
		150	+
		151	+	func (r *garbageCollectionFenceRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error {
		152	+	r.mu.Lock()
		153	+	defer r.mu.Unlock()
		154	+	if !r.collectionFinished {
		155	+	r.publishBeforeGC = true
		156	+	}
		157	+	if r.generations == nil {
		158	+	r.generations = make(map[domain.IncarnationID]map[uint64]struct{})
		159	+	}
		160	+	if r.generations[publication.IncarnationID] == nil {
		161	+	r.generations[publication.IncarnationID] = make(map[uint64]struct{})
		162	+	}
		163	+	r.generations[publication.IncarnationID][publication.Generation] = struct{}{}
		164	+	r.events = append(r.events, "publish")
		165	+	return nil
		166	+	}
		167	+
		168	+	func (r *garbageCollectionFenceRepository) addIncarnation(record domain.CatalogueRecord) {
		169	+	r.mu.Lock()
		170	+	defer r.mu.Unlock()
		171	+	if r.incarnations == nil {
		172	+	r.incarnations = make(map[domain.IncarnationID]struct{})
		173	+	}
		174	+	r.incarnations[record.IncarnationID] = struct{}{}
		175	+	}
		176	+
		177	+	func (r *garbageCollectionFenceRepository) hasIncarnation(id domain.IncarnationID) bool {
		178	+	r.mu.Lock()
		179	+	defer r.mu.Unlock()
		180	+	_, ok := r.incarnations[id]
		181	+	return ok
		182	+	}
		183	+
		184	+	func (r *garbageCollectionFenceRepository) hasGeneration(id domain.IncarnationID, generation uint64) bool {
		185	+	r.mu.Lock()
		186	+	defer r.mu.Unlock()
		187	+	_, ok := r.generations[id][generation]
		188	+	return ok
		189	+	}
		190	+
		191	+	func (r *garbageCollectionFenceRepository) eventLog() ([]string, bool) {
		192	+	r.mu.Lock()
		193	+	defer r.mu.Unlock()
		194	+	return append([]string(nil), r.events...), r.publishBeforeGC
		195	+	}
		196	+
108	func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {	197		func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
109	r.mu.Lock()	198		r.mu.Lock()
110	defer r.mu.Unlock()	199		defer r.mu.Unlock()

@@ -124,6 +213,87 @@ func degradedTransactionRecord() domain.CatalogueRecord {
124	}	213		}
125	}	214		}
126		215
		216	+	func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *testing.T) {
		217	+	t.Parallel()
		218	+
		219	+	t.Run("a stale keep snapshot cannot remove a newly created incarnation", func(t *testing.T) {
		220	+	t.Parallel()
		221	+	catalogue := newTransactionCatalogue()
		222	+	repository := &garbageCollectionFenceRepository{
		223	+	collectionEntered: make(chan struct{}),
		224	+	releaseCollection: make(chan struct{}),
		225	+	}
		226	+	catalogue.onCreate = repository.addIncarnation
		227	+	coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
		228	+
		229	+	collected := make(chan error, 1)
		230	+	go func() {
		231	+	_, err := coordinator.CollectGarbage(context.Background())
		232	+	collected <- err
		233	+	}()
		234	+	<-repository.collectionEntered
		235	+
		236	+	created := make(chan error, 1)
		237	+	go func() {
		238	+	_, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new"})
		239	+	created <- err
		240	+	}()
		241	+	close(repository.releaseCollection)
		242	+
		243	+	require.NoError(t, <-collected)
		244	+	require.NoError(t, <-created)
		245	+	got, ok, err := catalogue.Record("new")
		246	+	require.NoError(t, err)
		247	+	require.True(t, ok)
		248	+	require.NotZero(t, got.IncarnationID)
		249	+	require.True(t, repository.hasIncarnation(got.IncarnationID), "GC's stale keep snapshot must finish before the new incarnation is published")
		250	+	events, _ := repository.eventLog()
		251	+	require.Equal(t, []string{"collect"}, events)
		252	+	})
		253	+
		254	+	t.Run("a stale keep snapshot cannot prune a newly committed generation", func(t *testing.T) {
		255	+	t.Parallel()
		256	+	committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
		257	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
		258	+	catalogue := newTransactionCatalogue(record)
		259	+	repository := &garbageCollectionFenceRepository{
		260	+	collectionEntered: make(chan struct{}),
		261	+	releaseCollection: make(chan struct{}),
		262	+	generations: map[domain.IncarnationID]map[uint64]struct{}{
		263	+	record.IncarnationID: {1: {}},
		264	+	},
		265	+	}
		266	+	coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
		267	+
		268	+	collected := make(chan error, 1)
		269	+	go func() {
		270	+	_, err := coordinator.CollectGarbage(context.Background())
		271	+	collected <- err
		272	+	}()
		273	+	<-repository.collectionEntered
		274	+
		275	+	published := make(chan error, 1)
		276	+	go func() {
		277	+	_, err := coordinator.PublishCheckpoint(context.Background(), "work", ports.SnapshotPublication{
		278	+	Name: "work",
		279	+	IncarnationID: record.IncarnationID,
		280	+	Generation: 2,
		281	+	ParentCheckpoint: &committed,
		282	+	Manifest: []byte("generation two"),
		283	+	})
		284	+	published <- err
		285	+	}()
		286	+	close(repository.releaseCollection)
		287	+
		288	+	require.NoError(t, <-collected)
		289	+	require.NoError(t, <-published)
		290	+	events, publishedBeforeGC := repository.eventLog()
		291	+	require.Equal(t, []string{"collect", "publish"}, events)
		292	+	require.False(t, publishedBeforeGC)
		293	+	require.True(t, repository.hasGeneration(record.IncarnationID, 2), "GC's stale keep snapshot must finish before generation two is published")
		294	+	})
		295	+	}
		296	+
127	func TestDiscardIsRetryIdempotent(t *testing.T) {	297		func TestDiscardIsRetryIdempotent(t *testing.T) {
128	t.Parallel()	298		t.Parallel()
129	failure := errors.New("injected discard failure")	299		failure := errors.New("injected discard failure")