Compare commits
2 Commits
55edebaea1
...
2a65d86245
| Author | SHA1 | Date | |
|---|---|---|---|
| 2a65d86245 | |||
| c3b6623be1 |
85
README.md
85
README.md
@@ -1001,7 +1001,14 @@ Every limiter here — receiver, login, and password change — identifies
|
||||
the client the same way, through one shared key function: the
|
||||
connection's own address, unless the peer is listed in
|
||||
`TRUSTED_PROXIES`, in which case the forwarded client address is used
|
||||
instead. See [Trusted proxies](#trusted-proxies). Deployed without that
|
||||
instead. That address becomes a bucket by family: IPv4 keys on the full
|
||||
address, IPv6 on its `/64` prefix. A routed `/64` is the normal
|
||||
residential and mobile IPv6 allocation, so keying IPv6 per address would
|
||||
let one subscriber rotate source addresses and mint a fresh bucket per
|
||||
request, evading these limits at the network layer without spoofing
|
||||
anything; the cost is that distinct clients inside one `/64` share a
|
||||
bucket. IPv4-mapped addresses (`::ffff:1.2.3.4`) key as the IPv4 address
|
||||
they carry. See [Trusted proxies](#trusted-proxies). Deployed without that
|
||||
variable set, a client behind a reverse proxy shares one bucket with
|
||||
every other client behind the same proxy. Set `TRUSTED_PROXIES` to the
|
||||
proxy's address to get per-client limits back. What the shared bucket
|
||||
@@ -1138,8 +1145,6 @@ webhooker/
|
||||
│ │ ├── archive_sweeper.go # Periodic pruning of idle archives
|
||||
│ │ ├── url_mask.go # Strips credentials from *url.Error
|
||||
│ │ └── ssrf.go # SSRF prevention (IP validation, safe HTTP transport)
|
||||
│ ├── lifecycle/
|
||||
│ │ └── lifecycle.go # Shared fx start/stop hook helpers
|
||||
│ ├── handlers/
|
||||
│ │ ├── handlers.go # Base handler struct, JSON helpers, template rendering
|
||||
│ │ ├── auth.go # Login, logout handlers
|
||||
@@ -1151,6 +1156,8 @@ webhooker/
|
||||
│ │ └── webhook.go # Webhook receiver handler
|
||||
│ ├── healthcheck/
|
||||
│ │ └── healthcheck.go # Health check service (uptime, version)
|
||||
│ ├── lifecycle/
|
||||
│ │ └── lifecycle.go # Shared stop-hook waiter, bounded by the stop context
|
||||
│ ├── logger/
|
||||
│ │ └── logger.go # slog setup with TTY detection
|
||||
│ ├── middleware/
|
||||
@@ -1309,6 +1316,78 @@ rather than global: **LoginRateLimit** on `/pages/login`,
|
||||
- GORM soft deletes on every entity that carries `BaseModel`, which is
|
||||
all of them but `Setting` (data preserved for audit)
|
||||
|
||||
### Shutdown
|
||||
|
||||
On SIGINT or SIGTERM, fx runs the registered stop hooks in reverse
|
||||
dependency order under a **5 second budget** (`fx.StopTimeout` in
|
||||
`cmd/webhooker/main.go`). That budget covers the whole sequence, not
|
||||
each hook. The order, read off the fx stop-hook log:
|
||||
|
||||
1. `ArchiveSweeper`
|
||||
2. `RetentionReaper`
|
||||
3. `server` — the HTTP drain, bounded separately by
|
||||
`server.ShutdownTimeout` (**3 seconds**), then a Sentry flush if
|
||||
`SENTRY_DSN` is set
|
||||
4. `delivery.Engine`
|
||||
5. `healthcheck`
|
||||
6. `WebhookDBManager`
|
||||
7. the database close
|
||||
|
||||
The two components that can realistically hold the budget run
|
||||
first: a retention sweep or an archive prune caught mid-tick each
|
||||
waits on its `WaitGroup` bounded by the stop context, so a wedge
|
||||
there consumes the 5 seconds before the HTTP server hook is ever
|
||||
entered. The hooks after the server are microsecond-scale in normal
|
||||
operation.
|
||||
|
||||
The HTTP drain budget is deliberately **shorter** than the sequence
|
||||
budget. Were the two equal, a drain that used its whole budget would
|
||||
exhaust the sequence budget at the instant it finished, and every
|
||||
later hook — the delivery engine, the healthcheck, the webhook DB
|
||||
manager and the database close — would be skipped in exactly the
|
||||
case where the drain mattered. 3 seconds leaves 2 seconds
|
||||
(`server.TailHookReserve`) for the tail, which is far more than the
|
||||
microseconds it needs.
|
||||
|
||||
That reserve belongs to the tail hooks, not to the server hook, and
|
||||
the Sentry flush is what could take it: it runs after the drain
|
||||
**inside the same hook**, and `sentry.Flush` takes a bare duration
|
||||
and honours no context, so an unreachable Sentry endpoint would add
|
||||
its own timeout on top of a full-length drain and consume the whole
|
||||
sequence budget by itself. It is therefore clamped to whatever is
|
||||
left on the stop context minus the reserve, and skipped when that
|
||||
leaves too little to be worth attempting — so a full-length drain
|
||||
means Sentry events are dropped rather than the database close being
|
||||
skipped.
|
||||
|
||||
This does not make the database close unconditional: a wedged
|
||||
`ArchiveSweeper` or `RetentionReaper` still runs first and can
|
||||
consume the whole budget on its own.
|
||||
|
||||
The value is chosen to sit inside the container stop grace period.
|
||||
Docker's default `docker stop` grace is 10 seconds and the Dockerfile
|
||||
sets no `STOPSIGNAL` or grace override, so the process must be gone
|
||||
before that. fx's own default is 15 seconds, which is past the grace:
|
||||
the container would be SIGKILLed (exit 137) before the bound could
|
||||
fire, and nothing that depends on it — including the
|
||||
`shutdown timed out, goroutines still running` error log that tells
|
||||
an operator a component is wedged — would ever be reached.
|
||||
|
||||
Two operational consequences follow from bounding the sequence:
|
||||
|
||||
- **A wedged component aborts the rest of the shutdown.** fx checks
|
||||
the stop context before each remaining hook and returns outright
|
||||
once it has expired, skipping the hooks it has not reached. If the
|
||||
first-stopped component consumes the whole budget, the later hooks
|
||||
never run — **the database close among them**. SQLite is crash-safe,
|
||||
so this is not corruption, but it is not a clean close either.
|
||||
- **Lowering the grace below 5 seconds reintroduces the silent
|
||||
truncation.** `docker stop --time`, Compose's `stop_grace_period`,
|
||||
or Kubernetes' `terminationGracePeriodSeconds` set under 5 seconds
|
||||
put SIGKILL back in front of the bound, and the process dies with
|
||||
no shutdown diagnostics at all. Keep the deployment's grace above
|
||||
the stop timeout.
|
||||
|
||||
### Docker
|
||||
|
||||
The Dockerfile uses a three-stage build. Each stage is pinned by
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
@@ -15,6 +17,33 @@ import (
|
||||
"sneak.berlin/go/webhooker/internal/session"
|
||||
)
|
||||
|
||||
// stopTimeout bounds the whole fx stop sequence, not each hook.
|
||||
//
|
||||
// fx defaults to 15s, which is longer than Docker's 10s default
|
||||
// stop grace: the container would be SIGKILLed before the bound
|
||||
// could fire, so nothing bounded by it would ever be observed.
|
||||
// 5s leaves headroom inside that grace for signal delivery and
|
||||
// process exit; the observed wedge case already exits at ~5.3s,
|
||||
// so a larger bound would trade a rare skipped database close for
|
||||
// a more common hard kill.
|
||||
//
|
||||
// The server's stop hook must fit inside it with room to spare: a
|
||||
// hook that used the whole budget would exhaust it at that instant,
|
||||
// and fx would skip every hook after the server — the delivery
|
||||
// engine, the healthcheck, the webhook DB manager and the database
|
||||
// close. That hook is the 3s HTTP drain plus the Sentry flush that
|
||||
// follows it in the same hook, so the flush is clamped to the stop
|
||||
// context's remaining time less server.TailHookReserve rather than
|
||||
// running for its own fixed 2s; the reserve is what the tail hooks
|
||||
// live on, and they are microsecond-scale in normal operation.
|
||||
// TestStopTimeout_LeavesHeadroomForTailHooks pins the arithmetic
|
||||
// across every drain length.
|
||||
//
|
||||
// This does not make the database close unconditional: the
|
||||
// ArchiveSweeper and RetentionReaper hooks run before the server
|
||||
// and can still consume the whole budget on their own.
|
||||
const stopTimeout = 5 * time.Second
|
||||
|
||||
// Build-time variables set via -ldflags.
|
||||
//
|
||||
//nolint:gochecknoglobals // Build-time variables injected by the linker.
|
||||
@@ -27,7 +56,14 @@ func main() {
|
||||
globals.Appname = appname
|
||||
globals.Version = version
|
||||
|
||||
fx.New(
|
||||
newApp().Run()
|
||||
}
|
||||
|
||||
// newApp builds the application graph. It is separate from main so
|
||||
// a test can assert the options it carries.
|
||||
func newApp() *fx.App {
|
||||
return fx.New(
|
||||
fx.StopTimeout(stopTimeout),
|
||||
fx.Provide(
|
||||
globals.New,
|
||||
logger.New,
|
||||
@@ -60,5 +96,5 @@ func main() {
|
||||
) {
|
||||
},
|
||||
),
|
||||
).Run()
|
||||
)
|
||||
}
|
||||
|
||||
75
cmd/webhooker/main_test.go
Normal file
75
cmd/webhooker/main_test.go
Normal file
@@ -0,0 +1,75 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/webhooker/internal/server"
|
||||
)
|
||||
|
||||
// dockerStopGrace is Docker's default `docker stop` grace period.
|
||||
// The Dockerfile sets no STOPSIGNAL or grace override, so this is
|
||||
// the deadline the container is actually held to, and the fx stop
|
||||
// timeout has to fit inside it with room for signal delivery and
|
||||
// process exit.
|
||||
const dockerStopGrace = 10 * time.Second
|
||||
|
||||
// TestNewApp_StopTimeout pins the fx stop timeout. Without the
|
||||
// explicit fx.StopTimeout option the app reads fx's 15s
|
||||
// DefaultTimeout, which exceeds dockerStopGrace: the container is
|
||||
// SIGKILLed before the bound fires and every shutdown hook bounded
|
||||
// by it — including the operator-facing timeout log — becomes
|
||||
// unreachable in the image this repo produces.
|
||||
//
|
||||
// fx.New applies options before it executes invokes, so the timeout
|
||||
// is set whether or not the graph itself can be constructed here.
|
||||
func TestNewApp_StopTimeout(t *testing.T) {
|
||||
t.Setenv("DATA_DIR", t.TempDir())
|
||||
|
||||
got := newApp().StopTimeout()
|
||||
|
||||
require.Equal(t, stopTimeout, got)
|
||||
require.Less(t, got, dockerStopGrace)
|
||||
}
|
||||
|
||||
// tailHeadroom is the slack the fx stop budget must keep beyond the
|
||||
// server stop hook. The hooks that run after the server — the
|
||||
// delivery engine, the healthcheck, the webhook DB manager and the
|
||||
// database close — are microsecond-scale in normal operation, so
|
||||
// this is generous for them.
|
||||
const tailHeadroom = 2 * time.Second
|
||||
|
||||
// TestStopTimeout_LeavesHeadroomForTailHooks pins the relationship
|
||||
// between the server's stop hook and the fx stop budget. fx bounds
|
||||
// the whole stop sequence, and returns without running its
|
||||
// remaining hooks once the stop context has expired. If the hook
|
||||
// could use the entire budget, every later hook — the database close
|
||||
// included — would be skipped in exactly the case where the drain
|
||||
// mattered.
|
||||
//
|
||||
// The hook is not just the HTTP drain: a Sentry flush follows it in
|
||||
// the same hook, and sentry.Flush honours no context, so both halves
|
||||
// have to be counted. The sweep walks every drain length the hook
|
||||
// can produce, since a shorter drain leaves the flush more room and
|
||||
// the worst case is not necessarily at either extreme.
|
||||
//
|
||||
// Shrinking either budget, or unbounding the flush again, must fail
|
||||
// here rather than silently recreating a hook that swallows the
|
||||
// whole sequence.
|
||||
func TestStopTimeout_LeavesHeadroomForTailHooks(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
require.Less(t, server.ShutdownTimeout, stopTimeout)
|
||||
|
||||
const step = 10 * time.Millisecond
|
||||
|
||||
for drain := time.Duration(0); drain <= server.ShutdownTimeout; drain += step {
|
||||
hook := drain + server.SentryFlushBudget(stopTimeout-drain)
|
||||
|
||||
require.LessOrEqual(
|
||||
t, hook+tailHeadroom, stopTimeout,
|
||||
"a %s drain leaves the tail hooks short", drain,
|
||||
)
|
||||
}
|
||||
}
|
||||
21
internal/lifecycle/export_test.go
Normal file
21
internal/lifecycle/export_test.go
Normal file
@@ -0,0 +1,21 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
)
|
||||
|
||||
// WaitDone exposes waitDone to the external test package. Only the
|
||||
// unexported waiter can be handed a channel that is already closed
|
||||
// before the call, which is the state the preamble exists for;
|
||||
// through WaitForShutdown the waiter goroutine may or may not have
|
||||
// closed the channel yet, so the case is not reachable
|
||||
// deterministically from outside.
|
||||
func WaitDone(
|
||||
ctx context.Context,
|
||||
log *slog.Logger,
|
||||
component string,
|
||||
done <-chan struct{},
|
||||
) error {
|
||||
return waitDone(ctx, log, component, done)
|
||||
}
|
||||
@@ -38,6 +38,29 @@ func WaitForShutdown(
|
||||
wg.Wait()
|
||||
}()
|
||||
|
||||
return waitDone(ctx, log, component, done)
|
||||
}
|
||||
|
||||
// waitDone waits for done to close, bounded by ctx.
|
||||
//
|
||||
// The non-blocking preamble is load-bearing. When the component has
|
||||
// already drained and ctx has already expired, both cases of the
|
||||
// bounded select are ready and Go picks between them uniformly at
|
||||
// random, so a clean shutdown would be reported as a timeout about
|
||||
// half the time. Draining wins: the goroutines are gone, and there
|
||||
// is nothing left for the operator to act on.
|
||||
func waitDone(
|
||||
ctx context.Context,
|
||||
log *slog.Logger,
|
||||
component string,
|
||||
done <-chan struct{},
|
||||
) error {
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
default:
|
||||
}
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
|
||||
@@ -37,6 +37,57 @@ func TestWaitForShutdown_DrainedGroup(t *testing.T) {
|
||||
)
|
||||
}
|
||||
|
||||
// racePasses is how many times the both-cases-ready race is run.
|
||||
// Without the preamble each pass is an independent coin flip, so
|
||||
// the probability of the whole loop passing by luck is 2^-N: at
|
||||
// this N the test is deterministic in practice, and it involves no
|
||||
// wall-clock waiting at all.
|
||||
const racePasses = 1000
|
||||
|
||||
// TestWaitDone_DrainedBeforeExpiredContext covers the case where a
|
||||
// component drained cleanly but the stop context had already
|
||||
// expired. Both select cases are ready, and Go chooses among ready
|
||||
// cases uniformly at random, so the drained case must be settled by
|
||||
// the preamble before the bounded select ever runs.
|
||||
func TestWaitDone_DrainedBeforeExpiredContext(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
done := make(chan struct{})
|
||||
close(done)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
for pass := range racePasses {
|
||||
require.NoErrorf(
|
||||
t,
|
||||
lifecycle.WaitDone(
|
||||
ctx, discardLogger(), "test component", done,
|
||||
),
|
||||
"pass %d reported a timeout for a drained component",
|
||||
pass,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWaitDone_ExpiredContext pins the other side of the preamble:
|
||||
// an expired context with a component that has not drained is still
|
||||
// a timeout.
|
||||
func TestWaitDone_ExpiredContext(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
err := lifecycle.WaitDone(
|
||||
ctx, discardLogger(), "test component",
|
||||
make(chan struct{}),
|
||||
)
|
||||
|
||||
require.ErrorIs(t, err, context.Canceled)
|
||||
require.ErrorContains(t, err, "test component")
|
||||
}
|
||||
|
||||
func TestWaitForShutdown_ContextExpires(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -48,6 +48,12 @@ const (
|
||||
// bound every request pays a walk proportional to whatever the
|
||||
// client sent.
|
||||
maxForwardedHops = 64
|
||||
|
||||
// ipv6BucketBits is the prefix length IPv6 clients are bucketed
|
||||
// on. A routed /64 is the normal residential and mobile
|
||||
// allocation, so it is the unit an attacker gets addresses in
|
||||
// and therefore the unit worth limiting.
|
||||
ipv6BucketBits = 64
|
||||
)
|
||||
|
||||
// normalizeAddr strips the IPv4-in-IPv6 wrapper and any zone from
|
||||
@@ -56,6 +62,40 @@ func normalizeAddr(addr netip.Addr) netip.Addr {
|
||||
return addr.Unmap().WithZone("")
|
||||
}
|
||||
|
||||
// bucketKey is the rate-limit bucket identity of a client address.
|
||||
// IPv4 keys on the full address; IPv6 keys on its /64 prefix,
|
||||
// because keying IPv6 per /128 lets one ordinary subscriber rotate
|
||||
// source addresses inside its own routed /64 and mint a fresh bucket
|
||||
// per request — evading every limiter here at the network layer,
|
||||
// with no spoofing and nothing to detect.
|
||||
//
|
||||
// An IPv4-mapped address (::ffff:1.2.3.4) is keyed as the IPv4
|
||||
// address it carries, never masked to a /64: mapped form all shares
|
||||
// the ::ffff:0:0/96 prefix, so masking would collapse every IPv4
|
||||
// client reaching a proxy that emits it into one bucket. Callers
|
||||
// pass addresses through normalizeAddr, which already unmaps; the
|
||||
// unmap here keeps the property true of the key function itself.
|
||||
//
|
||||
// The two families cannot collide: an IPv4 key is a bare dotted
|
||||
// quad, and an IPv6 key always carries a "/64" suffix.
|
||||
func bucketKey(addr netip.Addr) string {
|
||||
addr = addr.Unmap()
|
||||
|
||||
if addr.Is4() {
|
||||
return addr.String()
|
||||
}
|
||||
|
||||
// Prefix errors only on a negative bit count, on over 32 bits
|
||||
// for an IPv4 address, or on over 128 for IPv6. The count here
|
||||
// is the constant 64 and the IPv4 case returned above, so the
|
||||
// error is unreachable. (The zero Addr does not error either: it
|
||||
// yields the zero Prefix. Neither call site can produce one,
|
||||
// since both parse the address first.)
|
||||
prefix, _ := addr.Prefix(ipv6BucketBits)
|
||||
|
||||
return prefix.String()
|
||||
}
|
||||
|
||||
// isTrustedProxy reports whether addr belongs to a network the
|
||||
// operator listed in TRUSTED_PROXIES. The list is empty by default,
|
||||
// so by default nothing is trusted.
|
||||
@@ -143,6 +183,9 @@ func (m *Middleware) forwardedClientAddr(
|
||||
// another client's bucket, by picking an X-Forwarded-For value —
|
||||
// which makes every limit here decorative against a deliberate
|
||||
// attacker.
|
||||
//
|
||||
// The address that identifies the client is then reduced to a bucket
|
||||
// by bucketKey: full address for IPv4, /64 prefix for IPv6.
|
||||
func (m *Middleware) rateLimitKey(r *http.Request) (string, error) {
|
||||
return m.clientKey(r), nil
|
||||
}
|
||||
@@ -152,23 +195,25 @@ func (m *Middleware) clientKey(r *http.Request) string {
|
||||
peer, err := netip.ParseAddr(ipFromHostPort(r.RemoteAddr))
|
||||
if err != nil {
|
||||
// Not an address we can reason about; key on the raw
|
||||
// value, the most specific identity left. On a
|
||||
// Unix-socket listener every peer carries the same
|
||||
// RemoteAddr and so shares one bucket, which is the
|
||||
// fail-closed direction.
|
||||
// value, the most specific identity left. Distinct
|
||||
// RemoteAddr values stay in distinct buckets, so this
|
||||
// path cannot silently collapse unrelated clients
|
||||
// together. On a Unix-socket listener every peer
|
||||
// carries the same RemoteAddr and so shares one bucket,
|
||||
// which is the fail-closed direction.
|
||||
return r.RemoteAddr
|
||||
}
|
||||
|
||||
peer = normalizeAddr(peer)
|
||||
if !m.isTrustedProxy(peer) {
|
||||
return peer.String()
|
||||
return bucketKey(peer)
|
||||
}
|
||||
|
||||
if addr, ok := m.forwardedClientAddr(r); ok {
|
||||
return addr.String()
|
||||
return bucketKey(addr)
|
||||
}
|
||||
|
||||
return peer.String()
|
||||
return bucketKey(peer)
|
||||
}
|
||||
|
||||
// tooManyRequests returns the 429 handler used by the login,
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
"sneak.berlin/go/webhooker/internal/middleware"
|
||||
)
|
||||
@@ -370,6 +371,30 @@ const (
|
||||
headerXFF = "X-Forwarded-For"
|
||||
headerReal = "X-Real-IP"
|
||||
headerTrue = "True-Client-IP"
|
||||
|
||||
// clientIPv4 is the sample IPv4 client address these tests key
|
||||
// on, both directly and in IPv4-mapped form. clientIPv4Alt is
|
||||
// its neighbour, used to show the two do not share a bucket.
|
||||
clientIPv4 = "198.51.100.7"
|
||||
clientIPv4Alt = "198.51.100.8"
|
||||
|
||||
// clientIPv6 and clientIPv6Same are two addresses inside one
|
||||
// routed /64, so both must key on clientBucketV6.
|
||||
// clientIPv6Other is a different allocation and must key on
|
||||
// clientOtherBucketV6.
|
||||
clientIPv6 = "2001:db8:1:2:3:4:5:6"
|
||||
clientIPv6Same = "2001:db8:1:2:aaaa:bbbb:cccc:dddd"
|
||||
clientIPv6Other = "2001:db8:1:3::1"
|
||||
clientBucketV6 = "2001:db8:1:2::/64"
|
||||
clientOtherBucketV6 = "2001:db8:1:3::/64"
|
||||
|
||||
// trustedProxyCIDR is the proxy network the forwarded-path
|
||||
// tests configure, and trustedPeer an address inside it. A
|
||||
// production deployment is required to run behind a reverse
|
||||
// proxy with TRUSTED_PROXIES set, so this is the shape the
|
||||
// bucketing has to hold in.
|
||||
trustedProxyCIDR = "10.0.0.0/8"
|
||||
trustedPeer = "10.0.0.1:44444"
|
||||
)
|
||||
|
||||
// assertSharedBucket drives the login limiter from peer with the
|
||||
@@ -458,8 +483,8 @@ func TestRateLimitKey_SingleValuedHeadersIgnoredFromTrustedPeer(
|
||||
t.Parallel()
|
||||
|
||||
assertSharedBucket(
|
||||
t, trustedProxies("10.0.0.0/8"),
|
||||
"10.0.0.1:44444",
|
||||
t, trustedProxies(trustedProxyCIDR),
|
||||
trustedPeer,
|
||||
func(i int) map[string]string {
|
||||
return map[string]string{
|
||||
header: fmt.Sprintf(
|
||||
@@ -495,8 +520,8 @@ func TestRateLimitKey_MalformedRightmostHopFallsBackToPeer(
|
||||
t.Parallel()
|
||||
|
||||
assertSharedBucket(
|
||||
t, trustedProxies("10.0.0.0/8"),
|
||||
"10.0.0.1:44444",
|
||||
t, trustedProxies(trustedProxyCIDR),
|
||||
trustedPeer,
|
||||
func(i int) map[string]string {
|
||||
return map[string]string{
|
||||
headerXFF: fmt.Sprintf(
|
||||
@@ -522,13 +547,13 @@ func TestRateLimitKey_ForwardedHonouredFromTrustedPeer(
|
||||
t.Parallel()
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{
|
||||
TrustedProxies: trustedProxies("10.0.0.0/8"),
|
||||
TrustedProxies: trustedProxies(trustedProxyCIDR),
|
||||
})
|
||||
handler := m.LoginRateLimit()(okHandler())
|
||||
|
||||
const peer = "10.0.0.1:44444"
|
||||
const peer = trustedPeer
|
||||
|
||||
first := map[string]string{headerXFF: "198.51.100.7"}
|
||||
first := map[string]string{headerXFF: clientIPv4}
|
||||
|
||||
for range middleware.LoginRateLimitConst {
|
||||
postWithHeaders(handler, peer, loginPath, first)
|
||||
@@ -542,7 +567,7 @@ func TestRateLimitKey_ForwardedHonouredFromTrustedPeer(
|
||||
|
||||
w = postWithHeaders(
|
||||
handler, peer, loginPath,
|
||||
map[string]string{headerXFF: "198.51.100.8"},
|
||||
map[string]string{headerXFF: clientIPv4Alt},
|
||||
)
|
||||
assert.Equal(
|
||||
t, http.StatusOK, w.Code,
|
||||
@@ -559,7 +584,7 @@ func TestRateLimitKey_ChainWalkSkipsClientPrepended(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
assertSharedBucket(
|
||||
t, trustedProxies("10.0.0.0/8"), "10.0.0.1:44444",
|
||||
t, trustedProxies(trustedProxyCIDR), trustedPeer,
|
||||
func(i int) map[string]string {
|
||||
return map[string]string{
|
||||
headerXFF: fmt.Sprintf(
|
||||
@@ -594,7 +619,7 @@ func TestRateLimitKey_LongChainCapsWalkAndFallsBackToPeer(
|
||||
start := time.Now()
|
||||
|
||||
assertSharedBucket(
|
||||
t, trustedProxies("10.0.0.0/8"), "10.0.0.1:44444",
|
||||
t, trustedProxies(trustedProxyCIDR), trustedPeer,
|
||||
func(i int) map[string]string {
|
||||
return map[string]string{
|
||||
headerXFF: fmt.Sprintf("9.9.9.%d%s", i+1, padding),
|
||||
@@ -633,13 +658,13 @@ func TestRateLimitKey_LongChainAllocationIsBounded(t *testing.T) {
|
||||
)
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{
|
||||
TrustedProxies: trustedProxies("10.0.0.0/8"),
|
||||
TrustedProxies: trustedProxies(trustedProxyCIDR),
|
||||
})
|
||||
|
||||
req := httptest.NewRequestWithContext(
|
||||
context.Background(), http.MethodPost, loginPath, nil,
|
||||
)
|
||||
req.RemoteAddr = "10.0.0.1:44444"
|
||||
req.RemoteAddr = trustedPeer
|
||||
req.Header.Set(
|
||||
headerXFF, "9.9.9.9"+strings.Repeat(", 10.0.0.2", hops),
|
||||
)
|
||||
@@ -835,3 +860,369 @@ func TestReceiverRateLimit_IgnoresForwardedFromUntrustedPeer(
|
||||
"not mint a fresh receiver bucket",
|
||||
)
|
||||
}
|
||||
|
||||
// clientKeyFor returns the bucket key m computes for a request whose
|
||||
// direct peer is remoteAddr and which carries no forwarded headers.
|
||||
func clientKeyFor(
|
||||
t *testing.T, m *middleware.Middleware, remoteAddr string,
|
||||
) string {
|
||||
t.Helper()
|
||||
|
||||
req := httptest.NewRequestWithContext(
|
||||
context.Background(), http.MethodPost, loginPath, nil,
|
||||
)
|
||||
req.RemoteAddr = remoteAddr
|
||||
|
||||
return middleware.ClientKeyForTest(m, req)
|
||||
}
|
||||
|
||||
// TestRateLimitKey_IPv6BucketsByPrefix pins the key function's
|
||||
// address-family behaviour. IPv6 clients must bucket by /64 — a
|
||||
// routed /64 is the normal residential and mobile allocation, so
|
||||
// per-/128 keying lets one subscriber rotate source addresses and
|
||||
// mint a fresh bucket per request — while IPv4 keeps keying on the
|
||||
// full address and IPv4-mapped form is keyed as the IPv4 address it
|
||||
// carries.
|
||||
func TestRateLimitKey_IPv6BucketsByPrefix(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{})
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
peer string
|
||||
want string
|
||||
about string
|
||||
}{{
|
||||
name: "ipv6",
|
||||
peer: "[" + clientIPv6 + "]:44444",
|
||||
want: clientBucketV6,
|
||||
about: "an IPv6 peer must key on its /64",
|
||||
}, {
|
||||
name: "ipv6-other-in-same-64",
|
||||
peer: "[" + clientIPv6Same + "]:1",
|
||||
want: clientBucketV6,
|
||||
about: "another address in the same /64 must key the same",
|
||||
}, {
|
||||
name: "ipv6-different-64",
|
||||
peer: "[" + clientIPv6Other + "]:44444",
|
||||
want: clientOtherBucketV6,
|
||||
about: "a different /64 must key differently",
|
||||
}, {
|
||||
name: "ipv4",
|
||||
peer: clientIPv4 + ":44444",
|
||||
want: clientIPv4,
|
||||
about: "IPv4 must keep keying on the full address",
|
||||
}, {
|
||||
name: "ipv4-neighbour",
|
||||
peer: clientIPv4Alt + ":44444",
|
||||
want: clientIPv4Alt,
|
||||
about: "adjacent IPv4 addresses must not share a bucket",
|
||||
}, {
|
||||
name: "ipv4-mapped",
|
||||
peer: "[::ffff:" + clientIPv4 + "]:44444",
|
||||
want: clientIPv4,
|
||||
about: "IPv4-mapped form must key as the IPv4 address, " +
|
||||
"not be masked to a /64: mapped addresses all share " +
|
||||
"::ffff:0:0/96, so masking would collapse every IPv4 " +
|
||||
"client behind a mapping proxy into one bucket",
|
||||
}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
assert.Equal(
|
||||
t, tc.want, clientKeyFor(t, m, tc.peer), tc.about,
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestRateLimitKey_FamiliesDoNotCollide pins the structure the
|
||||
// no-collision property rests on, rather than one sample pair: every
|
||||
// IPv4 key is a bare address and every IPv6 key is a /64 in CIDR
|
||||
// form, so the two name spaces are disjoint by shape. Dropping the
|
||||
// masking strips the suffix that guarantees it, which is why this
|
||||
// asserts the form of each key and not just that two of them differ.
|
||||
func TestRateLimitKey_FamiliesDoNotCollide(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// Restated here rather than imported from the package under
|
||||
// test, so that changing the production bucket width fails this
|
||||
// test instead of silently moving with it.
|
||||
const wantBits = 64
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{})
|
||||
|
||||
v4Keys := map[string]bool{}
|
||||
|
||||
for _, peer := range []string{
|
||||
clientIPv4 + ":44444",
|
||||
clientIPv4Alt + ":44444",
|
||||
"[::ffff:" + clientIPv4 + "]:44444",
|
||||
} {
|
||||
key := clientKeyFor(t, m, peer)
|
||||
|
||||
addr, err := netip.ParseAddr(key)
|
||||
require.NoError(
|
||||
t, err, "%s: an IPv4 key must be a bare address", peer,
|
||||
)
|
||||
assert.True(
|
||||
t, addr.Is4(),
|
||||
"%s: an IPv4 key must be a dotted quad, got %q", peer, key,
|
||||
)
|
||||
|
||||
v4Keys[key] = true
|
||||
}
|
||||
|
||||
for _, peer := range []string{
|
||||
"[" + clientIPv6 + "]:44444",
|
||||
"[" + clientIPv6Same + "]:44444",
|
||||
"[" + clientIPv6Other + "]:44444",
|
||||
"[2001:db8::" + clientIPv4 + "]:44444",
|
||||
} {
|
||||
key := clientKeyFor(t, m, peer)
|
||||
|
||||
prefix, err := netip.ParsePrefix(key)
|
||||
require.NoError(
|
||||
t, err, "%s: an IPv6 key must be a CIDR prefix", peer,
|
||||
)
|
||||
assert.Equal(
|
||||
t, wantBits, prefix.Bits(),
|
||||
"%s: an IPv6 key must name a /64", peer,
|
||||
)
|
||||
assert.False(
|
||||
t, v4Keys[key],
|
||||
"%s: an IPv6 key must never equal an IPv4 key", peer,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRateLimitKey_UnparseablePeerKeepsDistinctBuckets covers the
|
||||
// fallback path. A RemoteAddr that is not an address must not panic,
|
||||
// and must not drop unrelated clients into one shared bucket by
|
||||
// accident: the raw value is the most specific identity left, so
|
||||
// distinct values stay in distinct buckets.
|
||||
func TestRateLimitKey_UnparseablePeerKeepsDistinctBuckets(
|
||||
t *testing.T,
|
||||
) {
|
||||
t.Parallel()
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{})
|
||||
|
||||
first := clientKeyFor(t, m, "not-an-address")
|
||||
second := clientKeyFor(t, m, "also-not-an-address:1234")
|
||||
|
||||
assert.NotEmpty(t, first)
|
||||
assert.NotEqual(
|
||||
t, first, second,
|
||||
"unparseable peers must not collapse into one bucket",
|
||||
)
|
||||
}
|
||||
|
||||
// TestLoginRateLimit_IPv6SharesBucketWithinSlash64 is the behavioural
|
||||
// half, and the regression test for the bypass itself: a client that
|
||||
// rotates source addresses inside its own routed /64 must stay in one
|
||||
// bucket. Reverting the masking makes this test fail, because each
|
||||
// rotated address would mint a fresh bucket and nothing would be
|
||||
// rejected.
|
||||
func TestLoginRateLimit_IPv6SharesBucketWithinSlash64(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{})
|
||||
handler := m.LoginRateLimit()(okHandler())
|
||||
|
||||
for i := range middleware.LoginRateLimitConst {
|
||||
w := postWithHeaders(
|
||||
handler,
|
||||
fmt.Sprintf("[2001:db8:1:2::%d]:44444", i+1),
|
||||
loginPath, nil,
|
||||
)
|
||||
assert.Equal(
|
||||
t, http.StatusOK, w.Code, "request %d should pass", i,
|
||||
)
|
||||
}
|
||||
|
||||
w := postWithHeaders(
|
||||
handler, "[2001:db8:1:2::ffff]:44444", loginPath, nil,
|
||||
)
|
||||
assert.Equal(
|
||||
t, http.StatusTooManyRequests, w.Code,
|
||||
"rotating source addresses inside one routed /64 must not "+
|
||||
"mint fresh buckets",
|
||||
)
|
||||
}
|
||||
|
||||
// TestLoginRateLimit_IPv6IndependentAcrossSlash64 is the other side
|
||||
// of the trade: bucketing by /64 must not merge separate allocations,
|
||||
// so a client in a different /64 keeps its own limit.
|
||||
func TestLoginRateLimit_IPv6IndependentAcrossSlash64(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{})
|
||||
handler := m.LoginRateLimit()(okHandler())
|
||||
|
||||
for range middleware.LoginRateLimitConst + 1 {
|
||||
postWithHeaders(
|
||||
handler, "[2001:db8:1:2::1]:44444", loginPath, nil,
|
||||
)
|
||||
}
|
||||
|
||||
w := postWithHeaders(
|
||||
handler, "[2001:db8:1:3::1]:44444", loginPath, nil,
|
||||
)
|
||||
assert.Equal(
|
||||
t, http.StatusOK, w.Code,
|
||||
"a different /64 must have its own bucket",
|
||||
)
|
||||
}
|
||||
|
||||
// TestLoginRateLimit_IPv4IndependentPerAddress guards against the
|
||||
// masking leaking into IPv4: two addresses one apart must still hold
|
||||
// separate buckets.
|
||||
func TestLoginRateLimit_IPv4IndependentPerAddress(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{})
|
||||
handler := m.LoginRateLimit()(okHandler())
|
||||
|
||||
for range middleware.LoginRateLimitConst + 1 {
|
||||
postWithHeaders(
|
||||
handler, clientIPv4+":44444", loginPath, nil,
|
||||
)
|
||||
}
|
||||
|
||||
w := postWithHeaders(
|
||||
handler, clientIPv4Alt+":44444", loginPath, nil,
|
||||
)
|
||||
assert.Equal(
|
||||
t, http.StatusOK, w.Code,
|
||||
"a second IPv4 address must have its own bucket",
|
||||
)
|
||||
}
|
||||
|
||||
// forwardedKeyFor returns the bucket key m computes for a request
|
||||
// that arrives from trustedPeer — a configured trusted proxy — and
|
||||
// names forwarded as its client in X-Forwarded-For. That is the
|
||||
// production path: a deployment is required to run behind a reverse
|
||||
// proxy with TRUSTED_PROXIES set, so the forwarded address, not the
|
||||
// peer, is what the limiters bucket on there.
|
||||
func forwardedKeyFor(
|
||||
t *testing.T, m *middleware.Middleware, forwarded string,
|
||||
) string {
|
||||
t.Helper()
|
||||
|
||||
req := httptest.NewRequestWithContext(
|
||||
context.Background(), http.MethodPost, loginPath, nil,
|
||||
)
|
||||
req.RemoteAddr = trustedPeer
|
||||
req.Header.Set(headerXFF, forwarded)
|
||||
|
||||
return middleware.ClientKeyForTest(m, req)
|
||||
}
|
||||
|
||||
// TestRateLimitKey_ForwardedIPv6BucketsByPrefix pins the /64
|
||||
// bucketing on the trusted-proxy branch. The direct-peer tests above
|
||||
// cannot reach it, so without this the masking could be reverted for
|
||||
// forwarded clients alone — the only shape a production deployment
|
||||
// runs in — and the rest of the suite would stay green.
|
||||
func TestRateLimitKey_ForwardedIPv6BucketsByPrefix(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{
|
||||
TrustedProxies: trustedProxies(trustedProxyCIDR),
|
||||
})
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
forwarded string
|
||||
want string
|
||||
about string
|
||||
}{{
|
||||
name: "ipv6",
|
||||
forwarded: clientIPv6,
|
||||
want: clientBucketV6,
|
||||
about: "a forwarded IPv6 client must key on its /64",
|
||||
}, {
|
||||
name: "ipv6-other-in-same-64",
|
||||
forwarded: clientIPv6Same,
|
||||
want: clientBucketV6,
|
||||
about: "another forwarded address in the same /64 must " +
|
||||
"key the same",
|
||||
}, {
|
||||
name: "ipv6-different-64",
|
||||
forwarded: clientIPv6Other,
|
||||
want: clientOtherBucketV6,
|
||||
about: "a forwarded address in another /64 must differ",
|
||||
}, {
|
||||
name: "ipv4",
|
||||
forwarded: clientIPv4,
|
||||
want: clientIPv4,
|
||||
about: "a forwarded IPv4 client must key on the address",
|
||||
}, {
|
||||
name: "ipv4-mapped",
|
||||
forwarded: "::ffff:" + clientIPv4,
|
||||
want: clientIPv4,
|
||||
about: "a proxy that forwards IPv4-mapped form must key as " +
|
||||
"the IPv4 address it carries, not be masked to a /64: " +
|
||||
"mapped addresses all share ::ffff:0:0/96",
|
||||
}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
assert.Equal(
|
||||
t, tc.want,
|
||||
forwardedKeyFor(t, m, tc.forwarded), tc.about,
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoginRateLimit_ForwardedIPv6SharesBucketWithinSlash64 is the
|
||||
// behavioural half on the production path: behind a trusted proxy, a
|
||||
// client rotating source addresses inside its own routed /64 must
|
||||
// stay in one bucket.
|
||||
func TestLoginRateLimit_ForwardedIPv6SharesBucketWithinSlash64(
|
||||
t *testing.T,
|
||||
) {
|
||||
t.Parallel()
|
||||
|
||||
assertSharedBucket(
|
||||
t, trustedProxies(trustedProxyCIDR), trustedPeer,
|
||||
func(i int) map[string]string {
|
||||
return map[string]string{
|
||||
headerXFF: fmt.Sprintf("2001:db8:1:2::%d", i+1),
|
||||
}
|
||||
},
|
||||
"rotating forwarded source addresses inside one routed /64 "+
|
||||
"must not mint fresh buckets",
|
||||
)
|
||||
}
|
||||
|
||||
// TestLoginRateLimit_ForwardedIPv6IndependentAcrossSlash64 is the
|
||||
// other side of that trade on the same path: bucketing by /64 must
|
||||
// not merge two allocations reaching the proxy.
|
||||
func TestLoginRateLimit_ForwardedIPv6IndependentAcrossSlash64(
|
||||
t *testing.T,
|
||||
) {
|
||||
t.Parallel()
|
||||
|
||||
m := rateLimitMiddleware(t, &config.Config{
|
||||
TrustedProxies: trustedProxies(trustedProxyCIDR),
|
||||
})
|
||||
handler := m.LoginRateLimit()(okHandler())
|
||||
|
||||
spent := map[string]string{headerXFF: clientIPv6}
|
||||
for range middleware.LoginRateLimitConst + 1 {
|
||||
postWithHeaders(handler, trustedPeer, loginPath, spent)
|
||||
}
|
||||
|
||||
w := postWithHeaders(
|
||||
handler, trustedPeer, loginPath,
|
||||
map[string]string{headerXFF: clientIPv6Other},
|
||||
)
|
||||
assert.Equal(
|
||||
t, http.StatusOK, w.Code,
|
||||
"a forwarded client in a different /64 must have its own "+
|
||||
"bucket",
|
||||
)
|
||||
}
|
||||
|
||||
@@ -24,15 +24,48 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
// shutdownTimeout is the maximum time to wait for the HTTP
|
||||
// ShutdownTimeout is the maximum time to wait for the HTTP
|
||||
// server to finish in-flight requests during shutdown.
|
||||
shutdownTimeout = 5 * time.Second
|
||||
//
|
||||
// It must stay strictly below the fx stop timeout in
|
||||
// cmd/webhooker, which bounds the whole stop sequence: a drain
|
||||
// that used the entire sequence budget would leave nothing for
|
||||
// the hooks that run after the server, including the database
|
||||
// close. It is exported so that relationship can be tested.
|
||||
ShutdownTimeout = 3 * time.Second
|
||||
|
||||
// sentryFlushTimeout is the maximum time to wait for Sentry
|
||||
// to flush pending events during shutdown.
|
||||
// TailHookReserve is the share of the fx stop budget this hook
|
||||
// refuses to spend, leaving it for the hooks that run after the
|
||||
// server: the delivery engine, the healthcheck, the webhook DB
|
||||
// manager and the database close.
|
||||
TailHookReserve = 2 * time.Second
|
||||
|
||||
// sentryFlushTimeout is the longest wait for Sentry to flush
|
||||
// pending events during shutdown, before the remaining stop
|
||||
// budget is taken into account.
|
||||
sentryFlushTimeout = 2 * time.Second
|
||||
|
||||
// minSentryFlush is the shortest flush worth attempting. Below
|
||||
// it the remaining budget goes to the tail hooks instead.
|
||||
minSentryFlush = 250 * time.Millisecond
|
||||
)
|
||||
|
||||
// SentryFlushBudget reports how long the Sentry flush may run when
|
||||
// remaining is the time left on the fx stop context after the HTTP
|
||||
// drain. sentry.Flush takes a bare duration and honours no context,
|
||||
// so this clamp is the only thing keeping a stalled flush from
|
||||
// spending the tail hooks' share of the budget on top of a
|
||||
// full-length drain. TailHookReserve is held back, and anything
|
||||
// under minSentryFlush is skipped rather than attempted uselessly.
|
||||
func SentryFlushBudget(remaining time.Duration) time.Duration {
|
||||
budget := min(remaining-TailHookReserve, sentryFlushTimeout)
|
||||
if budget < minSentryFlush {
|
||||
return 0
|
||||
}
|
||||
|
||||
return budget
|
||||
}
|
||||
|
||||
//nolint:revive // ServerParams is a standard fx naming convention.
|
||||
type ServerParams struct {
|
||||
fx.In
|
||||
@@ -164,7 +197,7 @@ func (s *Server) cleanShutdown(ctx context.Context) {
|
||||
s.exitCode = 0
|
||||
|
||||
ctxShutdown, shutdownCancel := context.WithTimeout(
|
||||
ctx, shutdownTimeout,
|
||||
ctx, ShutdownTimeout,
|
||||
)
|
||||
defer shutdownCancel()
|
||||
|
||||
@@ -178,10 +211,31 @@ func (s *Server) cleanShutdown(ctx context.Context) {
|
||||
s.cleanupForExit()
|
||||
|
||||
if s.sentryEnabled {
|
||||
sentry.Flush(sentryFlushTimeout)
|
||||
s.flushSentry(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
// flushSentry drains Sentry's queue inside what is left of the fx
|
||||
// stop budget. A context carrying no deadline — a caller outside the
|
||||
// fx lifecycle — gets the full timeout.
|
||||
func (s *Server) flushSentry(ctx context.Context) {
|
||||
flush := sentryFlushTimeout
|
||||
|
||||
if deadline, ok := ctx.Deadline(); ok {
|
||||
flush = SentryFlushBudget(time.Until(deadline))
|
||||
}
|
||||
|
||||
if flush <= 0 {
|
||||
s.log.Warn(
|
||||
"skipping sentry flush, stop budget exhausted",
|
||||
)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
sentry.Flush(flush)
|
||||
}
|
||||
|
||||
func (s *Server) configure() {
|
||||
// identify ourselves in the logs
|
||||
s.params.Logger.Identify()
|
||||
|
||||
59
internal/server/shutdown_test.go
Normal file
59
internal/server/shutdown_test.go
Normal file
@@ -0,0 +1,59 @@
|
||||
package server_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/webhooker/internal/server"
|
||||
)
|
||||
|
||||
// TestSentryFlushBudget covers the clamp that keeps the Sentry flush
|
||||
// from spending the tail hooks' share of the fx stop budget.
|
||||
// sentry.Flush ignores the stop context, so without the clamp a
|
||||
// stalled flush adds its whole timeout on top of the HTTP drain.
|
||||
func TestSentryFlushBudget(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
remaining time.Duration
|
||||
want time.Duration
|
||||
}{
|
||||
{
|
||||
name: "full drain leaves only the reserve",
|
||||
remaining: server.TailHookReserve,
|
||||
want: 0,
|
||||
},
|
||||
{
|
||||
name: "expired budget",
|
||||
remaining: -time.Second,
|
||||
want: 0,
|
||||
},
|
||||
{
|
||||
name: "sliver above the reserve is not worth it",
|
||||
remaining: server.TailHookReserve + 10*time.Millisecond,
|
||||
want: 0,
|
||||
},
|
||||
{
|
||||
name: "partial flush when some room is left",
|
||||
remaining: server.TailHookReserve + time.Second,
|
||||
want: time.Second,
|
||||
},
|
||||
{
|
||||
name: "capped at the nominal timeout",
|
||||
remaining: time.Hour,
|
||||
want: 2 * time.Second,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
require.Equal(
|
||||
t, tt.want, server.SentryFlushBudget(tt.remaining),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user