Compare commits
1 Commits
190cabe0f2
...
issue-97-l
| Author | SHA1 | Date | |
|---|---|---|---|
| 20a050b49d |
6
TODO.md
6
TODO.md
@@ -28,6 +28,12 @@ databases currently grow without bound.
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-08-09 Root the delivery engine's worker pool and the retention
|
||||||
|
reaper's sweep loop at `context.Background()` rather than the fx
|
||||||
|
`OnStart` hook context (#97), which carries fx's 15s start timeout and
|
||||||
|
killed both roughly fifteen seconds after boot: the proxy silently
|
||||||
|
stopped delivering webhooks entirely, and the reaper never ran a
|
||||||
|
single sweep under its default one-hour interval
|
||||||
- 2026-08-07 Update golangci-lint to v2.12.2 (Docker image digest in
|
- 2026-08-07 Update golangci-lint to v2.12.2 (Docker image digest in
|
||||||
`Dockerfile`, release-archive sha256 pins in `script/bootstrap`),
|
`Dockerfile`, release-archive sha256 pins in `script/bootstrap`),
|
||||||
adopt the canonical `.golangci.yml` (v2 `linters.settings` layout so
|
adopt the canonical `.golangci.yml` (v2 `linters.settings` layout so
|
||||||
|
|||||||
@@ -5,6 +5,8 @@ import (
|
|||||||
"log/slog"
|
"log/slog"
|
||||||
"os"
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"go.uber.org/fx"
|
||||||
)
|
)
|
||||||
|
|
||||||
// NewTestRetentionReaper builds a RetentionReaper backed by the given
|
// NewTestRetentionReaper builds a RetentionReaper backed by the given
|
||||||
@@ -29,3 +31,26 @@ func NewTestRetentionReaper(
|
|||||||
func (r *RetentionReaper) ExportSweep(ctx context.Context) {
|
func (r *RetentionReaper) ExportSweep(ctx context.Context) {
|
||||||
r.sweep(ctx)
|
r.sweep(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ExportRegisterHooks registers the reaper'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 (r *RetentionReaper) ExportRegisterHooks(lc fx.Lifecycle) {
|
||||||
|
r.registerHooks(lc)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExportStart starts the reaper's background loop for tests.
|
||||||
|
func (r *RetentionReaper) ExportStart() {
|
||||||
|
r.start()
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExportStop stops the reaper's background loop for tests.
|
||||||
|
func (r *RetentionReaper) ExportStop() {
|
||||||
|
r.stop()
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExportSetInterval overrides the sweep interval for tests.
|
||||||
|
func (r *RetentionReaper) ExportSetInterval(d time.Duration) {
|
||||||
|
r.interval = d
|
||||||
|
}
|
||||||
|
|||||||
@@ -56,9 +56,20 @@ func NewRetentionReaper(
|
|||||||
interval: params.Config.RetentionSweepInterval,
|
interval: params.Config.RetentionSweepInterval,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
r.registerHooks(lc)
|
||||||
|
|
||||||
|
return r
|
||||||
|
}
|
||||||
|
|
||||||
|
// registerHooks wires the reaper's start and stop into the fx
|
||||||
|
// lifecycle. The start hook's context is deliberately ignored: see
|
||||||
|
// start for why the sweep loop must not inherit it.
|
||||||
|
func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
|
||||||
lc.Append(fx.Hook{
|
lc.Append(fx.Hook{
|
||||||
OnStart: func(ctx context.Context) error {
|
//nolint:contextcheck // Not inheriting the hook context is
|
||||||
r.start(ctx)
|
// the point: see start.
|
||||||
|
OnStart: func(_ context.Context) error {
|
||||||
|
r.start()
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
@@ -68,12 +79,20 @@ func NewRetentionReaper(
|
|||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
return r
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *RetentionReaper) start(ctx context.Context) {
|
// start launches the background sweep loop.
|
||||||
ctx, cancel := context.WithCancel(ctx)
|
//
|
||||||
|
// The loop's 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 a loop derived from it dies 45 minutes before its
|
||||||
|
// first tick under the default one-hour sweep interval, leaving a
|
||||||
|
// reaper that never reaps. 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 (r *RetentionReaper) start() {
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
r.cancel = cancel
|
r.cancel = cancel
|
||||||
|
|
||||||
r.wg.Add(1)
|
r.wg.Add(1)
|
||||||
|
|||||||
209
internal/database/retention_lifecycle_test.go
Normal file
209
internal/database/retention_lifecycle_test.go
Normal file
@@ -0,0 +1,209 @@
|
|||||||
|
package database_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"go.uber.org/fx"
|
||||||
|
"gorm.io/gorm"
|
||||||
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
// reaperTestInterval is the sweep interval a lifecycle test
|
||||||
|
// runs the reaper at, so a loop that survives startup produces
|
||||||
|
// an observable sweep quickly.
|
||||||
|
reaperTestInterval = 10 * time.Millisecond
|
||||||
|
|
||||||
|
// reaperStopTimeout bounds how long a lifecycle test waits for
|
||||||
|
// the reaper's OnStop hook to return before declaring the
|
||||||
|
// shutdown hung.
|
||||||
|
reaperStopTimeout = 10 * time.Second
|
||||||
|
|
||||||
|
// reaperTestRetentionDays is the retention policy the lifecycle
|
||||||
|
// tests give their webhook.
|
||||||
|
reaperTestRetentionDays = 30
|
||||||
|
)
|
||||||
|
|
||||||
|
// 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)
|
||||||
|
}
|
||||||
|
|
||||||
|
// startReaperViaHook drives the genuine fx hooks the application
|
||||||
|
// registers for the reaper, handing OnStart a context that is
|
||||||
|
// already done. It returns the recorded lifecycle so the caller
|
||||||
|
// can drive OnStop too.
|
||||||
|
func startReaperViaHook(
|
||||||
|
t *testing.T, r *database.RetentionReaper,
|
||||||
|
) *recordingLifecycle {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
lc := &recordingLifecycle{}
|
||||||
|
r.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
|
||||||
|
}
|
||||||
|
|
||||||
|
// eventGone reports whether an event row has been removed. It
|
||||||
|
// takes no *testing.T because it is polled from an
|
||||||
|
// assert.Eventually condition, which runs off the test goroutine
|
||||||
|
// where testify assertions must not be used.
|
||||||
|
func eventGone(db *gorm.DB, eventID string) bool {
|
||||||
|
var n int64
|
||||||
|
|
||||||
|
err := db.Unscoped().Model(&database.Event{}).
|
||||||
|
Where("id = ?", eventID).Count(&n).Error
|
||||||
|
if err != nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
return n == 0
|
||||||
|
}
|
||||||
|
|
||||||
|
// seedExpiredWebhook creates a webhook with a finite retention
|
||||||
|
// policy plus one long-expired event chain, and returns the
|
||||||
|
// webhook's database and the chain's event ID.
|
||||||
|
func seedExpiredWebhook(
|
||||||
|
t *testing.T, env *retentionTestEnv,
|
||||||
|
) (*gorm.DB, string) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
webhookID := createWebhook(
|
||||||
|
t, env.mainDB.DB(), reaperTestRetentionDays,
|
||||||
|
)
|
||||||
|
|
||||||
|
db, err := env.mgr.GetDB(webhookID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
chain := seedEventChain(
|
||||||
|
t, db, webhookID,
|
||||||
|
time.Now().Add(-365*24*time.Hour),
|
||||||
|
)
|
||||||
|
|
||||||
|
return db, chain.eventID
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRetentionReaper_LoopOutlivesStartHookContext is the
|
||||||
|
// regression test for a reaper that never reaped. 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
|
||||||
|
// sweep loop rooted in it is dead three quarters of an hour
|
||||||
|
// before its first tick under the default one-hour interval, and
|
||||||
|
// per-webhook event databases grow without bound exactly as they
|
||||||
|
// did before retention existed.
|
||||||
|
//
|
||||||
|
// Driving OnStart with an already-cancelled context is that
|
||||||
|
// defect taken to its limit: a loop that inherits the hook
|
||||||
|
// context never ticks once, while a correctly rooted loop keeps
|
||||||
|
// sweeping for as long as the process lives.
|
||||||
|
func TestRetentionReaper_LoopOutlivesStartHookContext(
|
||||||
|
t *testing.T,
|
||||||
|
) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := setupRetentionTest(t)
|
||||||
|
|
||||||
|
db, eventID := seedExpiredWebhook(t, env)
|
||||||
|
|
||||||
|
env.reaper.ExportSetInterval(reaperTestInterval)
|
||||||
|
|
||||||
|
lc := startReaperViaHook(t, env.reaper)
|
||||||
|
t.Cleanup(func() {
|
||||||
|
_ = lc.hooks[0].OnStop(context.Background())
|
||||||
|
})
|
||||||
|
|
||||||
|
assert.Eventually(
|
||||||
|
t,
|
||||||
|
func() bool { return eventGone(db, eventID) },
|
||||||
|
5*time.Second,
|
||||||
|
reaperTestInterval,
|
||||||
|
"the sweep loop must keep running after the start "+
|
||||||
|
"hook's context is done; it reaped nothing, so it "+
|
||||||
|
"inherited the hook context and died",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRetentionReaper_StopHookStopsLoop proves the fix did not
|
||||||
|
// trade a startup bug for a shutdown hang: now that the sweep
|
||||||
|
// loop 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 loop stopped.
|
||||||
|
func TestRetentionReaper_StopHookStopsLoop(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := setupRetentionTest(t)
|
||||||
|
|
||||||
|
db, eventID := seedExpiredWebhook(t, env)
|
||||||
|
|
||||||
|
env.reaper.ExportSetInterval(reaperTestInterval)
|
||||||
|
|
||||||
|
lc := startReaperViaHook(t, env.reaper)
|
||||||
|
|
||||||
|
// Let the loop prove it is running before stopping it, so a
|
||||||
|
// fast OnStop cannot pass by stopping something already dead.
|
||||||
|
require.Eventually(
|
||||||
|
t,
|
||||||
|
func() bool { return eventGone(db, eventID) },
|
||||||
|
5*time.Second,
|
||||||
|
reaperTestInterval,
|
||||||
|
)
|
||||||
|
|
||||||
|
var stopErr error
|
||||||
|
|
||||||
|
stopped := make(chan struct{})
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
defer close(stopped)
|
||||||
|
|
||||||
|
// stop blocks on the loop's WaitGroup, so returning at all
|
||||||
|
// proves the goroutine observed the cancellation.
|
||||||
|
stopErr = lc.hooks[0].OnStop(context.Background())
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-stopped:
|
||||||
|
case <-time.After(reaperStopTimeout):
|
||||||
|
t.Fatal(
|
||||||
|
"OnStop did not return: the retention reaper's " +
|
||||||
|
"WaitGroup is still waiting on a loop that never " +
|
||||||
|
"observed cancellation",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NoError(t, stopErr)
|
||||||
|
|
||||||
|
// With the loop gone, a newly expired chain must survive.
|
||||||
|
survivor := seedEventChain(
|
||||||
|
t, db, "stopped-webhook",
|
||||||
|
time.Now().Add(-365*24*time.Hour),
|
||||||
|
)
|
||||||
|
|
||||||
|
time.Sleep(20 * reaperTestInterval)
|
||||||
|
|
||||||
|
assert.False(
|
||||||
|
t,
|
||||||
|
eventGone(db, survivor.eventID),
|
||||||
|
"a stopped reaper must not sweep anything",
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -149,18 +149,7 @@ func New(
|
|||||||
Transport: NewSSRFSafeTransport(),
|
Transport: NewSSRFSafeTransport(),
|
||||||
})
|
})
|
||||||
|
|
||||||
lc.Append(fx.Hook{
|
e.registerHooks(lc)
|
||||||
OnStart: func(ctx context.Context) error {
|
|
||||||
e.start(ctx)
|
|
||||||
|
|
||||||
return nil
|
|
||||||
},
|
|
||||||
OnStop: func(_ context.Context) error {
|
|
||||||
e.stop()
|
|
||||||
|
|
||||||
return nil
|
|
||||||
},
|
|
||||||
})
|
|
||||||
|
|
||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
@@ -210,8 +199,40 @@ func (e *Engine) ScheduleRetry(
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
func (e *Engine) start(ctx context.Context) {
|
// registerHooks wires the engine's start and stop into the fx
|
||||||
ctx, cancel := context.WithCancel(ctx)
|
// 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
|
e.cancel = cancel
|
||||||
|
|
||||||
for range e.workers {
|
for range e.workers {
|
||||||
|
|||||||
@@ -476,7 +476,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
|
|||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
s := newISetup(t)
|
s := newISetup(t)
|
||||||
s.Engine.ExportStart(context.Background())
|
s.Engine.ExportStart()
|
||||||
|
|
||||||
event := iSeedEvent(
|
event := iSeedEvent(
|
||||||
t, s.WebhookDB, s.WebhookID,
|
t, s.WebhookDB, s.WebhookID,
|
||||||
@@ -499,21 +499,17 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
|
|||||||
|
|
||||||
s.Engine.Notify([]delivery.Task{task})
|
s.Engine.Notify([]delivery.Task{task})
|
||||||
|
|
||||||
iWaitForStatus(
|
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
||||||
t, s.WebhookDB, d.ID,
|
|
||||||
database.DeliveryStatusDelivered,
|
|
||||||
)
|
|
||||||
|
|
||||||
s.Engine.ExportStop()
|
s.Engine.ExportStop()
|
||||||
}
|
}
|
||||||
|
|
||||||
// iWaitForStatus polls until the delivery reaches the
|
// iWaitForDelivered polls until the delivery reaches the
|
||||||
// expected status.
|
// delivered status.
|
||||||
func iWaitForStatus(
|
func iWaitForDelivered(
|
||||||
t *testing.T,
|
t *testing.T,
|
||||||
db *gorm.DB,
|
db *gorm.DB,
|
||||||
deliveryID string,
|
deliveryID string,
|
||||||
expected database.DeliveryStatus,
|
|
||||||
) {
|
) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
@@ -527,7 +523,7 @@ func iWaitForStatus(
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
return d.Status == expected
|
return d.Status == database.DeliveryStatusDelivered
|
||||||
}, 5*time.Second, 50*time.Millisecond)
|
}, 5*time.Second, 50*time.Millisecond)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -558,7 +554,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
|
|||||||
database.DeliveryStatusRetrying,
|
database.DeliveryStatusRetrying,
|
||||||
)
|
)
|
||||||
|
|
||||||
s.Engine.ExportStart(context.Background())
|
s.Engine.ExportStart()
|
||||||
|
|
||||||
bodyStr := event.Body
|
bodyStr := event.Body
|
||||||
cfg := iHTTPConfig(ts.URL)
|
cfg := iHTTPConfig(ts.URL)
|
||||||
@@ -569,10 +565,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
|
|||||||
|
|
||||||
s.Engine.ExportRetryCh() <- task
|
s.Engine.ExportRetryCh() <- task
|
||||||
|
|
||||||
iWaitForStatus(
|
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
||||||
t, s.WebhookDB, d.ID,
|
|
||||||
database.DeliveryStatusDelivered,
|
|
||||||
)
|
|
||||||
|
|
||||||
s.Engine.ExportStop()
|
s.Engine.ExportStop()
|
||||||
}
|
}
|
||||||
|
|||||||
199
internal/delivery/engine_lifecycle_test.go
Normal file
199
internal/delivery/engine_lifecycle_test.go
Normal file
@@ -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",
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"go.uber.org/fx"
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
"sneak.berlin/go/webhooker/internal/database"
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
)
|
)
|
||||||
@@ -189,8 +190,16 @@ func (e *Engine) ExportRecoverInFlight(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ExportStart exposes start for testing.
|
// ExportStart exposes start for testing.
|
||||||
func (e *Engine) ExportStart(ctx context.Context) {
|
func (e *Engine) ExportStart() {
|
||||||
e.start(ctx)
|
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.
|
// ExportStop exposes stop for testing.
|
||||||
|
|||||||
Reference in New Issue
Block a user