Evict archive writers on deletion and sweep idle archives (closes #89 )

The per-webhook archiveWriter registry in the database delivery target was never evicted, so a deleted webhook's writer -- and any archive file handle open within its debounce window -- lingered for the process lifetime. Separately, expiry pruning ran only when an archive was (re)opened, and reopens only happen on writes, so an archive belonging to a webhook that stopped receiving events kept its expired rows forever. Eviction: a new one-method delivery.WebhookEvictor interface (kept separate from Notifier: archiving lifecycle is not notification) is implemented by the Engine and injected into the handlers. Deleting a webhook, or deleting its last database target, drops the writer from the registry and closes its handle under the writer's own mutex, so eviction can never race an in-flight write. An evicted writer refuses further writes rather than reopening a file nothing holds. The archive file is deliberately left on disk: it is long-term storage an operator may want to keep or move away, and destroying it as a side effect of deleting a webhook would be unrecoverable. Idle sweep: a new ArchiveSweeper, modelled on the event RetentionReaper (fx lifecycle hooks, cancellable context, WaitGroup, ticker loop), prunes archives whose database target declares a positive expiry. It reuses the existing RETENTION_SWEEP_INTERVAL rather than adding a config key. It never creates an archive -- a missing file is skipped, and the reopen uses SQLite mode=rw so the file cannot be conjured even if it disappears mid-sweep -- routes the prune through the per-webhook writer so its mutex orders the sweep against concurrent writes, and leaves the archive closed so the move-the-file-away workflow keeps working. A failure for one webhook is logged and the sweep continues. Archives with no expiry or the expiry "never" are untouched. The sweep loop's context is rooted at context.Background(), not at the fx OnStart hook context. The hook context carries fx's 15 second start timeout, so a loop derived from it is cancelled three quarters of an hour before the first tick under the default one-hour interval, giving a sweeper that never sweeps. OnStop still cancels the loop and waits on the WaitGroup, so shutdown is unchanged. The sweep also never leaves a registry entry behind. Reaching the writer through the ordinary create-and-cache accessor would let a sweep that raced a webhook deletion re-insert a writer for a webhook that no longer exists, which nothing would ever evict again -- the very leak this change closes. An entry the sweep has to create is marked sweep-owned and released when the prune finishes, unless a delivery claimed it meanwhile, in which case it belongs to the registry and an eviction can still reach it. A writer evicted underneath a sweep is an ordinary interleaving and is logged at debug, not error. Each of those guards is pinned by a test that fails when the guard is removed. The adopt-during-sweep window -- a delivery claiming the sweep's own registry entry while that sweep is still running -- is driven directly against the registry, because a delivery placed between two sweeps never reaches the release path at all. The requirement that an idle archive ends the sweep closed is asserted both on a writer proven to hold an open handle beforehand and, end to end, on a delivery-owned entry the sweep keeps, rather than on an entry the sweep has already released and which therefore reports "not open" either way. The "never" expiry short circuit is checked against an archive file that has never been migrated, so any open of it would be observable as a created table.
2026-08-09 05:22:49 +00:00
8 changed files with 37 additions and 517 deletions
--- a/TODO.md
+++ b/TODO.md
@@ -29,12 +29,6 @@ databases currently grow without bound.
 # Completed Steps
 - 2026-08-09 Root the delivery engine's worker pool and the retention
  reaper's sweep loop at `context.Background()` rather than the fx
  `OnStart` hook context (#97), which carries fx's 15s start timeout and
  killed both roughly fifteen seconds after boot: the proxy silently
  stopped delivering webhooks entirely, and the reaper never ran a
  single sweep under its default one-hour interval
 - 2026-08-09 Archive writer lifecycle (#89): deleting a webhook (or its
  last `database` target) evicts the cached archive writer and closes
  its handle while deliberately leaving `archive-{webhookID}.db` on
--- a/internal/database/export_test.go
+++ b/internal/database/export_test.go
@@ -5,8 +5,6 @@ import (
 	"log/slog"
 	"os"
 	"time"
 	"go.uber.org/fx"
 )
 // NewTestRetentionReaper builds a RetentionReaper backed by the given
@@ -31,26 +29,3 @@ func NewTestRetentionReaper(
 func (r *RetentionReaper) ExportSweep(ctx context.Context) {
 	r.sweep(ctx)
 }
 // ExportRegisterHooks registers the reaper's real fx lifecycle hooks
 // on a lifecycle supplied by a test, so a test can drive the exact
 // OnStart/OnStop functions the application runs and hand OnStart the
 // kind of context fx actually supplies.
 func (r *RetentionReaper) ExportRegisterHooks(lc fx.Lifecycle) {
 	r.registerHooks(lc)
 }
 // ExportStart starts the reaper's background loop for tests.
 func (r *RetentionReaper) ExportStart() {
 	r.start()
 }
 // ExportStop stops the reaper's background loop for tests.
 func (r *RetentionReaper) ExportStop() {
 	r.stop()
 }
 // ExportSetInterval overrides the sweep interval for tests.
 func (r *RetentionReaper) ExportSetInterval(d time.Duration) {
 	r.interval = d
 }
--- a/internal/database/retention.go
+++ b/internal/database/retention.go
@@ -56,20 +56,9 @@ func NewRetentionReaper(
 		interval:  params.Config.RetentionSweepInterval,
 	}
 	r.registerHooks(lc)
 	return r
 }
 // registerHooks wires the reaper's start and stop into the fx
 // lifecycle. The start hook's context is deliberately ignored: see
 // start for why the sweep loop must not inherit it.
 func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
 	lc.Append(fx.Hook{
-		//nolint:contextcheck // Not inheriting the hook context is
+		OnStart: func(ctx context.Context) error {
-		// the point: see start.
+			r.start(ctx)
 		OnStart: func(_ context.Context) error {
 			r.start()
 			return nil
 		},
@@ -79,20 +68,12 @@ func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
 			return nil
 		},
 	})
 	return r
 }
-// start launches the background sweep loop.
+func (r *RetentionReaper) start(ctx context.Context) {
-//
+	ctx, cancel := context.WithCancel(ctx)
 // The loop's context is derived from context.Background(), NOT from
 // the fx OnStart hook context. The hook context carries fx's start
 // timeout (15s by default) and is cancelled once the start phase
 // completes, so a loop derived from it dies 45 minutes before its
 // first tick under the default one-hour sweep interval, leaving a
 // reaper that never reaps. A long-lived goroutine must outlive the
 // startup phase, so its lifetime is bounded by OnStop instead: stop
 // cancels this context and waits on the WaitGroup.
 func (r *RetentionReaper) start() {
 	ctx, cancel := context.WithCancel(context.Background())
 	r.cancel = cancel
 	r.wg.Add(1)
--- a/internal/database/retention_lifecycle_test.go
+++ b/internal/database/retention_lifecycle_test.go
@@ -1,209 +0,0 @@
 package database_test
 import (
 	"context"
 	"testing"
 	"time"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/fx"
 	"gorm.io/gorm"
 	"sneak.berlin/go/webhooker/internal/database"
 )
 const (
 	// reaperTestInterval is the sweep interval a lifecycle test
 	// runs the reaper at, so a loop that survives startup produces
 	// an observable sweep quickly.
 	reaperTestInterval = 10 * time.Millisecond
 	// reaperStopTimeout bounds how long a lifecycle test waits for
 	// the reaper's OnStop hook to return before declaring the
 	// shutdown hung.
 	reaperStopTimeout = 10 * time.Second
 	// reaperTestRetentionDays is the retention policy the lifecycle
 	// tests give their webhook.
 	reaperTestRetentionDays = 30
 )
 // recordingLifecycle is a minimal fx.Lifecycle that records the
 // hooks a component registers, so a test can invoke the real
 // OnStart/OnStop functions with a context of its choosing.
 type recordingLifecycle struct {
 	hooks []fx.Hook
 }
 func (l *recordingLifecycle) Append(h fx.Hook) {
 	l.hooks = append(l.hooks, h)
 }
 // startReaperViaHook drives the genuine fx hooks the application
 // registers for the reaper, handing OnStart a context that is
 // already done. It returns the recorded lifecycle so the caller
 // can drive OnStop too.
 func startReaperViaHook(
 	t *testing.T, r *database.RetentionReaper,
 ) *recordingLifecycle {
 	t.Helper()
 	lc := &recordingLifecycle{}
 	r.ExportRegisterHooks(lc)
 	require.Len(t, lc.hooks, 1)
 	// fx hands OnStart a context carrying the application start
 	// timeout, and cancels it when the start phase ends. An
 	// already-cancelled context is that same defect taken to its
 	// limit, and unlike a plain context.Background() it actually
 	// distinguishes a correctly rooted loop from a broken one.
 	hookCtx, cancel := context.WithCancel(context.Background())
 	cancel()
 	require.NoError(t, lc.hooks[0].OnStart(hookCtx))
 	return lc
 }
 // eventGone reports whether an event row has been removed. It
 // takes no *testing.T because it is polled from an
 // assert.Eventually condition, which runs off the test goroutine
 // where testify assertions must not be used.
 func eventGone(db *gorm.DB, eventID string) bool {
 	var n int64
 	err := db.Unscoped().Model(&database.Event{}).
 		Where("id = ?", eventID).Count(&n).Error
 	if err != nil {
 		return false
 	}
 	return n == 0
 }
 // seedExpiredWebhook creates a webhook with a finite retention
 // policy plus one long-expired event chain, and returns the
 // webhook's database and the chain's event ID.
 func seedExpiredWebhook(
 	t *testing.T, env *retentionTestEnv,
 ) (*gorm.DB, string) {
 	t.Helper()
 	webhookID := createWebhook(
 		t, env.mainDB.DB(), reaperTestRetentionDays,
 	)
 	db, err := env.mgr.GetDB(webhookID)
 	require.NoError(t, err)
 	chain := seedEventChain(
 		t, db, webhookID,
 		time.Now().Add(-365*24*time.Hour),
 	)
 	return db, chain.eventID
 }
 // TestRetentionReaper_LoopOutlivesStartHookContext is the
 // regression test for a reaper that never reaped. fx calls
 // OnStart with a context carrying the application's start timeout
 // (15s by default) and cancels it when the start phase ends, so a
 // sweep loop rooted in it is dead three quarters of an hour
 // before its first tick under the default one-hour interval, and
 // per-webhook event databases grow without bound exactly as they
 // did before retention existed.
 //
 // Driving OnStart with an already-cancelled context is that
 // defect taken to its limit: a loop that inherits the hook
 // context never ticks once, while a correctly rooted loop keeps
 // sweeping for as long as the process lives.
 func TestRetentionReaper_LoopOutlivesStartHookContext(
 	t *testing.T,
 ) {
 	t.Parallel()
 	env := setupRetentionTest(t)
 	db, eventID := seedExpiredWebhook(t, env)
 	env.reaper.ExportSetInterval(reaperTestInterval)
 	lc := startReaperViaHook(t, env.reaper)
 	t.Cleanup(func() {
 		_ = lc.hooks[0].OnStop(context.Background())
 	})
 	assert.Eventually(
 		t,
 		func() bool { return eventGone(db, eventID) },
 		5*time.Second,
 		reaperTestInterval,
 		"the sweep loop must keep running after the start "+
 			"hook's context is done; it reaped nothing, so it "+
 			"inherited the hook context and died",
 	)
 }
 // TestRetentionReaper_StopHookStopsLoop proves the fix did not
 // trade a startup bug for a shutdown hang: now that the sweep
 // loop no longer observes the start hook's cancellation, OnStop
 // is the only thing that can stop it, and it must both return
 // promptly and actually leave the loop stopped.
 func TestRetentionReaper_StopHookStopsLoop(t *testing.T) {
 	t.Parallel()
 	env := setupRetentionTest(t)
 	db, eventID := seedExpiredWebhook(t, env)
 	env.reaper.ExportSetInterval(reaperTestInterval)
 	lc := startReaperViaHook(t, env.reaper)
 	// Let the loop prove it is running before stopping it, so a
 	// fast OnStop cannot pass by stopping something already dead.
 	require.Eventually(
 		t,
 		func() bool { return eventGone(db, eventID) },
 		5*time.Second,
 		reaperTestInterval,
 	)
 	var stopErr error
 	stopped := make(chan struct{})
 	go func() {
 		defer close(stopped)
 		// stop blocks on the loop's WaitGroup, so returning at all
 		// proves the goroutine observed the cancellation.
 		stopErr = lc.hooks[0].OnStop(context.Background())
 	}()
 	select {
 	case <-stopped:
 	case <-time.After(reaperStopTimeout):
 		t.Fatal(
 			"OnStop did not return: the retention reaper's " +
 				"WaitGroup is still waiting on a loop that never " +
 				"observed cancellation",
 		)
 	}
 	require.NoError(t, stopErr)
 	// With the loop gone, a newly expired chain must survive.
 	survivor := seedEventChain(
 		t, db, "stopped-webhook",
 		time.Now().Add(-365*24*time.Hour),
 	)
 	time.Sleep(20 * reaperTestInterval)
 	assert.False(
 		t,
 		eventGone(db, survivor.eventID),
 		"a stopped reaper must not sweep anything",
 	)
 }
--- a/internal/delivery/engine.go
+++ b/internal/delivery/engine.go
@@ -170,7 +170,18 @@ func New(
 		Transport: NewSSRFSafeTransport(),
 	})
-	e.registerHooks(lc)
+	lc.Append(fx.Hook{
 		OnStart: func(ctx context.Context) error {
 			e.start(ctx)
 			return nil
 		},
 		OnStop: func(_ context.Context) error {
 			e.stop()
 			return nil
 		},
 	})
 	return e
 }
@@ -233,40 +244,8 @@ func (e *Engine) ScheduleRetry(
 	})
 }
