clawbot
ae06f7e3a1
check / check (push) Successful in 1m22s
Shutdown waits for deliveries already in flight instead of dropping them. Reviewed and green; squashed to next by the dispatcher, dnswatcher having no manager. Model: opus-5
120 lines
3.3 KiB
Go
120 lines
3.3 KiB
Go
package notify
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
)
|
|
|
|
// ErrDeliveryAbandoned is returned by a retry loop that was
|
|
// cut short because shutdown drained past its deadline.
|
|
var ErrDeliveryAbandoned = errors.New(
|
|
"notification delivery abandoned at shutdown",
|
|
)
|
|
|
|
// startDelivery runs fn on its own goroutine while tracking it,
|
|
// so that drain can wait for it during shutdown.
|
|
//
|
|
// The WaitGroup counter is incremented here, on the caller's
|
|
// goroutine, before the worker exists: incrementing it inside
|
|
// the worker would race with drain's Wait and could let
|
|
// shutdown sail past a delivery that had not started yet.
|
|
//
|
|
// Once draining has begun the delivery is refused outright
|
|
// rather than queued, so a steady stream of newly submitted
|
|
// notifications cannot keep extending the drain.
|
|
func (svc *Service) startDelivery(endpoint string, fn func()) {
|
|
svc.drainMu.Lock()
|
|
|
|
if svc.draining {
|
|
svc.drainMu.Unlock()
|
|
|
|
svc.log.Warn(
|
|
"notification not dispatched: shutdown in progress",
|
|
"endpoint", endpoint,
|
|
)
|
|
|
|
return
|
|
}
|
|
|
|
svc.outstanding.Add(1)
|
|
|
|
// WaitGroup.Go increments the counter synchronously, here,
|
|
// and only then starts the goroutine.
|
|
svc.inFlight.Go(func() {
|
|
// Runs before the WaitGroup counter is decremented, so
|
|
// a drain that times out reports an accurate count.
|
|
defer svc.outstanding.Add(-1)
|
|
|
|
fn()
|
|
})
|
|
|
|
svc.drainMu.Unlock()
|
|
}
|
|
|
|
// drain waits for in-flight notification deliveries to finish.
|
|
//
|
|
// It first stops accepting new deliveries, then waits until
|
|
// either every outstanding delivery has completed or ctx
|
|
// expires — whichever comes first. ctx is the context fx
|
|
// passes to the OnStop hook, so a permanently dead webhook
|
|
// cannot hang shutdown indefinitely.
|
|
//
|
|
// When the deadline arrives with deliveries still outstanding,
|
|
// the count is logged at warn level and the abandon channel is
|
|
// 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
|
|
svc.drainMu.Unlock()
|
|
|
|
done := make(chan struct{})
|
|
|
|
go func() {
|
|
svc.inFlight.Wait()
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
svc.log.Debug(
|
|
"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)
|
|
}
|
|
})
|
|
|
|
svc.log.Warn(
|
|
"shutdown deadline reached with notifications "+
|
|
"still in flight; abandoning them",
|
|
"abandoned", abandoned,
|
|
"error", ctx.Err(),
|
|
)
|
|
}
|
|
}
|