webhooker/internal/delivery/engine_lifecycle_test.go

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