-// registerHooks wires the engine's start and stop into the fx
+func (e *Engine) start(ctx context.Context) {
-// lifecycle. The start hook's context is deliberately ignored:
+	ctx, cancel := context.WithCancel(ctx)
 // see start for why the worker pool must not inherit it.
 func (e *Engine) registerHooks(lc fx.Lifecycle) {
 	lc.Append(fx.Hook{
 		//nolint:contextcheck // Not inheriting the hook context
 		// is the point: see start.
 		OnStart: func(_ context.Context) error {
 			e.start()
 			return nil
 		},
 		OnStop: func(_ context.Context) error {
 			e.stop()
 			return nil
 		},
 	})
 }
 // start launches the worker pool, restart recovery, and the
 // periodic retry sweep.
 //
 // Their context is derived from context.Background(), NOT from
 // the fx OnStart hook context. The hook context carries fx's
 // start timeout (15s by default) and is cancelled once the start
 // phase completes, so goroutines derived from it stop a few
 // seconds into the process: every worker would return and the
 // engine would silently stop delivering webhooks entirely. A
 // long-lived goroutine must outlive the startup phase, so its
 // lifetime is bounded by OnStop instead: stop cancels this
 // context and waits on the WaitGroup.
 func (e *Engine) start() {
 	ctx, cancel := context.WithCancel(context.Background())
 	e.cancel = cancel
 	for range e.workers {
--- a/internal/delivery/engine_integration_test.go
+++ b/internal/delivery/engine_integration_test.go
@@ -476,7 +476,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
 	t.Parallel()
 	s := newISetup(t)
-	s.Engine.ExportStart()
+	s.Engine.ExportStart(context.Background())
 	event := iSeedEvent(
 		t, s.WebhookDB, s.WebhookID,
@@ -499,17 +499,21 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
 	s.Engine.Notify([]delivery.Task{task})
-	iWaitForDelivered(t, s.WebhookDB, d.ID)
+	iWaitForStatus(
 		t, s.WebhookDB, d.ID,
 		database.DeliveryStatusDelivered,
 	)
 	s.Engine.ExportStop()
 }
-// iWaitForDelivered polls until the delivery reaches the
+// iWaitForStatus polls until the delivery reaches the
-// delivered status.
+// expected status.
-func iWaitForDelivered(
+func iWaitForStatus(
 	t *testing.T,
 	db *gorm.DB,
 	deliveryID string,
 	expected database.DeliveryStatus,
 ) {
 	t.Helper()
@@ -523,7 +527,7 @@ func iWaitForDelivered(
 			return false
 		}
-		return d.Status == database.DeliveryStatusDelivered
+		return d.Status == expected
 	}, 5*time.Second, 50*time.Millisecond)
 }
@@ -554,7 +558,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
 		database.DeliveryStatusRetrying,
 	)
-	s.Engine.ExportStart()
+	s.Engine.ExportStart(context.Background())
 	bodyStr := event.Body
 	cfg := iHTTPConfig(ts.URL)
@@ -565,7 +569,10 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
 	s.Engine.ExportRetryCh() <- task
-	iWaitForDelivered(t, s.WebhookDB, d.ID)
+	iWaitForStatus(
 		t, s.WebhookDB, d.ID,
 		database.DeliveryStatusDelivered,
 	)
 	s.Engine.ExportStop()
 }
