1 Commits

Author SHA1 Message Date
e83eb2977e Bound shutdown hooks by their stop context (closes #102)
All checks were successful
check / check (push) Successful in 3m45s
fx hands OnStop a context carrying the application's stop timeout,
and the delivery engine, the retention reaper, and the archive
sweeper all discarded it and called wg.Wait() bare. A worker wedged
inside a delivery target that never returns, or a sweep blocked on a
locked SQLite database, hung the process forever instead of letting
it exit when the timeout expired.

All three now wait through internal/lifecycle.WaitForShutdown, which
selects the drained WaitGroup against the stop context and, on
timeout, logs at error naming the component and returns an error
rather than reporting a clean stop.

Engine.stop also gains the cancel != nil guard its two mirrored
components already had.
2026-08-12 09:45:25 +00:00
14 changed files with 396 additions and 131 deletions

View File

@@ -46,8 +46,19 @@ func (r *RetentionReaper) ExportStart() {
} }
// ExportStop stops the reaper's background loop for tests. // ExportStop stops the reaper's background loop for tests.
func (r *RetentionReaper) ExportStop() { func (r *RetentionReaper) ExportStop(ctx context.Context) error {
r.stop() return r.stop(ctx)
}
// ExportWedgeLoop adds a goroutine to the reaper's WaitGroup that
// never observes cancellation and returns only when release is
// closed. It stands in for a sweep stuck on a locked database.
func (r *RetentionReaper) ExportWedgeLoop(
release <-chan struct{},
) {
r.wg.Go(func() {
<-release
})
} }
// ExportSetInterval overrides the sweep interval for tests. // ExportSetInterval overrides the sweep interval for tests.

View File

@@ -10,6 +10,7 @@ import (
"go.uber.org/fx" "go.uber.org/fx"
"gorm.io/gorm" "gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/config" "sneak.berlin/go/webhooker/internal/config"
"sneak.berlin/go/webhooker/internal/lifecycle"
"sneak.berlin/go/webhooker/internal/logger" "sneak.berlin/go/webhooker/internal/logger"
) )
@@ -62,8 +63,9 @@ func NewRetentionReaper(
} }
// registerHooks wires the reaper's start and stop into the fx // registerHooks wires the reaper's start and stop into the fx
// lifecycle. The start hook's context is deliberately ignored: see // lifecycle. The start hook's context is deliberately ignored (see
// start for why the sweep loop must not inherit it. // start for why the sweep loop must not inherit it); the stop hook's
// context is honoured (see stop).
func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) { func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
lc.Append(fx.Hook{ lc.Append(fx.Hook{
//nolint:contextcheck // Not inheriting the hook context is //nolint:contextcheck // Not inheriting the hook context is
@@ -73,10 +75,8 @@ func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(_ context.Context) error { OnStop: func(ctx context.Context) error {
r.stop() return r.stop(ctx)
return nil
}, },
}) })
} }
@@ -105,15 +105,27 @@ func (r *RetentionReaper) start() {
) )
} }
func (r *RetentionReaper) stop() { // stop cancels the sweep loop's context and waits for it to
// exit, bounded by the stop hook's context: a sweep wedged on a
// locked database must not hang the process past fx's stop
// timeout.
func (r *RetentionReaper) stop(ctx context.Context) error {
r.log.Info("retention reaper stopping") r.log.Info("retention reaper stopping")
if r.cancel != nil { if r.cancel != nil {
r.cancel() r.cancel()
} }
r.wg.Wait() err := lifecycle.WaitForShutdown(
ctx, r.log, "retention reaper", &r.wg,
)
if err != nil {
return err
}
r.log.Info("retention reaper stopped") r.log.Info("retention reaper stopped")
return nil
} }
func (r *RetentionReaper) run(ctx context.Context) { func (r *RetentionReaper) run(ctx context.Context) {

View File

@@ -26,6 +26,13 @@ const (
// reaperTestRetentionDays is the retention policy the lifecycle // reaperTestRetentionDays is the retention policy the lifecycle
// tests give their webhook. // tests give their webhook.
reaperTestRetentionDays = 30 reaperTestRetentionDays = 30
// reaperWedgeStopTimeout is the stop timeout the wedged-shutdown
// test hands OnStop, standing in for fx's StopTimeout. The test
// asserts only that the hook returns at all, and allows it
// reaperStopTimeout — forty times this budget — to do so, so no
// assertion races the wall clock.
reaperWedgeStopTimeout = 250 * time.Millisecond
) )
// recordingLifecycle is a minimal fx.Lifecycle that records the // recordingLifecycle is a minimal fx.Lifecycle that records the
@@ -207,3 +214,59 @@ func TestRetentionReaper_StopHookStopsLoop(t *testing.T) {
"a stopped reaper must not sweep anything", "a stopped reaper must not sweep anything",
) )
} }
// TestRetentionReaper_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 sweep blocked on a locked SQLite database
// — 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 TestRetentionReaper_StopHookHonoursStopTimeout(
t *testing.T,
) {
t.Parallel()
env := setupRetentionTest(t)
env.reaper.ExportSetInterval(reaperTestInterval)
lc := startReaperViaHook(t, env.reaper)
release := make(chan struct{})
t.Cleanup(func() { close(release) })
env.reaper.ExportWedgeLoop(release)
stopCtx, cancel := context.WithTimeout(
context.Background(), reaperWedgeStopTimeout,
)
defer cancel()
var stopErr error
stopped := make(chan struct{})
go func() {
defer close(stopped)
stopErr = lc.hooks[0].OnStop(stopCtx)
}()
select {
case <-stopped:
case <-time.After(reaperStopTimeout):
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, "retention reaper")
}

