1 Commits

Author SHA1 Message Date
clawbot
0384de4a7b Terminally fail retrying deliveries with a non-retry target type (closes #82)
All checks were successful
check / check (push) Successful in 4m4s
Restart recovery and the 60s retry sweep both looked an orphaned
`retrying` delivery's target up in the registry and silently returned
when it did not implement `rescheduler`. If a target's type was edited
from a retry type (`http`/`slack`) to a fire-and-forget type
(`database`/`log`) or an unknown one while a delivery was still
retrying, that delivery stayed `retrying` forever.

Both sites now hand the delivery to one shared helper,
`failUnretryableRetry`, which records a `DeliveryResult` naming the
current target type as the reason and marks the delivery `failed`. It
logs at warn, not error: this is operator-caused state, not a system
fault.

Re-dispatching under the new type was rejected as it would perform a
delivery the operator never asked for; the event itself stays in the
per-webhook event database, so manual redelivery can recover it
deliberately.

Fire-and-forget targets never set status `retrying` under normal
operation, so this path stays unreachable for them in practice.
2026-08-09 05:51:51 +00:00
9 changed files with 312 additions and 521 deletions

View File

@@ -636,6 +636,18 @@ This means:
durable fallback that ensures no retry is permanently lost, even under durable fallback that ensures no retry is permanently lost, even under
extreme backpressure. extreme backpressure.
**Changing a target's type does not migrate in-flight deliveries.** Only
`http` and `slack` targets own durable retries; `database` and `log`
targets are fire-and-forget and never produce a `retrying` delivery. If a
target's `type` is edited from a retrying type to a non-retrying (or
unknown) one while one of its deliveries is still `retrying`, both
recovery paths above terminally mark that delivery `failed` and record a
`DeliveryResult` naming the current target type as the reason, logging it
at warn level. The delivery is not re-dispatched under the new type — the
operator never asked for that delivery — and the event itself remains
stored in the per-webhook event database, so it can be redelivered
manually.
### Circuit Breaker (HTTP Targets with Retries) ### Circuit Breaker (HTTP Targets with Retries)
HTTP targets with `max_retries` > 0 are protected by a **per-target circuit breaker** that HTTP targets with `max_retries` > 0 are protected by a **per-target circuit breaker** that

10
TODO.md
View File

@@ -28,12 +28,10 @@ databases currently grow without bound.
# Completed Steps # Completed Steps
- 2026-08-09 Root the delivery engine's worker pool and the retention - 2026-08-09 Restart recovery and the 60s retry sweep terminally fail an
reaper's sweep loop at `context.Background()` rather than the fx orphaned `retrying` delivery whose target type no longer supports
`OnStart` hook context (#97), which carries fx's 15s start timeout and retries, recording a `DeliveryResult` with the reason instead of
killed both roughly fifteen seconds after boot: the proxy silently leaving the delivery stuck forever (#82)
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

View File

@@ -5,8 +5,6 @@ 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
@@ -31,26 +29,3 @@ 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
}

View File

@@ -56,20 +56,9 @@ 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{
//nolint:contextcheck // Not inheriting the hook context is OnStart: func(ctx context.Context) error {
// the point: see start. r.start(ctx)
OnStart: func(_ context.Context) error {
r.start()
return nil return nil
}, },
@@ -79,20 +68,12 @@ func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
}) })
return r
} }
// start launches the background sweep loop. func (r *RetentionReaper) start(ctx context.Context) {
// 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)

View File

@@ -1,209 +0,0 @@
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",
)
}

View File

