webhooker/internal/delivery/engine_lifecycle_test.go

package delivery_test

import (
	"context"
	"io"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.uber.org/fx"
	"gorm.io/gorm/clause"
	"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")
}

// seedHeldDelivery gives the setup's webhook an http target with
// maxRetries, pointed at a receiver that keeps every request open until
// the client gives up on it, and seeds one delivery to that target at
// status. It returns the task that sends attempt attemptNum of the
// delivery, and a channel that is closed when a request reaches the
// receiver.
func seedHeldDelivery(
	t *testing.T,
	s iSetup,
	maxRetries int,
	status database.DeliveryStatus,
	attemptNum int,
) (delivery.Task, <-chan struct{}) {
	t.Helper()

	arrived := make(chan struct{})

	var once sync.Once

	ts := httptest.NewServer(http.HandlerFunc(
		func(_ http.ResponseWriter, r *http.Request) {
			// The server notices the client hanging up only once the
			// body has been read; until then r.Context() never ends.
			_, _ = io.Copy(io.Discard, r.Body)

			once.Do(func() { close(arrived) })

			<-r.Context().Done()
		},
	))
	t.Cleanup(ts.Close)

	cfg := iHTTPConfig(ts.URL)
	targetID := uuid.New().String()

	iCreateWebhook(t, s.MainDB, s.WebhookID, "hook")
	iCreateTarget(
		t, s.MainDB, targetID, s.WebhookID, "held",
		database.TargetTypeHTTP, cfg, maxRetries,
	)

	event := iSeedEvent(t, s.WebhookDB, s.WebhookID, `{"held":true}`)
	d := iSeedDelivery(t, s.WebhookDB, event.ID, targetID, status)

	bodyStr := event.Body

	return iTask(
		d, event, s.WebhookID, targetID, "held", cfg,
		maxRetries, attemptNum, &bodyStr,
	), arrived
}

// stopDuringAttempt waits until the receiver holds an attempt open, then
// stops the engine through its real stop hook.
func stopDuringAttempt(
	t *testing.T, lc *recordingLifecycle, arrived <-chan struct{},
) {
	t.Helper()

	select {
	case <-arrived:
	case <-time.After(5 * time.Second):
		t.Fatal("the delivery attempt never reached the receiver")
	}

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

// requireRequeued runs restart recovery on the stopped engine and
// requires it to queue task's attempt again on ch.
func requireRequeued(
	t *testing.T, s iSetup, ch <-chan delivery.Task, task delivery.Task,
) {
	t.Helper()

	s.Engine.ExportRecoverWebhookDeliveries(
		context.Background(), s.WebhookID,
	)

	select {
	case got := <-ch:
		assert.Equal(t, task.DeliveryID, got.DeliveryID)
		assert.Equal(t, task.AttemptNum, got.AttemptNum)
	case <-time.After(5 * time.Second):
		t.Fatal("restart recovery did not queue the delivery again")
	}
}

// TestEngine_StopLeavesFireAndForgetAttemptPending is the regression
// test for a stop that turned an attempt in flight into a terminal
// failure. Stopping the engine cancels the request; for a target with
// no retries that was the delivery's only attempt, recorded as failed
// with "context canceled", and nothing ever sent it again. The
// delivery must instead stay pending, with no result, for restart
// recovery to send.
func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) {
	t.Parallel()

	s := newISetup(t)
	lc := startEngineViaHook(t, s.Engine)

	task, arrived := seedHeldDelivery(
		t, s, 0, database.DeliveryStatusPending, 1,
	)

	s.Engine.Notify([]delivery.Task{task})
	stopDuringAttempt(t, lc, arrived)

	iAssertStatus(
		t, s.WebhookDB, task.DeliveryID, database.DeliveryStatusPending,
	)
	assert.Empty(t, iResults(t, s.WebhookDB, task.DeliveryID))

	requireRequeued(t, s, s.Engine.ExportDeliveryCh(), task)
}

// TestEngine_StopLeavesLastRetryAttemptRetrying is the same regression
// for a target with retries, on its last attempt: the cancelled attempt
// used to exhaust the retries and end the delivery failed. It must stay
// retrying with only its earlier attempt recorded, and restart recovery
// must send that last attempt again.
func TestEngine_StopLeavesLastRetryAttemptRetrying(t *testing.T) {
	t.Parallel()

	const maxRetries = 2

	s := newISetup(t)
	lc := startEngineViaHook(t, s.Engine)

	task, arrived := seedHeldDelivery(
		t, s, maxRetries, database.DeliveryStatusRetrying, maxRetries,
	)
	iSeedFailedResult(t, s.WebhookDB, task.DeliveryID)

	s.Engine.ExportScheduleRetry(task, 0)
	stopDuringAttempt(t, lc, arrived)

	iAssertStatus(
		t, s.WebhookDB, task.DeliveryID, database.DeliveryStatusRetrying,
	)
	assert.Len(t, iResults(t, s.WebhookDB, task.DeliveryID), 1)

	requireRequeued(t, s, s.Engine.ExportRetryCh(), task)
}

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

	iCreateWebhook(t, s.MainDB, s.WebhookID, "hook")

	tgt := &database.Target{
		WebhookID: s.WebhookID,
		Name:      "archive",
		Type:      database.TargetTypeDatabase,
	}
	require.NoError(
		t, s.MainDB.Omit(clause.Associations).Create(tgt).Error,
	)

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

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

	iWaitForDelivered(t, s.WebhookDB, deliveryID)

	return tgt.ID, filepath.Join(
		filepath.Dir(s.DBMgr.DBPath(s.WebhookID)),
		"archive-hook-archive-"+tgt.ID+".db",
	)
}

// 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)

	targetID, _ := 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(targetID),
		"a stop that timed out must not close archive writers",
	)
}