View File

@@ -10,6 +10,7 @@ import (
"go.uber.org/fx" "go.uber.org/fx"
"sneak.berlin/go/webhooker/internal/config" "sneak.berlin/go/webhooker/internal/config"
"sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/lifecycle"
"sneak.berlin/go/webhooker/internal/logger" "sneak.berlin/go/webhooker/internal/logger"
) )
@@ -67,10 +68,9 @@ func NewArchiveSweeper(
} }
// registerHooks wires the sweeper's start and stop into the fx // registerHooks wires the sweeper's start and stop into the fx
// lifecycle. Both hook contexts are deliberately ignored: see // lifecycle. The start hook's context is deliberately ignored
// start for why the background loop must not inherit the start // (see start for why the background loop must not inherit it);
// hook's context, and stop for why shutdown blocks on the loop // the stop hook's context is honoured (see stop).
// rather than on the stop hook's deadline.
func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) { func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
lc.Append(fx.Hook{ lc.Append(fx.Hook{
//nolint:contextcheck // Not passing the hook context is //nolint:contextcheck // Not passing the hook context is
@@ -80,10 +80,8 @@ func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(_ context.Context) error { OnStop: func(ctx context.Context) error {
s.stop() return s.stop(ctx)
return nil
}, },
}) })
} }
@@ -113,15 +111,27 @@ func (s *ArchiveSweeper) start() {
) )
} }
func (s *ArchiveSweeper) stop() { // stop cancels the sweep loop's context and waits for it to
// exit, bounded by the stop hook's context: a prune wedged on a
// locked archive must not hang the process past fx's stop
// timeout.
func (s *ArchiveSweeper) stop(ctx context.Context) error {
s.log.Info("archive sweeper stopping") s.log.Info("archive sweeper stopping")
if s.cancel != nil { if s.cancel != nil {
s.cancel() s.cancel()
} }
s.wg.Wait() err := lifecycle.WaitForShutdown(
ctx, s.log, "archive sweeper", &s.wg,
)
if err != nil {
return err
}
s.log.Info("archive sweeper stopped") s.log.Info("archive sweeper stopped")
return nil
} }
func (s *ArchiveSweeper) run(ctx context.Context) { func (s *ArchiveSweeper) run(ctx context.Context) {

View File

@@ -14,7 +14,6 @@ import (
"github.com/google/uuid" "github.com/google/uuid"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"go.uber.org/fx"
"gorm.io/driver/sqlite" "gorm.io/driver/sqlite"
"gorm.io/gorm" "gorm.io/gorm"
"gorm.io/gorm/clause" "gorm.io/gorm/clause"
@@ -226,17 +225,6 @@ func countArchivedRows(path string) (int64, error) {
return count, nil return count, nil
} }
// captureLifecycle 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 captureLifecycle struct {
hooks []fx.Hook
}
func (l *captureLifecycle) Append(h fx.Hook) {
l.hooks = append(l.hooks, h)
}
// TestArchiveSweeper_LoopOutlivesStartHookContext is the // TestArchiveSweeper_LoopOutlivesStartHookContext is the
// regression test for a sweeper that never swept. fx calls // regression test for a sweeper that never swept. fx calls
// OnStart with a context carrying the application's start // OnStart with a context carrying the application's start
@@ -270,7 +258,7 @@ func TestArchiveSweeper_LoopOutlivesStartHookContext(
// Drive the genuine fx hooks the application registers, // Drive the genuine fx hooks the application registers,
// rather than a test-only entry point. // rather than a test-only entry point.
lc := &captureLifecycle{} lc := &recordingLifecycle{}
env.sweeper.ExportRegisterHooks(lc) env.sweeper.ExportRegisterHooks(lc)
require.Len(t, lc.hooks, 1) require.Len(t, lc.hooks, 1)
@@ -924,7 +912,36 @@ func TestArchiveSweeper_StopsCleanly(t *testing.T) {
env.sweeper.ExportSetInterval(time.Millisecond) env.sweeper.ExportSetInterval(time.Millisecond)
env.sweeper.ExportStart() env.sweeper.ExportStart()
// stop blocks on the loop's WaitGroup, so returning at all // stop blocks on the loop's WaitGroup, so returning without
// proves the loop observed the cancellation and exited. // error proves the loop observed the cancellation and exited
env.sweeper.ExportStop() // well inside the stop context.
require.NoError(
t, env.sweeper.ExportStop(context.Background()),
)
}
// TestArchiveSweeper_StopHookHonoursStopTimeout is the sweeper's
// half of the same shutdown defect the engine and the retention
// reaper carried: an OnStop that discards its context and waits
// on the WaitGroup bare hangs the process forever on a prune
// wedged inside a locked archive.
func TestArchiveSweeper_StopHookHonoursStopTimeout(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
lc := &recordingLifecycle{}
env.sweeper.ExportRegisterHooks(lc)
require.Len(t, lc.hooks, 1)
require.NoError(t, lc.hooks[0].OnStart(context.Background()))
release := make(chan struct{})
t.Cleanup(func() { close(release) })
env.sweeper.ExportWedgeLoop(release)
requireStopHookExpires(t, lc.hooks[0], "archive sweeper")
} }

View File

@@ -13,6 +13,7 @@ import (
"go.uber.org/fx" "go.uber.org/fx"
"gorm.io/gorm" "gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/lifecycle"
"sneak.berlin/go/webhooker/internal/logger" "sneak.berlin/go/webhooker/internal/logger"
) )
@@ -234,8 +235,9 @@ func (e *Engine) ScheduleRetry(
} }
// registerHooks wires the engine's start and stop into the fx // registerHooks wires the engine's start and stop into the fx
// lifecycle. The start hook's context is deliberately ignored: // lifecycle. The start hook's context is deliberately ignored
// see start for why the worker pool must not inherit it. // (see start for why the worker pool must not inherit it); the
// stop hook's context is honoured (see stop).
func (e *Engine) registerHooks(lc fx.Lifecycle) { func (e *Engine) registerHooks(lc fx.Lifecycle) {
lc.Append(fx.Hook{ lc.Append(fx.Hook{
//nolint:contextcheck // Not inheriting the hook context //nolint:contextcheck // Not inheriting the hook context
@@ -245,10 +247,8 @@ func (e *Engine) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(_ context.Context) error { OnStop: func(ctx context.Context) error {
e.stop() return e.stop(ctx)
return nil
}, },
}) })
} }
@@ -289,11 +289,26 @@ func (e *Engine) start() {
) )
} }
func (e *Engine) stop() { // stop cancels the worker pool's context and waits for the pool
// to drain, bounded by the stop hook's context: a wedged worker
// must not hang the process past fx's stop timeout.
func (e *Engine) stop(ctx context.Context) error {
e.log.Info("delivery engine stopping") e.log.Info("delivery engine stopping")
e.cancel()
e.wg.Wait() if e.cancel != nil {
e.cancel()
}
err := lifecycle.WaitForShutdown(
ctx, e.log, "delivery engine", &e.wg,
)
if err != nil {
return err
}
e.log.Info("delivery engine stopped") e.log.Info("delivery engine stopped")
return nil
} }
func (e *Engine) worker(ctx context.Context) { func (e *Engine) worker(ctx context.Context) {

View File

@@ -501,7 +501,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
iWaitForDelivered(t, s.WebhookDB, d.ID) iWaitForDelivered(t, s.WebhookDB, d.ID)
s.Engine.ExportStop() require.NoError(t, s.Engine.ExportStop(context.Background()))
} }
// iWaitForDelivered polls until the delivery reaches the // iWaitForDelivered polls until the delivery reaches the
@@ -567,7 +567,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
iWaitForDelivered(t, s.WebhookDB, d.ID) iWaitForDelivered(t, s.WebhookDB, d.ID)
s.Engine.ExportStop() require.NoError(t, s.Engine.ExportStop(context.Background()))
} }
// --- processDelivery: unknown target type --- // --- processDelivery: unknown target type ---

