webhooker/internal/delivery/engine_lifecycle_test.go

package delivery_test

import (
	"context"
	"fmt"
	"path/filepath"
	"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

	// wedgeStopTimeout is the stop timeout a wedged-shutdown test
	// hands OnStop, standing in for fx's StopTimeout. The test
	// asserts only that the hook returns at all, and allows it
	// hookStopTimeout — forty times this budget — to do so, so no
	// assertion here races the wall clock.
	wedgeStopTimeout = 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)
}

// requireStopHookExpires drives hook.OnStop with a stop context
// that expires while a wedged goroutine is still running, and
// requires the hook to return the deadline error naming
// component instead of blocking on the WaitGroup forever.
func requireStopHookExpires(
	t *testing.T, hook fx.Hook, component string,
) {
	t.Helper()

	stopCtx, cancel := context.WithTimeout(
		context.Background(), wedgeStopTimeout,
	)
	defer cancel()

	var stopErr error

	stopped := make(chan struct{})

	go func() {
		defer close(stopped)

		stopErr = hook.OnStop(stopCtx)
	}()

	select {
	case <-stopped:
	case <-time.After(hookStopTimeout):
		t.Fatal(
			"OnStop did not return: it discarded the stop " +
				"context and is waiting on a wedged goroutine " +
				"that will never observe cancellation",
		)
	}

	require.ErrorIs(t, stopErr, context.DeadlineExceeded)
	require.ErrorContains(t, stopErr, component)
}

// 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",
	)
}

// TestEngine_StopHookHonoursStopTimeout is the regression test
// for a shutdown that could never complete. fx hands OnStop a
// context carrying the application's stop timeout; an OnStop
// that discards it and calls wg.Wait() bare hangs the process
// forever on a single worker stuck inside a delivery target that
// never returns — precisely when a bounded shutdown matters
// most.
//
// The wedged goroutine here never observes cancellation, so the
// hook can only return by honouring its context, and it must say
// so rather than reporting a clean stop.
func TestEngine_StopHookHonoursStopTimeout(t *testing.T) {
	t.Parallel()

	s := newISetup(t)

	lc := startEngineViaHook(t, s.Engine)

	release := make(chan struct{})

	t.Cleanup(func() { close(release) })

	s.Engine.ExportWedgeWorker(release)

	requireStopHookExpires(t, lc.hooks[0], "delivery engine")
}

// deliverToArchive runs one delivery to a database target through
// the running engine and returns the webhook's archive file path.
// The archive writer holds the file open afterwards.
func deliverToArchive(t *testing.T, s iSetup) string {
	t.Helper()

	deliveryID, task := seedLogTask(t, s)
	task.TargetType = database.TargetTypeDatabase

	s.Engine.Notify([]delivery.Task{task})

	iWaitForDelivered(t, s.WebhookDB, deliveryID)

	return filepath.Join(
		filepath.Dir(s.DBMgr.DBPath(s.WebhookID)),
		fmt.Sprintf("archive-%s.db", s.WebhookID),
	)
}

// TestEngine_StopHookClosesArchives is the regression test for an
// archive split across two files by a clean stop. The engine never
// closed its archive writers, so after a stop the archived rows
// could sit in archive-{id}.db-wal while archive-{id}.db held no
// table at all, and copying the .db on its own gave an empty
// database.
func TestEngine_StopHookClosesArchives(t *testing.T) {
	t.Parallel()

	s := newISetup(t)

	lc := startEngineViaHook(t, s.Engine)

	path := deliverToArchive(t, s)
	require.FileExists(
		t, path+"-wal",
		"an open archive should have a -wal for the stop to remove",
	)

	require.NoError(t, lc.hooks[0].OnStop(context.Background()))

	wals, err := filepath.Glob(
		filepath.Join(filepath.Dir(path), "archive-*.db-wal"),
	)
	require.NoError(t, err)
	require.Empty(
		t, wals, "a clean stop must leave no archive -wal behind",
	)

	// With no -wal beside it, the row can only be in the .db.
	count, err := countArchivedRows(path)
	require.NoError(t, err)
	require.Equal(t, int64(1), count)
}

// TestEngine_StopHookTimeoutLeavesArchivesOpen covers a stop whose
// budget runs out while a worker is still running. The archive
// writers are left open, as a kill would leave them: closing them
// would wait for any write in progress, and that worker would then
// open new writers that nothing closes.
func TestEngine_StopHookTimeoutLeavesArchivesOpen(t *testing.T) {
	t.Parallel()

	s := newISetup(t)

	lc := startEngineViaHook(t, s.Engine)

	deliverToArchive(t, s)

	release := make(chan struct{})

	t.Cleanup(func() {
		close(release)
		s.Engine.EvictWebhook(s.WebhookID)
	})

	s.Engine.ExportWedgeWorker(release)

	requireStopHookExpires(t, lc.hooks[0], "delivery engine")

	require.True(
		t, s.Engine.ExportArchiveHandleOpen(s.WebhookID),
		"a stop that timed out must not close archive writers",
	)
}