--- a/internal/delivery/engine_lifecycle_test.go
+++ b/internal/delivery/engine_lifecycle_test.go
@@ -1,199 +0,0 @@
 package delivery_test
 import (
 	"context"
 	"testing"
 	"time"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/fx"
 	"sneak.berlin/go/webhooker/internal/database"
 	"sneak.berlin/go/webhooker/internal/delivery"
 )
 const (
 	// hookStopTimeout bounds how long a lifecycle test waits for
 	// the engine's OnStop hook to return before declaring the
 	// shutdown hung.
 	hookStopTimeout = 10 * time.Second
 	// hookSettleDelay is how long startEngineViaHook waits after
 	// OnStart before the caller may enqueue work. A worker pool
 	// wrongly rooted in the already-done hook context has nothing
 	// but ctx.Done() ready in its select, so it is deterministically
 	// gone by the end of this window. Without the wait, Notify would
 	// race the pool's very first select, in which a ready ctx.Done()
 	// and a ready deliveryCh are chosen between at random and a
 	// doomed pool still delivers.
 	hookSettleDelay = 250 * time.Millisecond
 )
 // recordingLifecycle is a minimal fx.Lifecycle that records the
 // hooks a component registers, so a test can invoke the real
 // OnStart/OnStop functions with a context of its choosing.
 type recordingLifecycle struct {
 	hooks []fx.Hook
 }
 func (l *recordingLifecycle) Append(h fx.Hook) {
 	l.hooks = append(l.hooks, h)
 }
 // startEngineViaHook drives the genuine fx hooks the application
 // registers for the engine, handing OnStart a context that is
 // already done, and returns only once a pool that inherited that
 // context would have exited. It returns the recorded lifecycle so
 // the caller can drive OnStop too.
 //
 // Callers must not seed pending or retrying deliveries before
 // calling this: restart recovery enqueues those during startup,
 // which would put work in the queue while the pool is still
 // racing its first select.
 func startEngineViaHook(
 	t *testing.T, eng *delivery.Engine,
 ) *recordingLifecycle {
 	t.Helper()
 	lc := &recordingLifecycle{}
 	eng.ExportRegisterHooks(lc)
 	require.Len(t, lc.hooks, 1)
 	// fx hands OnStart a context carrying the application start
 	// timeout, and cancels it when the start phase ends. An
 	// already-cancelled context is that same defect taken to its
 	// limit, and unlike a plain context.Background() it actually
 	// distinguishes a correctly rooted loop from a broken one.
 	hookCtx, cancel := context.WithCancel(context.Background())
 	cancel()
 	require.NoError(t, lc.hooks[0].OnStart(hookCtx))
 	time.Sleep(hookSettleDelay)
 	return lc
 }
 // seedLogTask seeds a pending delivery for a log target and
 // returns its ID together with the task that drives it. The log
 // target needs no network, so a delivery completing proves only
 // that a worker picked the task up.
 func seedLogTask(
 	t *testing.T, s iSetup,
 ) (string, delivery.Task) {
 	t.Helper()
 	event := iSeedEvent(
 		t, s.WebhookDB, s.WebhookID,
 		`{"lifecycle":"hook-context"}`,
 	)
 	targetID := uuid.New().String()
 	d := iSeedDelivery(
 		t, s.WebhookDB, event.ID, targetID,
 		database.DeliveryStatusPending,
 	)
 	bodyStr := event.Body
 	task := iTask(
 		d, event, s.WebhookID, targetID,
 		"hook-context-test", "", 0, 1, &bodyStr,
 	)
 	task.TargetType = database.TargetTypeLog
 	return d.ID, task
 }
 // TestEngine_WorkersOutliveStartHookContext is the regression
 // test for a delivery engine that stopped delivering roughly
 // fifteen seconds after boot. fx calls OnStart with a context
 // carrying the application's start timeout (15s by default) and
 // cancels it when the start phase ends, so a worker pool rooted
 // in it exits shortly after startup: the process keeps accepting
 // and persisting events while nothing at all forwards them.
 //
 // Driving OnStart with an already-cancelled context is that
 // defect taken to its limit. A pool that inherits the hook
 // context is gone before the task is even enqueued; a correctly
 // rooted pool keeps working for as long as the process lives.
 func TestEngine_WorkersOutliveStartHookContext(t *testing.T) {
 	t.Parallel()
 	s := newISetup(t)
 	lc := startEngineViaHook(t, s.Engine)
 	t.Cleanup(func() {
 		_ = lc.hooks[0].OnStop(context.Background())
 	})
 	// Seeded only after the pool has settled, so restart recovery
 	// cannot enqueue it during startup.
 	deliveryID, task := seedLogTask(t, s)
 	s.Engine.Notify([]delivery.Task{task})
 	iWaitForDelivered(t, s.WebhookDB, deliveryID)
 }
 // TestEngine_StopHookStopsWorkers proves the fix did not trade a
 // startup bug for a shutdown hang: now that the worker pool no
 // longer observes the start hook's cancellation, OnStop is the
 // only thing that can stop it, and it must both return promptly
 // and actually leave the pool drained.
 func TestEngine_StopHookStopsWorkers(t *testing.T) {
 	t.Parallel()
 	s := newISetup(t)
 	lc := startEngineViaHook(t, s.Engine)
 	// Let the pool prove it is running before stopping it, so a
 	// fast OnStop cannot pass by stopping something already dead.
 	firstID, firstTask := seedLogTask(t, s)
 	s.Engine.Notify([]delivery.Task{firstTask})
 	iWaitForDelivered(t, s.WebhookDB, firstID)
 	var stopErr error
 	stopped := make(chan struct{})
 	go func() {
 		defer close(stopped)
 		// stop blocks on the workers' WaitGroup, so returning at
 		// all proves every goroutine observed the cancellation.
 		stopErr = lc.hooks[0].OnStop(context.Background())
 	}()
 	select {
 	case <-stopped:
 	case <-time.After(hookStopTimeout):
 		t.Fatal(
 			"OnStop did not return: the delivery engine's " +
 				"WaitGroup is still waiting on a goroutine that " +
 				"never observed cancellation",
 		)
 	}
 	require.NoError(t, stopErr)
 	// With every worker gone, a freshly notified task must sit
 	// untouched in the queue rather than being delivered.
 	secondID, secondTask := seedLogTask(t, s)
 	s.Engine.Notify([]delivery.Task{secondTask})
 	time.Sleep(200 * time.Millisecond)
 	var after database.Delivery
 	require.NoError(
 		t,
 		s.WebhookDB.First(&after, "id = ?", secondID).Error,
 	)
 	require.Equal(
 		t,
 		database.DeliveryStatusPending,
 		after.Status,
 		"a stopped engine must not deliver anything",
 	)
 }
--- a/internal/delivery/export_test.go
+++ b/internal/delivery/export_test.go
@@ -196,16 +196,8 @@ func (e *Engine) ExportRecoverInFlight(
 }
 // ExportStart exposes start for testing.
-func (e *Engine) ExportStart() {
+func (e *Engine) ExportStart(ctx context.Context) {
-	e.start()
+	e.start(ctx)
 }
 // ExportRegisterHooks registers the engine's real fx lifecycle
 // hooks on a lifecycle supplied by a test, so a test can drive
 // the exact OnStart/OnStop functions the application runs and
 // hand OnStart the kind of context fx actually supplies.
 func (e *Engine) ExportRegisterHooks(lc fx.Lifecycle) {
 	e.registerHooks(lc)
 }
 // ExportStop exposes stop for testing.