View File

@@ -27,6 +27,13 @@ const (
// and a ready deliveryCh are chosen between at random and a // and a ready deliveryCh are chosen between at random and a
// doomed pool still delivers. // doomed pool still delivers.
hookSettleDelay = 250 * time.Millisecond 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 // recordingLifecycle is a minimal fx.Lifecycle that records the
@@ -40,6 +47,44 @@ func (l *recordingLifecycle) Append(h fx.Hook) {
l.hooks = append(l.hooks, h) 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 // startEngineViaHook drives the genuine fx hooks the application
// registers for the engine, handing OnStart a context that is // registers for the engine, handing OnStart a context that is
// already done, and returns only once a pool that inherited that // already done, and returns only once a pool that inherited that
@@ -197,3 +242,30 @@ func TestEngine_StopHookStopsWorkers(t *testing.T) {
"a stopped engine must not deliver anything", "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")
}

View File

@@ -216,8 +216,19 @@ func (e *Engine) ExportRegisterHooks(lc fx.Lifecycle) {
} }
// ExportStop exposes stop for testing. // ExportStop exposes stop for testing.
func (e *Engine) ExportStop() { func (e *Engine) ExportStop(ctx context.Context) error {
e.stop() return e.stop(ctx)
}
// ExportWedgeWorker adds a goroutine to the engine's WaitGroup
// that never observes cancellation and returns only when release
// is closed. It stands in for a worker stuck inside a delivery
// target that never returns, which is the only way stop can be
// made to outlast its context.
func (e *Engine) ExportWedgeWorker(release <-chan struct{}) {
e.wg.Go(func() {
<-release
})
} }
// ExportDeliveryCh returns the delivery channel. // ExportDeliveryCh returns the delivery channel.
@@ -518,8 +529,19 @@ func (s *ArchiveSweeper) ExportRegisterHooks(lc fx.Lifecycle) {
} }
// ExportStop stops the sweeper's background loop for tests. // ExportStop stops the sweeper's background loop for tests.
func (s *ArchiveSweeper) ExportStop() { func (s *ArchiveSweeper) ExportStop(ctx context.Context) error {
s.stop() return s.stop(ctx)
}
// ExportWedgeLoop adds a goroutine to the sweeper's WaitGroup
// that never observes cancellation and returns only when release
// is closed. It stands in for a prune stuck on a locked archive.
func (s *ArchiveSweeper) ExportWedgeLoop(
release <-chan struct{},
) {
s.wg.Go(func() {
<-release
})
} }
// ExportSetInterval overrides the sweep interval for tests. // ExportSetInterval overrides the sweep interval for tests.

