Compare commits
1 Commits
issue-97-l
...
issue-82-o
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0384de4a7b |
12
README.md
12
README.md
@@ -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
10
TODO.md
@@ -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
|
||||||
|
|||||||
@@ -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
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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",
|
|
||||||
)
|
|
||||||
}
|
|
||||||
@@ -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(
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -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",
|
|
||||||
)
|
|
||||||
}
|
|
||||||
@@ -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.
|
||||||
|
|||||||
Reference in New Issue
Block a user