@@ -149,7 +149,18 @@ func New(
Transport: NewSSRFSafeTransport(), Transport: NewSSRFSafeTransport(),
}) })
e.registerHooks(lc) lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
e.start(ctx)
return nil
},
OnStop: func(_ context.Context) error {
e.stop()
return nil
},
})
return e return e
} }
@@ -199,40 +210,8 @@ func (e *Engine) ScheduleRetry(
}) })
} }
// registerHooks wires the engine's start and stop into the fx func (e *Engine) start(ctx context.Context) {
// lifecycle. The start hook's context is deliberately ignored: ctx, cancel := context.WithCancel(ctx)
// 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 {
@@ -474,8 +453,9 @@ func (e *Engine) recoverRetryingDeliveries(
// recoverSingleRetry hands an orphaned retrying delivery back // recoverSingleRetry hands an orphaned retrying delivery back
// to its target to recompute the remaining backoff, then // to its target to recompute the remaining backoff, then
// reschedules it. Targets that do not own durable retries // reschedules it. Targets that do not own durable retries
// (fire-and-forget) never produce retrying deliveries, so // (fire-and-forget) never produce retrying deliveries, so a
// they are skipped. // delivery found in that state has had its target's type
// changed underneath it and is terminally failed.
func (e *Engine) recoverSingleRetry( func (e *Engine) recoverSingleRetry(
webhookDB *gorm.DB, webhookDB *gorm.DB,
webhookID string, webhookID string,
@@ -496,6 +476,10 @@ func (e *Engine) recoverSingleRetry(
rs, ok := e.targets[target.Type].(rescheduler) rs, ok := e.targets[target.Type].(rescheduler)
if !ok { if !ok {
e.failUnretryableRetry(
webhookDB, webhookID, d, &target,
)
return return
} }
@@ -670,8 +654,8 @@ func (e *Engine) sweepWebhookRetries(
// sweepSingleRetry re-enqueues an orphaned retrying delivery // sweepSingleRetry re-enqueues an orphaned retrying delivery
// whose backoff window has elapsed, delegating the backoff // whose backoff window has elapsed, delegating the backoff
// decision to the delivery's target. Targets that do not own // decision to the delivery's target. A delivery whose target
// durable retries are skipped. // no longer owns durable retries is terminally failed.
func (e *Engine) sweepSingleRetry( func (e *Engine) sweepSingleRetry(
webhookDB *gorm.DB, webhookDB *gorm.DB,
webhookID string, webhookID string,
@@ -691,6 +675,10 @@ func (e *Engine) sweepSingleRetry(
rs, ok := e.targets[target.Type].(rescheduler) rs, ok := e.targets[target.Type].(rescheduler)
if !ok { if !ok {
e.failUnretryableRetry(
webhookDB, webhookID, d, &target,
)
return return
} }
@@ -731,6 +719,59 @@ func (e *Engine) sweepSingleRetry(
} }
} }
// failUnretryableRetry terminally fails an orphaned retrying
// delivery whose target type no longer supports retries. Both
// restart recovery and the periodic sweep call it, so the
// terminal transition exists once.
//
// This is only reachable when a target's type has been changed
// out from under an in-flight retrying delivery (or the type is
// unknown to the registry): fire-and-forget targets never set
// status retrying themselves. Re-dispatching under the new type
// would be a delivery the operator never asked for, and leaving
// the row retrying strands it forever, so the delivery is
// failed with a recorded reason and can be redelivered
// manually. Logged at warn, not error: this is operator-caused
// state, not a system fault.
func (e *Engine) failUnretryableRetry(
webhookDB *gorm.DB,
webhookID string,
d *database.Delivery,
target *database.Target,
) {
e.log.Warn(
"failing orphaned retrying delivery: target "+
"type no longer supports retries",
"webhook_id", webhookID,
"delivery_id", d.ID,
"target_id", target.ID,
"target_name", target.Name,
"target_type", target.Type,
)
reason := fmt.Sprintf(
"target type %q does not support retries; "+
"delivery was left retrying by a previous "+
"target type and has been failed terminally",
target.Type,
)
e.recordResult(
webhookDB,
d,
e.countAttempts(webhookDB, d.ID)+1,
false,
0,
"",
reason,
0,
)
e.updateDeliveryStatus(
webhookDB, d, database.DeliveryStatusFailed,
)
}
// processDelivery dispatches a delivery to the target that // processDelivery dispatches a delivery to the target that
// owns its type. Unknown target types fail the delivery. // owns its type. Unknown target types fail the delivery.
func (e *Engine) processDelivery( func (e *Engine) processDelivery(

View File

@@ -476,7 +476,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
t.Parallel() t.Parallel()
s := newISetup(t) s := newISetup(t)
s.Engine.ExportStart() s.Engine.ExportStart(context.Background())
event := iSeedEvent( event := iSeedEvent(
t, s.WebhookDB, s.WebhookID, t, s.WebhookDB, s.WebhookID,
@@ -499,17 +499,21 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
s.Engine.Notify([]delivery.Task{task}) s.Engine.Notify([]delivery.Task{task})
iWaitForDelivered(t, s.WebhookDB, d.ID) iWaitForStatus(
t, s.WebhookDB, d.ID,
database.DeliveryStatusDelivered,
)
s.Engine.ExportStop() s.Engine.ExportStop()
} }
// iWaitForDelivered polls until the delivery reaches the // iWaitForStatus polls until the delivery reaches the
// delivered status. // expected status.
func iWaitForDelivered( func iWaitForStatus(
t *testing.T, t *testing.T,
db *gorm.DB, db *gorm.DB,
deliveryID string, deliveryID string,
expected database.DeliveryStatus,
) { ) {
t.Helper() t.Helper()
@@ -523,7 +527,7 @@ func iWaitForDelivered(
return false return false
} }
return d.Status == database.DeliveryStatusDelivered return d.Status == expected
}, 5*time.Second, 50*time.Millisecond) }, 5*time.Second, 50*time.Millisecond)
} }
@@ -554,7 +558,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
database.DeliveryStatusRetrying, database.DeliveryStatusRetrying,
) )
s.Engine.ExportStart() s.Engine.ExportStart(context.Background())
bodyStr := event.Body bodyStr := event.Body
cfg := iHTTPConfig(ts.URL) cfg := iHTTPConfig(ts.URL)
@@ -565,7 +569,10 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
s.Engine.ExportRetryCh() <- task s.Engine.ExportRetryCh() <- task
iWaitForDelivered(t, s.WebhookDB, d.ID) iWaitForStatus(
t, s.WebhookDB, d.ID,
database.DeliveryStatusDelivered,
)
s.Engine.ExportStop() s.Engine.ExportStop()
} }
@@ -741,6 +748,193 @@ func TestRecoverWebhookDeliveries_RetryingDeliveries(
case <-time.After(5 * time.Second): case <-time.After(5 * time.Second):
t.Fatal("expected retry task from recovery") t.Fatal("expected retry task from recovery")
} }
// Regression guard: a target that still supports retries
// must be rescheduled, never terminally failed, and must
// not gain a synthetic result row.
iAssertStatus(
t, s.WebhookDB, d.ID,
database.DeliveryStatusRetrying,
)
assert.Len(t, iResults(t, s.WebhookDB, d.ID), 1)
}
// --- Retrying deliveries whose target type changed ---
// iSeedRetryingWithType seeds a retrying delivery with one
// recorded failed attempt against a target of the given type,
// standing in for a target whose type was edited in the main
// database while the delivery was still retrying.
func iSeedRetryingWithType(
t *testing.T,
s iSetup,
targetType database.TargetType,
) string {
t.Helper()
targetID := uuid.New().String()
iCreateTarget(t, s.MainDB, targetID,
s.WebhookID, "mutated-target", targetType,
iHTTPConfig("http://example.com/hook"), 5,
)
event := iSeedEvent(
t, s.WebhookDB, s.WebhookID,
`{"orphaned":"retry"}`,
)
d := iSeedDelivery(
t, s.WebhookDB, event.ID, targetID,
database.DeliveryStatusRetrying,
)
iSeedFailedResult(t, s.WebhookDB, d.ID)
return d.ID
}
// iResults loads a delivery's results in attempt order.
func iResults(
t *testing.T, db *gorm.DB, deliveryID string,
) []database.DeliveryResult {
t.Helper()
var results []database.DeliveryResult
require.NoError(t, db.
Where("delivery_id = ?", deliveryID).
Order("attempt_num").
Find(&results).Error)
return results
}
// iAssertTerminallyFailed asserts the delivery ended failed
// with a result row recording why, and was not rescheduled.
func iAssertTerminallyFailed(
t *testing.T,
s iSetup,
deliveryID string,
targetType database.TargetType,
) {
t.Helper()
iAssertStatus(
t, s.WebhookDB, deliveryID,
database.DeliveryStatusFailed,
)
results := iResults(t, s.WebhookDB, deliveryID)
require.Len(t, results, 2)
last := results[1]
assert.False(t, last.Success)
assert.Equal(t, 2, last.AttemptNum)
assert.Contains(
t, last.Error, string(targetType),
)
assert.Contains(
t, last.Error, "does not support retries",
)
assert.Empty(t, s.Engine.ExportRetryCh())
}
func TestRecoverSingleRetry_TypeNoLongerRetries(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "mutated-type",
)
deliveryID := iSeedRetryingWithType(
t, s, database.TargetTypeLog,
)
s.Engine.ExportRecoverWebhookDeliveries(
context.Background(), s.WebhookID,
)
iAssertTerminallyFailed(
t, s, deliveryID, database.TargetTypeLog,
)
}
func TestSweepSingleRetry_TypeNoLongerRetries(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "mutated-type-sweep",
)
deliveryID := iSeedRetryingWithType(
t, s, database.TargetTypeDatabase,
)
s.Engine.ExportSweepWebhookRetries(
context.Background(), s.WebhookID,
)
iAssertTerminallyFailed(
t, s, deliveryID, database.TargetTypeDatabase,
)
}
func TestRecoverSingleRetry_UnknownTargetType(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "unknown-type",
)
unknown := database.TargetType("not-a-target-type")
deliveryID := iSeedRetryingWithType(t, s, unknown)
s.Engine.ExportRecoverWebhookDeliveries(
context.Background(), s.WebhookID,
)
iAssertTerminallyFailed(t, s, deliveryID, unknown)
}
func TestSweepSingleRetry_UnknownTargetType(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "unknown-type-sweep",
)
unknown := database.TargetType("not-a-target-type")
deliveryID := iSeedRetryingWithType(t, s, unknown)
s.Engine.ExportSweepWebhookRetries(
context.Background(), s.WebhookID,
)
iAssertTerminallyFailed(t, s, deliveryID, unknown)
} }
// iSeedFailedResult creates a failed delivery result. // iSeedFailedResult creates a failed delivery result.

View File

@@ -1,199 +0,0 @@
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",
)
}

View File

@@ -7,7 +7,6 @@ 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,17 +188,16 @@ func (e *Engine) ExportRecoverInFlight(
e.recoverInFlight(ctx) e.recoverInFlight(ctx)
} }
// ExportStart exposes start for testing. // ExportSweepWebhookRetries exposes sweepWebhookRetries.
func (e *Engine) ExportStart() { func (e *Engine) ExportSweepWebhookRetries(
e.start() ctx context.Context, webhookID string,
) {
e.sweepWebhookRetries(ctx, webhookID)
} }
// ExportRegisterHooks registers the engine's real fx lifecycle // ExportStart exposes start for testing.
// hooks on a lifecycle supplied by a test, so a test can drive func (e *Engine) ExportStart(ctx context.Context) {
// the exact OnStart/OnStop functions the application runs and e.start(ctx)
// 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.