View File

@@ -106,12 +106,6 @@ func slackConfigFields(configJSON string) []ConfigField {
// and its retry settings. Header values are not shown — they // and its retry settings. Header values are not shown — they
// routinely carry authorization tokens — only how many are // routinely carry authorization tokens — only how many are
// configured. // configured.
//
// The destination is masked to scheme and host by the same
// rule the Slack target uses. An HTTP target's destination is
// commonly a Slack, Discord or Teams incoming-webhook endpoint
// whose path segments are the credential, and the field takes
// an arbitrary URL, so no segment can be assumed non-secret.
func httpConfigFields(t *database.Target) []ConfigField { func httpConfigFields(t *database.Target) []ConfigField {
cfg, err := parseHTTPConfig(t.Config) cfg, err := parseHTTPConfig(t.Config)
if err != nil { if err != nil {
@@ -120,7 +114,7 @@ func httpConfigFields(t *database.Target) []ConfigField {
fields := []ConfigField{{ fields := []ConfigField{{
Label: "Destination URL", Label: "Destination URL",
Value: MaskURL(cfg.URL), Value: cfg.URL,
}} }}
if cfg.Timeout > 0 { if cfg.Timeout > 0 {

View File

@@ -19,7 +19,6 @@ const (
viewExampleOrigin = "https://example.com" viewExampleOrigin = "https://example.com"
viewExampleHook = viewExampleOrigin + "/hook" viewExampleHook = viewExampleOrigin + "/hook"
viewMaskedOrigin = viewExampleOrigin + "/..."
viewUnavailable = "(unavailable)" viewUnavailable = "(unavailable)"
viewExpiryNever = "never" viewExpiryNever = "never"
) )
@@ -163,7 +162,7 @@ func TestNewTargetViews_HTTP(t *testing.T) {
assert.Equal( assert.Equal(
t, t,
map[string]string{ map[string]string{
"Destination URL": viewMaskedOrigin, "Destination URL": viewExampleHook,
"Timeout": "30s", "Timeout": "30s",
"Headers": "1 configured", "Headers": "1 configured",
"Max Retries": "5", "Max Retries": "5",
@@ -189,41 +188,13 @@ func TestNewTargetViews_HTTPFireAndForget(t *testing.T) {
assert.Equal( assert.Equal(
t, t,
map[string]string{ map[string]string{
"Destination URL": viewMaskedOrigin, "Destination URL": viewExampleHook,
"Max Retries": "0 (fire-and-forget)", "Max Retries": "0 (fire-and-forget)",
}, },
fieldMap(view.Config), fieldMap(view.Config),
) )
} }
// TestNewTargetViews_HTTPMasksDestinationURL proves the rule
// holds for the http target too: an http destination is
// routinely an incoming-webhook endpoint whose path segments
// are the credential, so none of them is shown.
func TestNewTargetViews_HTTPMasksDestinationURL(t *testing.T) {
t.Parallel()
view := viewFor(t, database.Target{
Type: database.TargetTypeHTTP,
Config: `{"url":"` + slackWebhookURL + `"}`,
})
fields := fieldMap(view.Config)
assert.Equal(
t,
"https://hooks.slack.com/...",
fields["Destination URL"],
)
for _, v := range fields {
assert.NotContains(t, v, slackSecretPath)
assert.NotContains(t, v, "T00000000")
assert.NotContains(t, v, "B00000000")
assert.NotContains(t, v, "XXXXXXXXXXXXXXXXXXXXXXXX")
}
}
func TestNewTargetViews_Database(t *testing.T) { func TestNewTargetViews_Database(t *testing.T) {
t.Parallel() t.Parallel()

View File

@@ -131,47 +131,6 @@ func TestHandleSourceDetail_MasksSlackWebhookURL(t *testing.T) {
assert.Contains(t, body, "https://hooks.slack.com/...") assert.Contains(t, body, "https://hooks.slack.com/...")
} }
// TestHandleSourceDetail_MasksHTTPDestinationURL is the
// regression test for the same leak reached through the http
// target: its destination is routinely an incoming-webhook
// endpoint whose path segments are the credential, so the
// rendered page must not contain them.
func TestHandleSourceDetail_MasksHTTPDestinationURL(
t *testing.T,
) {
t.Parallel()
var (
h *handlers.Handlers
sess *session.Session
db *database.Database
)
app := newTestApp(t, &h, &sess, &db)
app.RequireStart()
t.Cleanup(app.RequireStop)
wh := seedWebhook(t, db)
seedConfiguredTarget(
t, db, wh.ID,
database.TargetTypeHTTP,
`{"url":"`+slackWebhookURL+`"}`,
)
body := renderSourceDetailPage(t, h, sess, wh.ID)
assert.NotContains(t, body, slackSecretPath)
assert.NotContains(t, body, "T00000000")
assert.NotContains(t, body, "B00000000")
assert.NotContains(
t, body, "XXXXXXXXXXXXXXXXXXXXXXXX",
)
assert.Contains(t, body, "Destination URL")
assert.Contains(t, body, "https://hooks.slack.com/...")
}
// TestHandleSourceDetail_RendersNamedTargetFields proves the // TestHandleSourceDetail_RendersNamedTargetFields proves the
// other target types render labelled fields rather than the // other target types render labelled fields rather than the
// stored blob. // stored blob.
@@ -213,7 +172,7 @@ func TestHandleSourceDetail_RendersNamedTargetFields(
body := renderSourceDetailPage(t, h, sess, wh.ID) body := renderSourceDetailPage(t, h, sess, wh.ID)
assert.Contains(t, body, "Destination URL") assert.Contains(t, body, "Destination URL")
assert.Contains(t, body, "https://example.com/...") assert.Contains(t, body, "https://example.com/hook")
assert.Contains(t, body, "Timeout") assert.Contains(t, body, "Timeout")
assert.Contains(t, body, "1 configured") assert.Contains(t, body, "1 configured")
assert.NotContains(t, body, "sekrit") assert.NotContains(t, body, "sekrit")

View File

@@ -0,0 +1,57 @@
// Package lifecycle holds helpers shared by the components that
// register fx start and stop hooks.
package lifecycle
import (
"context"
"fmt"
"log/slog"
"sync"
)
// WaitForShutdown waits for wg to drain, bounded by ctx.
//
// fx hands OnStop a context carrying the application's stop
// timeout. A bare wg.Wait() discards that deadline, so a single
// goroutine that never observes cancellation — a delivery target
// that never returns, a SQLite operation blocked on a lock —
// hangs the process forever instead of letting it exit when the
// timeout expires, which is exactly when a clean shutdown matters
// most.
//
// On timeout it logs at error naming component and returns an
// error: the goroutines are still running, and reporting success
// would hide an unclean shutdown from the operator. The waiting
// goroutine outlives this call and exits when (if) wg drains; it
// holds nothing but the channel it closes.
func WaitForShutdown(
ctx context.Context,
log *slog.Logger,
component string,
wg *sync.WaitGroup,
) error {
done := make(chan struct{})
go func() {
defer close(done)
wg.Wait()
}()
select {
case <-done:
return nil
case <-ctx.Done():
log.Error(
"shutdown timed out, goroutines still running",
"component", component,
"error", ctx.Err(),
)
return fmt.Errorf(
"%s: shutdown timed out, "+
"goroutines still running: %w",
component, ctx.Err(),
)
}
}

View File

@@ -0,0 +1,62 @@
package lifecycle_test
import (
"context"
"log/slog"
"sync"
"testing"
"time"
"github.com/stretchr/testify/require"
"sneak.berlin/go/webhooker/internal/lifecycle"
)
// waitTimeout is the stop budget the timeout case gives a
// goroutine that never returns. The test's own patience is the
// go test deadline, so the only thing this value affects is how
// long the case takes.
const waitTimeout = 100 * time.Millisecond
func discardLogger() *slog.Logger {
return slog.New(slog.DiscardHandler)
}
func TestWaitForShutdown_DrainedGroup(t *testing.T) {
t.Parallel()
var wg sync.WaitGroup
wg.Go(func() {})
require.NoError(
t,
lifecycle.WaitForShutdown(
context.Background(), discardLogger(),
"test component", &wg,
),
)
}
func TestWaitForShutdown_ContextExpires(t *testing.T) {
t.Parallel()
release := make(chan struct{})
t.Cleanup(func() { close(release) })
var wg sync.WaitGroup
wg.Go(func() { <-release })
ctx, cancel := context.WithTimeout(
context.Background(), waitTimeout,
)
defer cancel()
err := lifecycle.WaitForShutdown(
ctx, discardLogger(), "test component", &wg,
)
require.ErrorIs(t, err, context.DeadlineExceeded)
require.ErrorContains(t, err, "test component")
}