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" ) // hookStopTimeout bounds how long a lifecycle test waits for the // engine's OnStop hook to return before declaring the shutdown // hung. const hookStopTimeout = 10 * time.Second // 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. It returns the recorded lifecycle so the caller // can drive OnStop too. 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)) 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 never processes a single task; a correctly rooted pool // keeps working for as long as the process lives. func TestEngine_WorkersOutliveStartHookContext(t *testing.T) { t.Parallel() s := newISetup(t) deliveryID, task := seedLogTask(t, s) lc := startEngineViaHook(t, s.Engine) t.Cleanup(func() { _ = lc.hooks[0].OnStop(context.Background()) }) s.Engine.Notify([]delivery.Task{task}) iWaitForStatus( t, s.WebhookDB, deliveryID, database.DeliveryStatusDelivered, ) } // 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}) iWaitForStatus( t, s.WebhookDB, firstID, database.DeliveryStatusDelivered, ) 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", ) }