notify: drain in-flight deliveries at shutdown (closes #106) #113

Open
clawbot wants to merge 2 commits from fix/106-notify-shutdown-drain into next
3 changed files with 115 additions and 19 deletions
Showing only changes of commit cd06bba034 - Show all commits

View File

@@ -31,7 +31,9 @@ confirm make check still passes.
`OnStop` context; on expiry the outstanding count is logged at warn
level and parked retry backoffs are released instead of being dropped
silently, and deliveries submitted after the drain begins are refused
so shutdown cannot be extended indefinitely
so shutdown cannot be extended indefinitely; an `OnStop` context that
is already expired on entry with nothing outstanding drains quietly
rather than warning about deliveries that were never abandoned
- 2026-08-07: golangci-lint bumped to v2.12.2 (commit-pinned installs
in `Dockerfile` and `script/bootstrap`); `.golangci.yml` set to the
org-standard v2-schema config used across the org's repos

View File

@@ -64,6 +64,11 @@ func (svc *Service) startDelivery(endpoint string, fn func()) {
// closed, which releases any retry loop sleeping in backoff.
// Deliveries already inside an HTTP round trip are bounded by
// the existing httpClientTimeout instead.
//
// A ctx that is already expired on entry is not by itself cause
// for alarm: if nothing is outstanding there is nothing to
// abandon, and the drain says so at debug level rather than
// warning about deliveries that do not exist.
func (svc *Service) drain(ctx context.Context) {
svc.drainMu.Lock()
svc.draining = true
@@ -82,6 +87,22 @@ func (svc *Service) drain(ctx context.Context) {
"all in-flight notifications completed",
)
case <-ctx.Done():
// outstanding is decremented before the WaitGroup
// counter, and startDelivery can no longer add to it
// now that draining is set, so a zero here means every
// delivery really did finish. ctx expiring in that
// state (an OnStop context that was already cancelled
// on entry is the usual way) abandons nothing, so it
// must not close abandon or warn about it.
abandoned := svc.outstanding.Load()
if abandoned == 0 {
svc.log.Debug(
"all in-flight notifications completed",
)
return
}
svc.abandonOnce.Do(func() {
if svc.abandon != nil {
close(svc.abandon)
@@ -91,7 +112,7 @@ func (svc *Service) drain(ctx context.Context) {
svc.log.Warn(
"shutdown deadline reached with notifications "+
"still in flight; abandoning them",
"abandoned", svc.outstanding.Load(),
"abandoned", abandoned,
"error", ctx.Err(),
)
}

View File

@@ -41,6 +41,16 @@ const (
// settleDelay is how long to wait before asserting that
// something did *not* happen.
settleDelay = 50 * time.Millisecond
// idleDrainBound is the upper bound on a drain that has
// nothing in flight. It is deliberately far above the cost
// of the goroutine hop through inFlight.Wait() — which
// reached 57ms on a loaded box under -race with the package's
// parallel tests — and far below drainSlack, the deadline
// such a drain is given. A drain that blocked until its
// deadline instead of returning on the WaitGroup therefore
// still fails this bound, but scheduling delay alone cannot.
idleDrainBound = 500 * time.Millisecond
)
// syncBuffer is an io.Writer safe for concurrent use, so log
@@ -128,6 +138,13 @@ func TestDrainWaitsForInFlightDelivery(t *testing.T) {
t.Fatal("delivery never reached the endpoint")
}
// As in TestDrainBoundedByContextDeadline: start is captured
// before the clock it is compared against, here the timer
// holding the delivery open, so elapsed covers the whole hold
// and the lower bound cannot come out short from scheduling
// delay alone.
start := time.Now()
timer := time.AfterFunc(inFlightHold, func() {
close(release)
})
@@ -138,8 +155,6 @@ func TestDrainWaitsForInFlightDelivery(t *testing.T) {
)
defer cancel()
start := time.Now()
svc.Drain(ctx)
elapsed := time.Since(start)
@@ -211,28 +226,53 @@ func TestDrainBoundedByContextDeadline(t *testing.T) {
svc.OutstandingDeliveries() == 1
})
// start must be captured *before* the deadline clock starts,
// so that the measured interval is a superset of the deadline
// interval. Capturing it after context.WithTimeout would
// make elapsed structurally smaller than drainDeadline and
// the lower bound below unfalsifiable-by-luck: it would fail
// whenever the two statements were separated by any
// scheduling delay, and pass otherwise, regardless of what
// the drain did.
start := time.Now()
ctx, cancel := context.WithTimeout(
context.Background(), drainDeadline,
)
defer cancel()
start := time.Now()
// The upper bound is enforced by a watchdog rather than by
// measuring after the fact: a drain that is not bounded at
// all never returns here (the delivery is parked in a backoff
// that never fires), so an unbounded drain must fail this
// test promptly instead of hanging the package until the test
// binary's 30s timeout.
returned := make(chan struct{})
go func() {
defer close(returned)
svc.Drain(ctx)
}()
elapsed := time.Since(start)
if elapsed < drainDeadline {
t.Errorf(
"drain returned after %v, before its %v deadline",
elapsed, drainDeadline,
select {
case <-returned:
case <-time.After(drainSlack):
t.Fatalf(
"drain did not return within %v; its %v deadline "+
"did not bound it",
drainSlack, drainDeadline,
)
}
if elapsed > drainSlack {
// The lower bound is the real assertion: the drain must have
// waited for its whole deadline rather than giving up on the
// outstanding delivery early. With start captured above, an
// early return is the only thing that can make it fail.
if elapsed := time.Since(start); elapsed < drainDeadline {
t.Errorf(
"drain took %v, want it bounded well under %v",
elapsed, drainSlack,
"drain returned after %v, before its %v deadline",
elapsed, drainDeadline,
)
}
@@ -440,19 +480,52 @@ func TestDrainWithoutDeliveriesReturnsImmediately(t *testing.T) {
svc := notify.NewTestService(http.DefaultTransport)
// Captured before the deadline clock, as elsewhere in this
// file; for an upper bound that is the conservative
// direction, since the measured interval can then only be
// longer than the drain itself.
start := time.Now()
ctx, cancel := context.WithTimeout(
context.Background(), drainSlack,
)
defer cancel()
start := time.Now()
svc.Drain(ctx)
if elapsed := time.Since(start); elapsed > settleDelay {
if elapsed := time.Since(start); elapsed > idleDrainBound {
t.Errorf(
"drain of an idle service took %v, want ~0",
elapsed,
"drain of an idle service took %v, want well "+
"under its %v deadline",
elapsed, drainSlack,
)
}
}
// TestDrainWithCancelledContextDoesNotWarn verifies that an
// OnStop context that is already dead on entry does not produce
// an "abandoning them" warning when there was nothing in flight
// to abandon. The expired context wins the select immediately,
// so only the outstanding count can tell the difference between
// a genuine timeout and a shutdown that had simply already run
// out of time with no work left.
func TestDrainWithCancelledContextDoesNotWarn(t *testing.T) {
t.Parallel()
svc, logs := newLoggingService(http.DefaultTransport)
ctx, cancel := context.WithCancel(context.Background())
cancel()
svc.Drain(ctx)
if output := logs.String(); strings.Contains(
output, `"level":"WARN"`,
) {
t.Errorf(
"drain with nothing in flight warned about "+
"abandoned deliveries; log output: %s",
output,
)
}
}