Root background loops at context.Background() (closes #97) (#100)

The delivery engine worker pool and the retention reaper both rooted their
goroutines in the fx OnStart hook context, which fx cancels 15s into startup.
Both now use context.WithCancel(context.Background()), bounded by OnStop.

internal/delivery/engine.go

@@ -149,18 +149,7 @@ func New(
 		Transport: NewSSRFSafeTransport(),
 	})
 
-	lc.Append(fx.Hook{
-		OnStart: func(ctx context.Context) error {
-			e.start(ctx)
-
-			return nil
-		},
-		OnStop: func(_ context.Context) error {
-			e.stop()
-
-			return nil
-		},
-	})
+	e.registerHooks(lc)
 
 	return e
 }
@@ -210,8 +199,40 @@ func (e *Engine) ScheduleRetry(
 	})
 }
 
-func (e *Engine) start(ctx context.Context) {
-	ctx, cancel := context.WithCancel(ctx)
+// registerHooks wires the engine's start and stop into the fx
+// lifecycle. The start hook's context is deliberately ignored:
+// 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 {

internal/delivery/engine_integration_test.go

@@ -476,7 +476,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
 	t.Parallel()
 
 	s := newISetup(t)
-	s.Engine.ExportStart(context.Background())
+	s.Engine.ExportStart()
 
 	event := iSeedEvent(
 		t, s.WebhookDB, s.WebhookID,
@@ -499,21 +499,17 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
 
 	s.Engine.Notify([]delivery.Task{task})
 
-	iWaitForStatus(
-		t, s.WebhookDB, d.ID,
-		database.DeliveryStatusDelivered,
-	)
+	iWaitForDelivered(t, s.WebhookDB, d.ID)
 
 	s.Engine.ExportStop()
 }
 
-// iWaitForStatus polls until the delivery reaches the
-// expected status.
-func iWaitForStatus(
+// iWaitForDelivered polls until the delivery reaches the
+// delivered status.
+func iWaitForDelivered(
 	t *testing.T,
 	db *gorm.DB,
 	deliveryID string,
-	expected database.DeliveryStatus,
 ) {
 	t.Helper()
 
@@ -527,7 +523,7 @@ func iWaitForStatus(
 			return false
 		}
 
-		return d.Status == expected
+		return d.Status == database.DeliveryStatusDelivered
 	}, 5*time.Second, 50*time.Millisecond)
 }
 
@@ -558,7 +554,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
 		database.DeliveryStatusRetrying,
 	)
 
-	s.Engine.ExportStart(context.Background())
+	s.Engine.ExportStart()
 
 	bodyStr := event.Body
 	cfg := iHTTPConfig(ts.URL)
@@ -569,10 +565,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
 
 	s.Engine.ExportRetryCh() <- task
 
-	iWaitForStatus(
-		t, s.WebhookDB, d.ID,
-		database.DeliveryStatusDelivered,
-	)
+	iWaitForDelivered(t, s.WebhookDB, d.ID)
 
 	s.Engine.ExportStop()
 }

internal/delivery/engine_lifecycle_test.go

@@ -0,0 +1,199 @@
+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",
+	)
+}

internal/delivery/export_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"time"
 
+	"go.uber.org/fx"
 	"gorm.io/gorm"
 	"sneak.berlin/go/webhooker/internal/database"
 )
@@ -189,8 +190,16 @@ func (e *Engine) ExportRecoverInFlight(
 }
 
 // ExportStart exposes start for testing.
-func (e *Engine) ExportStart(ctx context.Context) {
-	e.start(ctx)
+func (e *Engine) ExportStart() {
+	e.start()
+}
+
+// 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.