From f443a799080aba4532df3664107152961a148f85 Mon Sep 17 00:00:00 2001 From: Baylem Date: Mon, 9 Nov 2026 15:26:09 -0500 Subject: [PATCH 1/4] fix(sentinel): retry shared wakes within their budget Signed-off-by: Baylem --- .github/workflows/ci.yaml | 6 + sentinel/main.go | 5 +- sentinel/specs.md | 11 +- sentinel/wake_load_test.go | 356 +++++++++++++++++++++++++++++ sentinel/wake_retry.go | 93 ++++++++ sentinel/wake_retry_test.go | 200 ++++++++++++++++ specs/031-wake-retry-load/spec.md | 48 ++++ specs/031-wake-retry-load/tasks.md | 12 + 8 files changed, 728 insertions(+), 3 deletions(-) create mode 100644 sentinel/wake_load_test.go create mode 100644 sentinel/wake_retry.go create mode 100644 sentinel/wake_retry_test.go create mode 100644 specs/031-wake-retry-load/spec.md create mode 100644 specs/031-wake-retry-load/tasks.md diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 81011338d..4e690573c 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -926,6 +926,12 @@ jobs: read -r -a packages <<<"$GO_PACKAGES" go test -covermode=atomic -coverpkg=./... \ -coverprofile=coverage/unit.out "${packages[@]}" + - name: sentinel mixed-load regression measurements + if: matrix.module == 'sentinel' && env.GO_PACKAGES != '' + working-directory: sentinel + # Keep successful fixture measurements visible, including race checks, + # without changing the matrix or package-selection policy. + run: go test -race -count=1 -v -run '^TestWakeMixedLoadOutageRecovery$' ./... - name: agent file access across real UIDs if: matrix.module == 'agent' && matrix.arch == 'amd64' working-directory: agent diff --git a/sentinel/main.go b/sentinel/main.go index 954962e98..37e72997e 100644 --- a/sentinel/main.go +++ b/sentinel/main.go @@ -441,7 +441,10 @@ func (w *waker) RequestWake(ctx context.Context) (result error) { return fmt.Errorf("encode wake patch: %w", err) } - if _, err := w.res.Patch(patchCtx, w.name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { + if err := retryWakePatch(patchCtx, func(attemptCtx context.Context) error { + _, err := w.res.Patch(attemptCtx, w.name, types.MergePatchType, patch, metav1.PatchOptions{}) + return err + }); err != nil { return fmt.Errorf("patch %s annotation: %w", idleWakeRequestedAnnotation, err) } return nil diff --git a/sentinel/specs.md b/sentinel/specs.md index d74e9ade8..37564d70c 100644 --- a/sentinel/specs.md +++ b/sentinel/specs.md @@ -36,6 +36,11 @@ Wake-on-connect daemon that runs as a replacement pod for dormant game servers. sentinel/ ├── main.go # Entry point, config parsing, listeners, handlers, upstream polling, proxying ├── main_test.go # Config parsing, ports parsing, UDP heuristic, waker, TCP/UDP handlers, proxying, bidirectional copy +├── wake_retry.go # Bounded transient retry inside the shared wake operation +├── wake_retry_test.go # Retry limits, classification, deadlines, sharing, jitter +├── wake_load_test.go # Mixed TCP/UDP outage, recovery, cold start and drain fixture +├── wake_budget_test.go # Wake/join budgets, UDP worker and session regressions +├── udp_wake.go # Capacity-one coalescing UDP wake worker ├── Dockerfile # Image build (sets Version at compile time) ├── go.mod # Dependencies: k8s.io/client-go, k8s.io/apimachinery, github.com/GameplanePanel/gameplane/gameproto ├── go.sum # Locked versions @@ -246,14 +251,14 @@ Each accepted TCP connection runs as a *session* under its own context, separate 5. **No data duplication or loss.** The complete original stream is: `Consumed + remaining(bufio.Reader) + new-data-from-client`. Both classifiers and proxying preserve this boundary. -6. **Annotation patch coalesces bursts.** Callers awaiting an in-flight PATCH observe that attempt's completion and error, unless their own context expires first. Only successful attempts start the `wakePatchInterval` cooldown; failures release the shared attempt so a later request can retry. One worker per UDP listener accepts a one-entry coalescing queue; packet reception and bounded source tracking continue while Kubernetes is stalled. Listener shutdown cancels and joins that worker. +6. **Annotation patch coalesces bursts.** Callers share an in-flight wake operation, including its retries and final error, unless their own context expires first. Each operation makes at most four application-level PATCH calls, reusing one immutable merge patch/token. It retries recognizable transient API failures (service unavailable, throttling, API/server timeout, internal error) and transport failures (EOF, reset/refused connections, broken pipes, network timeouts). Permanent API statuses (including authorization, invalid requests and missing resources), context cancellation and unknown errors return without retry. Backoff starts at 100 ms, doubles per failure, uses +/-25% jitter and is capped at 800 ms after jitter. A Kubernetes suggested retry delay is a minimum and may exceed that cap. All PATCH calls and waits share the original five-second budget, further bounded by the owning caller's join context; client-go's own transport retries are also within that context. Errors retain the final underlying cause for `errors.Is/As`, including context expiry during backoff. Canceled callers never retry; live waiters may replace a canceled owner within their own remaining budget. Only successful operations start the `wakePatchInterval` cooldown; terminal failures release the operation for a later caller. One worker per UDP listener accepts a one-entry coalescing queue; packet reception and bounded source tracking continue while Kubernetes is stalled. Listener shutdown cancels and joins that worker. 7. **UDP sources are evicted fairly.** The least-recently-active source is evicted (LRU), not a random or oldest source, so brief bursts of scanners don't unfairly starve legitimate players trying to rejoin. ## Dependencies **Stdlib:** -- `bufio`, `bytes`, `context`, `encoding/json`, `fmt`, `io`, `log`, `net`, `os`, `os/signal`, `strconv`, `strings`, `sync`, `syscall`, `time`. +- `bufio`, `bytes`, `context`, `encoding/json`, `fmt`, `io`, `log`, `math/rand/v2`, `net`, `os`, `os/signal`, `strconv`, `strings`, `sync`, `syscall`, `time`. **Kubernetes:** - `k8s.io/apimachinery/pkg/apis/meta/v1` — metav1 types (PatchOptions, GetOptions). @@ -293,6 +298,8 @@ Each accepted TCP connection runs as a *session* under its own context, separate - **Shutdown drain (F-179):** `TestRunDrainsProxiedSessionAfterContextCancel` (replaces `TestProxyBidirectionalStopsOnContextCancel`, which asserted the old cut-on-cancel behaviour) verifies that after cancel new connections are refused while a handed-through session keeps working in both directions and `run` returns only once it ends; `TestRunForceClosesSessionsAfterDrainTimeout` verifies the drain is bounded; `TestLoadConfigShutdownDrainTimeout` covers `SHUTDOWN_DRAIN_TIMEOUT` parsing. - **Listener failure (F-180):** `TestRunSurfacesUDPListenFailureWhileTCPHealthy` verifies a UDP bind failure is returned promptly while a TCP listener is still healthy. - **Waker (annotation patching):** `TestWakerRequestWakeSetsAnnotation`, `TestWakerRequestWakePreservesOtherAnnotations`, `TestWakerRequestWakeCoalescesBursts` verify merge patch isolation, burst coalescing. +- **Wake retries:** `wake_retry_test.go` covers transient failure then success, shared callers observing one retry sequence, immutable patch bytes, four-attempt exhaustion, permanent/unknown errors, cancellation/deadlines during backoff, retry-after and jitter bounds. Runtime red/green verification runs in CI only. +- **Mixed load:** `TestWakeMixedLoadOutageRecovery` uses real loopback listeners and an HTTP fake Kubernetes endpoint. A concurrent burst of 16 TCP players and 25 UDP sources joins a stalled shared PATCH; an outage response forces retry, followed by deliberately delayed game startup. Observable receive/waiter barriers assert UDP reception during the API stall, one shared PATCH at a time and one active plus one queued UDP wake. All TCP players exchange bytes after handoff, beyond their recorded join deadlines and during listener shutdown/session drain. The conditional verbose sentinel CI step logs actual attempts, handoffs, UDP receives, maximum concurrency, elapsed time and handoff latency, with race detection. These are measurements of the fixture, not a production capacity maximum; every player retains its own upstream connection. - **Wake availability:** `wake_budget_test.go` covers a hanging HTTP API, shared failure and retry, the total join deadline, worker cancellation/coalescing, successful gameplay beyond the join deadline, and receiving a 101-datagram burst while wake is stalled. CI logs elapsed receive time; this fixture does not establish a production capacity ceiling. - **Registry dispatch:** `TestRegistryProtocolDispatchMinecraftJoinWakes`, `TestRegistryProtocolDispatchMinecraftStatusPingNoWake`, `TestRegistryProtocolDispatchTerrariaJoinWakes`, `TestRegistryProtocolDispatchUnknownNoWake`, `TestRegistryProtocolReplayConsumedBytes` verify registry lookup, Classifier.Classify invocation, protocol-agnostic join/status/unknown handling, and consumed-byte replay. - **Startup validation:** `TestParsePortsConfigRejectsUnknownProtocol` verifies that unknown protocol names cause a fatal startup error with a clear message listing available protocols. diff --git a/sentinel/wake_load_test.go b/sentinel/wake_load_test.go new file mode 100644 index 000000000..582bf5043 --- /dev/null +++ b/sentinel/wake_load_test.go @@ -0,0 +1,356 @@ +package main + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "k8s.io/client-go/dynamic" + "k8s.io/client-go/rest" +) + +// These timeouts only prevent a stuck fixture from hanging CI. The logged +// timings describe this loopback run, not a supported production capacity. +const loadSafetyTimeout = 10 * time.Second + +func receiveLoadEvent[T any](t *testing.T, events <-chan T, name string) T { + t.Helper() + select { + case value := <-events: + return value + case <-time.After(loadSafetyTimeout): + t.Fatalf("timed out waiting for %s", name) + var zero T + return zero + } +} + +func waitLoadEvent(t *testing.T, events <-chan struct{}, name string) { + t.Helper() + receiveLoadEvent(t, events, name) +} + +func updateLoadMaximum(maximum *atomic.Int32, value int32) { + for old := maximum.Load(); value > old; old = maximum.Load() { + if maximum.CompareAndSwap(old, value) { + return + } + } +} + +type loadWaitContext struct { + context.Context + joined chan<- struct{} + once sync.Once +} + +func (c *loadWaitContext) Done() <-chan struct{} { + c.once.Do(func() { c.joined <- struct{}{} }) + return c.Context.Done() +} + +type loadWakeObserver struct { + wake wakeRequester + joined chan struct{} + deadlines chan time.Time + finished chan error + calls atomic.Int32 + active atomic.Int32 + maximum atomic.Int32 +} + +func (w *loadWakeObserver) RequestWake(ctx context.Context) error { + w.calls.Add(1) + updateLoadMaximum(&w.maximum, w.active.Add(1)) + defer w.active.Add(-1) + if w.deadlines != nil { + deadline, _ := ctx.Deadline() + w.deadlines <- deadline + } + if w.joined != nil { + ctx = &loadWaitContext{Context: ctx, joined: w.joined} + } + err := w.wake.RequestWake(ctx) + w.finished <- err + return err +} + +func TestWakeMixedLoadOutageRecovery(t *testing.T) { + const tcpPlayers, udpSources = 16, 24 + const joinBudget = 4 * time.Second + started := time.Now() + listenerCtx, stopListeners := context.WithCancel(t.Context()) + defer stopListeners() + sessionCtx, stopSessions := context.WithCancel(context.WithoutCancel(t.Context())) + defer stopSessions() + sessions := &sessionGroup{ctx: sessionCtx} + defer sessions.drain(0, stopSessions) + + // The first API request is deliberately stalled until the entire mixed + // burst has joined/arrived. It then fails; only the retry can recover. + releaseOutage := make(chan struct{}) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseOutage) }) } + defer release() + apiCalls := make(chan int32, 16) + var patches, apiActive, apiMaximum atomic.Int32 + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + defer apiActive.Add(-1) + updateLoadMaximum(&apiMaximum, apiActive.Add(1)) + _, _ = io.Copy(io.Discard, req.Body) + if req.Method != http.MethodPatch { + t.Errorf("API method = %s, want PATCH", req.Method) + } + attempt := patches.Add(1) + apiCalls <- attempt + w.Header().Set("Content-Type", "application/json") + if attempt == 1 { + select { + case <-releaseOutage: + case <-req.Context().Done(): + return + } + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = io.WriteString(w, `{"kind":"Status","apiVersion":"v1","status":"Failure","reason":"ServiceUnavailable","message":"fixture outage","code":503}`) + return + } + _, _ = io.WriteString(w, `{"apiVersion":"gameplane.local/v1alpha1","kind":"GameServer","metadata":{"name":"alpha"}}`) + })) + defer api.Close() + // Unblock the handler before Close even if an assertion fails mid-outage. + defer release() + client, err := dynamic.NewForConfig(&rest.Config{Host: api.URL}) + if err != nil { + t.Fatal(err) + } + shared := newWaker(client.Resource(gameServerGVR).Namespace("games"), "alpha", time.Minute) + udpWake := &loadWakeObserver{wake: shared, finished: make(chan error, 4)} + tcpWake := &loadWakeObserver{wake: shared, joined: make(chan struct{}, tcpPlayers), deadlines: make(chan time.Time, tcpPlayers), finished: make(chan error, tcpPlayers)} + + lc := &net.ListenConfig{} + tcpListener, err := lc.Listen(listenerCtx, "tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer tcpListener.Close() + udpListener, err := lc.ListenPacket(listenerCtx, "udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer udpListener.Close() + // Reserve then release the upstream port: there is no game listener + // during outage/recovery. Binding is an explicit cold-start event below. + probe, err := lc.Listen(listenerCtx, "tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + upstreamAddr := probe.Addr().String() + _ = probe.Close() + cfg := Config{WakeDeadline: joinBudget, MaxConnections: tcpPlayers, UDPThreshold: 1, UDPWindow: time.Minute, UDPMaxSources: udpSources + 1, upstreamAddrOverride: upstreamAddr} + reads := make(chan struct{}, udpSources+2) + errCh := make(chan error, 2) + tcpDone, udpDone := make(chan struct{}), make(chan struct{}) + go func() { + defer close(tcpDone) + serveTCP(listenerCtx, tcpListener, PortConfig{Protocol: "TCP", WakeProtocol: "generic"}, tcpWake, cfg, make(chan struct{}, tcpPlayers), sessions, errCh) + }() + go func() { + defer close(udpDone) + serveUDPPackets(listenerCtx, observedPacketConn{udpListener, reads}, PortConfig{Protocol: "UDP"}, udpWake, cfg, errCh) + }() + + udpClients := make([]net.Conn, udpSources+1) + for i := range udpClients { + conn, dialErr := (&net.Dialer{}).DialContext(listenerCtx, "udp", udpListener.LocalAddr().String()) + if dialErr != nil { + t.Fatal(dialErr) + } + udpClients[i] = conn + defer conn.Close() + } + if _, err := udpClients[0].Write([]byte("start wake")); err != nil { + t.Fatal(err) + } + if attempt := receiveLoadEvent(t, apiCalls, "stalled API request"); attempt != 1 { + t.Fatalf("first API attempt = %d", attempt) + } + waitLoadEvent(t, reads, "initial UDP packet") + + type playerResult struct { + conn net.Conn + latency time.Duration + err error + } + players := make(chan playerResult, tcpPlayers) + udpSent := make(chan error, udpSources) + burst := make(chan struct{}) + for i := range tcpPlayers { + go func() { + <-burst + joinStarted := time.Now() + conn, dialErr := (&net.Dialer{}).DialContext(listenerCtx, "tcp", tcpListener.Addr().String()) + if dialErr != nil { + players <- playerResult{err: dialErr} + return + } + t.Cleanup(func() { _ = conn.Close() }) + _ = conn.SetDeadline(time.Now().Add(loadSafetyTimeout)) + payload := fmt.Sprintf("player-%02d", i) + err := exchangeLoadPayload(conn, payload) + players <- playerResult{conn: conn, latency: time.Since(joinStarted), err: err} + }() + } + for _, conn := range udpClients[1:] { + go func() { + <-burst + _, err := conn.Write([]byte("burst wake")) + udpSent <- err + }() + } + close(burst) + for range udpSources { + if err := receiveLoadEvent(t, udpSent, "UDP burst send"); err != nil { + t.Fatal(err) + } + waitLoadEvent(t, reads, "UDP receive while API stalled") + } + // An additional read proves the previous datagram finished heuristic + // processing (ReadFrom is called again only after worker.request returns). + if _, err := udpClients[0].Write([]byte("receive barrier")); err != nil { + t.Fatal(err) + } + waitLoadEvent(t, reads, "UDP heuristic barrier") + var latestDeadline time.Time + for range tcpPlayers { + waitLoadEvent(t, tcpWake.joined, "TCP caller sharing stalled wake") + deadline := receiveLoadEvent(t, tcpWake.deadlines, "TCP join deadline") + if deadline.After(latestDeadline) { + latestDeadline = deadline + } + } + if patches.Load() != 1 || apiActive.Load() != 1 || udpWake.calls.Load() != 1 || udpWake.maximum.Load() != 1 { + t.Fatalf("stalled burst: patches=%d API active=%d UDP calls=%d UDP max=%d", patches.Load(), apiActive.Load(), udpWake.calls.Load(), udpWake.maximum.Load()) + } + udpElapsed := time.Since(started) + release() + if attempt := receiveLoadEvent(t, apiCalls, "recovery PATCH"); attempt != 2 { + t.Fatalf("recovery attempt = %d, want 2", attempt) + } + for range tcpPlayers { + if err := receiveLoadEvent(t, tcpWake.finished, "shared TCP wake completion"); err != nil { + t.Fatalf("TCP shared wake: %v", err) + } + } + // The worker may consume just one queued request after its active call. + for range 2 { + if err := receiveLoadEvent(t, udpWake.finished, "bounded UDP wake completion"); err != nil { + t.Fatal(err) + } + } + if udpWake.calls.Load() != 2 || udpWake.maximum.Load() != 1 { + t.Fatalf("UDP work calls=%d max=%d, want one active plus one queued", udpWake.calls.Load(), udpWake.maximum.Load()) + } + + // This timer models deliberately delayed game startup after successful + // API recovery. It is not a guess about when another goroutine will run. + coldStart := time.NewTimer(200 * time.Millisecond) + defer coldStart.Stop() + <-coldStart.C + game, err := lc.Listen(listenerCtx, "tcp", upstreamAddr) + if err != nil { + t.Fatal(err) + } + defer game.Close() + gameDone := make(chan struct{}) + go func() { + defer close(gameDone) + var echoes sync.WaitGroup + defer echoes.Wait() + for range tcpPlayers { + conn, acceptErr := game.Accept() + if acceptErr != nil { + return + } + echoes.Add(1) + go func() { + defer echoes.Done() + defer conn.Close() + _, _ = io.Copy(conn, conn) + }() + } + }() + var maxLatency, totalLatency time.Duration + connected := make([]net.Conn, 0, tcpPlayers) + for range tcpPlayers { + player := receiveLoadEvent(t, players, "TCP game handoff and echo") + if player.err != nil { + t.Fatalf("TCP handoff: %v", player.err) + } + connected = append(connected, player.conn) + totalLatency += player.latency + if player.latency > maxLatency { + maxLatency = player.latency + } + } + if patches.Load() != 2 || apiMaximum.Load() != 1 { + t.Fatalf("API attempts=%d maximum concurrency=%d, want 2 and 1", patches.Load(), apiMaximum.Load()) + } + // Waiting for the actual last recorded join deadline, rather than an + // estimated goroutine start time, verifies every session outlives it. + deadlineElapsed := time.NewTimer(time.Until(latestDeadline) + time.Millisecond) + defer deadlineElapsed.Stop() + <-deadlineElapsed.C + for _, conn := range connected { + if err := exchangeLoadPayload(conn, "after join deadline"); err != nil { + t.Fatalf("gameplay beyond join budget: %v", err) + } + } + + stopListeners() + waitLoadEvent(t, tcpDone, "TCP listener shutdown") + waitLoadEvent(t, udpDone, "UDP listener and worker shutdown") + drainDone := make(chan struct{}) + go func() { + sessions.drain(loadSafetyTimeout, stopSessions) + close(drainDone) + }() + // Established sessions retain the drain context after listeners stop. + for _, conn := range connected { + if err := exchangeLoadPayload(conn, "during shutdown drain"); err != nil { + t.Fatalf("draining gameplay: %v", err) + } + _ = conn.Close() + } + waitLoadEvent(t, drainDone, "session drain") + waitLoadEvent(t, gameDone, "upstream echo shutdown") + if sessions.active.Load() != 0 || udpWake.active.Load() != 0 || apiActive.Load() != 0 { + t.Fatalf("shutdown left sessions=%d UDP wakes=%d API requests=%d", sessions.active.Load(), udpWake.active.Load(), apiActive.Load()) + } + select { + case err := <-errCh: + t.Fatalf("listener failure: %v", err) + default: + } + t.Logf("mixed load: TCP attempts=%d handoffs=%d UDP sources=%d received=%d API PATCH attempts=%d max_API_concurrency=%d UDP wake calls=%d max_UDP_wake_concurrency=%d UDP_receive_elapsed=%s handoff_mean=%s handoff_max=%s total_elapsed=%s", tcpPlayers, len(connected), len(udpClients), udpSources+2, patches.Load(), apiMaximum.Load(), udpWake.calls.Load(), udpWake.maximum.Load(), udpElapsed, totalLatency/tcpPlayers, maxLatency, time.Since(started)) +} + +func exchangeLoadPayload(conn net.Conn, payload string) error { + if _, err := io.WriteString(conn, payload); err != nil { + return fmt.Errorf("write payload: %w", err) + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil { + return fmt.Errorf("read echo: %w", err) + } + if string(got) != payload { + return fmt.Errorf("echo = %q, want %q", got, payload) + } + return nil +} diff --git a/sentinel/wake_retry.go b/sentinel/wake_retry.go new file mode 100644 index 000000000..43f7d2166 --- /dev/null +++ b/sentinel/wake_retry.go @@ -0,0 +1,93 @@ +package main + +import ( + "context" + "errors" + "io" + "math/rand/v2" + "net" + "syscall" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" +) + +const ( + wakePatchAttempts = 4 + wakeRetryInitial = 100 * time.Millisecond + wakeRetryMaximum = 800 * time.Millisecond +) + +// retryWakePatch runs inside the shared wake operation's existing context. +// Neither an attempt nor a backoff renews that operation's total budget. +func retryWakePatch(ctx context.Context, patch func(context.Context) error) error { + var lastErr error + for attempt := range wakePatchAttempts { + if err := ctx.Err(); err != nil { + return errors.Join(err, lastErr) + } + lastErr = patch(ctx) + if lastErr == nil { + return nil + } + if err := ctx.Err(); err != nil { + return errors.Join(err, lastErr) + } + if !retryableWakeError(lastErr) || attempt == wakePatchAttempts-1 { + return lastErr + } + timer := time.NewTimer(wakeRetryDelay(attempt, lastErr, rand.Float64())) + select { + case <-ctx.Done(): + timer.Stop() + // Keep both cancellation and the last underlying API error usable + // with errors.Is/As when the budget expires during backoff. + return errors.Join(ctx.Err(), lastErr) + case <-timer.C: + } + } + return lastErr +} + +func retryableWakeError(err error) bool { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + if apierrors.IsServiceUnavailable(err) || apierrors.IsTooManyRequests(err) || + apierrors.IsTimeout(err) || apierrors.IsServerTimeout(err) || apierrors.IsInternalError(err) { + return true + } + // A permanent API status must not become retryable because its cause also + // happens to implement a network error interface. + var status apierrors.APIStatus + if errors.As(err, &status) { + return false + } + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || + errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.ECONNREFUSED) || + errors.Is(err, syscall.ETIMEDOUT) || errors.Is(err, syscall.EPIPE) { + return true + } + var networkErr net.Error + return errors.As(err, &networkErr) && networkErr.Timeout() +} + +// wakeRetryDelay uses +/-25% jitter, capped after jitter so even a positive +// sample never exceeds the local 800 ms cap. Server hints are a minimum and +// may exceed that cap; retryWakePatch still waits only within ctx's budget. +func wakeRetryDelay(failure int, err error, fraction float64) time.Duration { + delay := wakeRetryInitial + for i := 0; i < failure && delay < wakeRetryMaximum; i++ { + delay *= 2 + } + delay = time.Duration(float64(delay) * (0.75 + 0.5*fraction)) + if delay > wakeRetryMaximum { + delay = wakeRetryMaximum + } + if seconds, ok := apierrors.SuggestsClientDelay(err); ok { + if suggested := time.Duration(seconds) * time.Second; suggested > delay { + delay = suggested + } + } + return delay +} diff --git a/sentinel/wake_retry_test.go b/sentinel/wake_retry_test.go new file mode 100644 index 000000000..08e24be6e --- /dev/null +++ b/sentinel/wake_retry_test.go @@ -0,0 +1,200 @@ +package main + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net" + "net/url" + "sync/atomic" + "syscall" + "testing" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + ktesting "k8s.io/client-go/testing" +) + +func TestWakeRetrySharedTransientRecovery(t *testing.T) { + w, client := newFakeGameServerWaker(t, "alpha", "games", time.Minute, nil) + started, release := make(chan struct{}), make(chan struct{}) + var calls atomic.Int32 + var firstPatch []byte + client.PrependReactor("patch", "gameservers", func(action ktesting.Action) (bool, runtime.Object, error) { + patch := action.(ktesting.PatchAction).GetPatch() + if calls.Add(1) == 1 { + firstPatch = bytes.Clone(patch) + close(started) + <-release + return true, nil, apierrors.NewServiceUnavailable("outage") + } + if !bytes.Equal(firstPatch, patch) { + t.Error("shared retry changed the annotation token or patch bytes") + } + return false, nil, nil + }) + defer func() { + select { + case <-release: + default: + close(release) + } + }() + results := make(chan error, 9) + go func() { results <- w.RequestWake(t.Context()) }() + waitLoadEvent(t, started, "first shared PATCH") + for range 8 { + waiting := &observedWaitContext{Context: t.Context(), observed: make(chan struct{})} + go func() { results <- w.RequestWake(waiting) }() + waitLoadEvent(t, waiting.observed, "shared waiter") + } + close(release) + for range 9 { + if err := receiveLoadEvent(t, results, "shared retry result"); err != nil { + t.Fatalf("shared wake: %v", err) + } + } + if err := w.RequestWake(t.Context()); err != nil { + t.Fatal(err) + } + if got := calls.Load(); got != 2 { + t.Fatalf("PATCH calls = %d, want one failed and one successful shared attempt", got) + } +} + +func TestWakeRetryAttemptLimitAndErrorIdentity(t *testing.T) { + failure := apierrors.NewServiceUnavailable("still unavailable") + calls := 0 + err := retryWakePatch(t.Context(), func(context.Context) error { + calls++ + return failure + }) + var status *apierrors.StatusError + if calls != 4 || !errors.Is(err, failure) || !errors.As(err, &status) { + t.Fatalf("calls=%d error=%v, want four attempts and original StatusError", calls, err) + } +} + +func TestWakeRetryErrorClassification(t *testing.T) { + resource := schema.GroupResource{Group: "gameplane.local", Resource: "gameservers"} + for _, tc := range []struct { + name string + err error + retry bool + }{ + {"unavailable", apierrors.NewServiceUnavailable("outage"), true}, + {"throttled", apierrors.NewTooManyRequests("busy", 0), true}, + {"timeout", apierrors.NewTimeoutError("slow", 0), true}, + {"server timeout", apierrors.NewServerTimeout(resource, "patch", 0), true}, + {"internal", apierrors.NewInternalError(errors.New("temporary")), true}, + {"EOF", io.EOF, true}, + {"unexpected EOF", io.ErrUnexpectedEOF, true}, + {"reset", fmt.Errorf("transport: %w", syscall.ECONNRESET), true}, + {"refused", &url.Error{Op: "Patch", URL: "http://localhost", Err: syscall.ECONNREFUSED}, true}, + {"timeout transport", &net.DNSError{Err: "timeout", Name: "api", IsTimeout: true}, true}, + {"forbidden", apierrors.NewForbidden(resource, "alpha", errors.New("denied")), false}, + {"unauthorized", apierrors.NewUnauthorized("denied"), false}, + {"missing", apierrors.NewNotFound(resource, "alpha"), false}, + {"invalid", apierrors.NewInvalid(schema.GroupKind{Group: resource.Group, Kind: "GameServer"}, "alpha", nil), false}, + {"bad request", apierrors.NewBadRequest("bad patch"), false}, + {"unknown", errors.New("API unavailable"), false}, + {"permanent DNS", &net.DNSError{Err: "not found", Name: "api", IsNotFound: true}, false}, + {"canceled", context.Canceled, false}, + {"deadline", context.DeadlineExceeded, false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := retryableWakeError(fmt.Errorf("PATCH: %w", tc.err)); got != tc.retry { + t.Fatalf("retryable = %v, want %v for %v", got, tc.retry, tc.err) + } + if !tc.retry { + calls := 0 + err := retryWakePatch(t.Context(), func(context.Context) error { calls++; return tc.err }) + if calls != 1 || !errors.Is(err, tc.err) { + t.Fatalf("permanent error calls=%d result=%v", calls, err) + } + } + }) + } +} + +func TestWakeRetryCancellationDuringBackoff(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + failure := apierrors.NewServiceUnavailable("outage") + calls := 0 + // Cancel synchronously when the helper evaluates Done for its backoff. + // No scheduler delay can let a second PATCH race the test's cancellation. + err := retryWakePatch(cancelBackoffContext{Context: ctx, cancel: cancel}, func(context.Context) error { calls++; return failure }) + if calls != 1 || !errors.Is(err, context.Canceled) || !errors.Is(err, failure) { + t.Fatalf("calls=%d error=%v, want cancellation and last API error without another PATCH", calls, err) + } +} + +type cancelBackoffContext struct { + context.Context + cancel context.CancelFunc +} + +func (c cancelBackoffContext) Done() <-chan struct{} { + c.cancel() + return c.Context.Done() +} + +func TestWakeRetryDeadlineAndRetryAfter(t *testing.T) { + // Longer than the first ordinary backoff, shorter than the server's hint. + ctx, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond) + defer cancel() + failure := apierrors.NewTooManyRequests("busy", 1) + calls := 0 + err := retryWakePatch(ctx, func(context.Context) error { calls++; return failure }) + if calls != 1 || !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, failure) { + t.Fatalf("calls=%d error=%v, retry-after must respect total deadline", calls, err) + } +} + +func TestWakeRetryCanceledBeforeFirstAttempt(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + err := retryWakePatch(ctx, func(context.Context) error { + t.Error("canceled caller issued a PATCH") + return nil + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("got %v, want canceled", err) + } +} + +func TestWakeRetryDelayBounds(t *testing.T) { + for _, tc := range []struct { + failure int + low, high time.Duration + }{ + {0, 75 * time.Millisecond, 125 * time.Millisecond}, + {1, 150 * time.Millisecond, 250 * time.Millisecond}, + {2, 300 * time.Millisecond, 500 * time.Millisecond}, + {3, 600 * time.Millisecond, 800 * time.Millisecond}, + {20, 600 * time.Millisecond, 800 * time.Millisecond}, + } { + for _, fraction := range []float64{0, 0.25, 0.5, 0.75, 1} { + got := wakeRetryDelay(tc.failure, nil, fraction) + if got < tc.low || got > tc.high { + t.Errorf("failure=%d jitter=%v delay=%v, want [%v,%v]", tc.failure, fraction, got, tc.low, tc.high) + } + } + if got := wakeRetryDelay(tc.failure, nil, 0); got != tc.low { + t.Errorf("failure=%d minimum=%v, want %v", tc.failure, got, tc.low) + } + if got := wakeRetryDelay(tc.failure, nil, 1); got != tc.high { + t.Errorf("failure=%d maximum=%v, want %v", tc.failure, got, tc.high) + } + } + for _, fraction := range []float64{0, 0.5, 1} { + if got := wakeRetryDelay(0, apierrors.NewTooManyRequests("busy", 2), fraction); got != 2*time.Second { + t.Errorf("server retry-after=%v, want 2s", got) + } + } +} diff --git a/specs/031-wake-retry-load/spec.md b/specs/031-wake-retry-load/spec.md new file mode 100644 index 000000000..43d340f8a --- /dev/null +++ b/specs/031-wake-retry-load/spec.md @@ -0,0 +1,48 @@ +# Wake retry and mixed-load regression (#729) + +## Intent and constraints + +A short Kubernetes API outage should not discard a genuine join's wake signal. +Retry within the existing shared wake operation and total join deadline, without +creating a retry loop per player. Preserve successful-PATCH suppression, the UDP +capacity-one worker, per-player upstream sockets, and session shutdown draining. +No new dependency, environment variable, or public configuration is needed. + +## Approach + +Implementation choice: a focused `sentinel/wake_retry.go` helper retries the +existing PATCH closure. Retrying in each handler would multiply load; replacing +the shared-operation lifecycle would unnecessarily change cancellation behavior. +Generate one annotation token and immutable merge patch per shared operation. + +Implementation choices: at most four application-level PATCH calls in the +existing five-second PATCH context (also bounded by the owner's join context). +Use exponential backoff starting at 100 ms, capped at 800 ms, with +/-25% jitter +and a final 800 ms cap. A Kubernetes suggested retry delay is a minimum, subject +to context cancellation. Recognizable transient API/transport errors are eligible; +permanent API errors and unknown errors return immediately. Preserve the last API +error for `errors.Is/As`, including cancellation during backoff. Never renew the +five-second budget per attempt. Existing client-go transport retries remain +subject to this same context; the application bound counts resource PATCH calls. + +Live waiters may still replace a canceled owning caller; canceled callers never +start another attempt. Only success starts the existing suppression interval. + +## Regression and measurement + +Write regression sources before production changes. Cover transient recovery, +attempt limits, permanent/unknown errors, cancellation and deadlines during +backoff, shared callers, retry-after, stable patch bytes, and jitter bounds. + +Add a deterministic mixed-load fixture with actual loopback TCP/UDP listeners +and an HTTP fake Kubernetes API. Hold the outage response until TCP callers and +UDP packet reception are observable, recover the API, and explicitly gate the +delayed cold start. Measure attempts, shared PATCH concurrency, TCP handoffs, +UDP receives, UDP wake concurrency, elapsed time, and handoff latency. Verify +gameplay beyond the join deadline and listener/session shutdown. Event channels +coordinate phases; time bounds are safety timeouts, not capacity claims. + +CI should print the mixed-load fixture's logs in the existing sentinel Go matrix +leg without changing test-selection policy. Runtime tests and lint run only in +CI. Local verification is limited to compilation; red/green runtime verification +is explicitly deferred to CI. diff --git a/specs/031-wake-retry-load/tasks.md b/specs/031-wake-retry-load/tasks.md new file mode 100644 index 000000000..64f0696cc --- /dev/null +++ b/specs/031-wake-retry-load/tasks.md @@ -0,0 +1,12 @@ +# Wake retry and mixed-load tasks + +- [X] T001 Read sentinel architecture/specification and CI contributor guidance; + record design and implementation choices before source edits. +- [X] T002 Write retry and mixed-load regression sources before implementation. +- [X] T003 Add bounded shared-operation retry helper and integrate RequestWake. +- [X] T004 Expose mixed-load measurements in the existing CI Go matrix. +- [X] T005 Update sentinel specification with implemented behavior and fixture scope. +- [X] T006 Compile sentinel locally without running tests or lint. +- [ ] T007 CI: run sentinel regressions, race checks, coverage and workflow lint; + red/green runtime verification is deferred to CI by repository policy. +- [ ] T008 Independent review, signed commit and upstream PR (root agent). From 09a7b122e4d2944d3251d8ec62eabc4c99c6ec25 Mon Sep 17 00:00:00 2001 From: Baylem Date: Mon, 9 Nov 2026 15:34:09 -0500 Subject: [PATCH 2/4] fix(sentinel): satisfy jitter and context checks Signed-off-by: Baylem --- sentinel/specs.md | 2 +- sentinel/wake_load_test.go | 14 ++++++++++---- sentinel/wake_retry.go | 9 +++++++-- 3 files changed, 18 insertions(+), 7 deletions(-) diff --git a/sentinel/specs.md b/sentinel/specs.md index 37564d70c..b1ab2daf7 100644 --- a/sentinel/specs.md +++ b/sentinel/specs.md @@ -258,7 +258,7 @@ Each accepted TCP connection runs as a *session* under its own context, separate ## Dependencies **Stdlib:** -- `bufio`, `bytes`, `context`, `encoding/json`, `fmt`, `io`, `log`, `math/rand/v2`, `net`, `os`, `os/signal`, `strconv`, `strings`, `sync`, `syscall`, `time`. +- `bufio`, `bytes`, `context`, `encoding/json`, `fmt`, `io`, `log`, `crypto/rand`, `net`, `os`, `os/signal`, `strconv`, `strings`, `sync`, `syscall`, `time`. **Kubernetes:** - `k8s.io/apimachinery/pkg/apis/meta/v1` — metav1 types (PatchOptions, GetOptions). diff --git a/sentinel/wake_load_test.go b/sentinel/wake_load_test.go index 582bf5043..17b9f2121 100644 --- a/sentinel/wake_load_test.go +++ b/sentinel/wake_load_test.go @@ -56,6 +56,15 @@ func (c *loadWaitContext) Done() <-chan struct{} { return c.Context.Done() } +// observeLoadWait derives the observation wrapper from the caller's context; +// cancellation and deadlines still come directly from that parent. +func observeLoadWait(ctx context.Context, joined chan struct{}) context.Context { + if joined == nil { + return ctx + } + return &loadWaitContext{Context: ctx, joined: joined} +} + type loadWakeObserver struct { wake wakeRequester joined chan struct{} @@ -74,10 +83,7 @@ func (w *loadWakeObserver) RequestWake(ctx context.Context) error { deadline, _ := ctx.Deadline() w.deadlines <- deadline } - if w.joined != nil { - ctx = &loadWaitContext{Context: ctx, joined: w.joined} - } - err := w.wake.RequestWake(ctx) + err := w.wake.RequestWake(observeLoadWait(ctx, w.joined)) w.finished <- err return err } diff --git a/sentinel/wake_retry.go b/sentinel/wake_retry.go index 43f7d2166..e3e5f7fe7 100644 --- a/sentinel/wake_retry.go +++ b/sentinel/wake_retry.go @@ -2,9 +2,10 @@ package main import ( "context" + "crypto/rand" "errors" + "fmt" "io" - "math/rand/v2" "net" "syscall" "time" @@ -36,7 +37,11 @@ func retryWakePatch(ctx context.Context, patch func(context.Context) error) erro if !retryableWakeError(lastErr) || attempt == wakePatchAttempts-1 { return lastErr } - timer := time.NewTimer(wakeRetryDelay(attempt, lastErr, rand.Float64())) + var sample [1]byte + if _, err := rand.Read(sample[:]); err != nil { + return errors.Join(lastErr, fmt.Errorf("generate wake retry jitter: %w", err)) + } + timer := time.NewTimer(wakeRetryDelay(attempt, lastErr, float64(sample[0])/255)) select { case <-ctx.Done(): timer.Stop() From 0b14f7422f9ef6fde9416f79661b9d50f3aa7f67 Mon Sep 17 00:00:00 2001 From: Baylem Date: Mon, 9 Nov 2026 23:45:27 -0500 Subject: [PATCH 3/4] docs(sentinel): address wake retry review feedback Signed-off-by: Baylem --- sentinel/specs.md | 4 ++-- specs/{031-wake-retry-load => 032-wake-retry-load}/spec.md | 2 +- specs/{031-wake-retry-load => 032-wake-retry-load}/tasks.md | 0 3 files changed, 3 insertions(+), 3 deletions(-) rename specs/{031-wake-retry-load => 032-wake-retry-load}/spec.md (96%) rename specs/{031-wake-retry-load => 032-wake-retry-load}/tasks.md (100%) diff --git a/sentinel/specs.md b/sentinel/specs.md index b1ab2daf7..9635156bc 100644 --- a/sentinel/specs.md +++ b/sentinel/specs.md @@ -69,7 +69,7 @@ Single executable module; no subdirectories or packages. - **`UDP_PACKET_WINDOW`** (optional, default `10s`): Time window for counting UDP packets. - **`UDP_MAX_SOURCES`** (optional, default `4096`): Maximum number of distinct UDP sources (keyed by `ip:port`, not just IP) to track. Older sources are evicted when this limit is reached. - **`MAX_CONNECTIONS`** (optional, default `256`): Maximum concurrent TCP connection handlers. Excess connections queue in the OS accept backlog. -- **`WAKE_PATCH_INTERVAL`** (optional, default `2s`): Minimum time between successful GameServer annotation patches. Concurrent callers share an in-flight PATCH and can cancel their own waits. If the first caller's context cancels that PATCH, live waiters retry through one new shared attempt within their own remaining budgets. API failures and the independent five-second PATCH timeout are returned to all waiters without an automatic retry. +- **`WAKE_PATCH_INTERVAL`** (optional, default `2s`): Minimum time between successful GameServer annotation patches. Concurrent callers share an in-flight wake operation, including its bounded retries, and can cancel their own waits. If the first caller's context cancels that operation, live waiters retry through one new shared attempt within their own remaining budgets. Transient API and transport failures are retried up to four PATCH calls within the same five-second budget. Terminal failures and the independent five-second PATCH timeout are returned to all waiters without starting another shared operation. - **`SHUTDOWN_DRAIN_TIMEOUT`** (optional, default `4h`, must be `>= 0`): How long in-flight sessions are kept alive after SIGTERM before they are force-closed. `0s` closes them immediately. The operator always sets it: `4h0m0s` in the ClusterIP/NodePort/LoadBalancer expose modes (with `terminationGracePeriodSeconds` = drain + 30s, so the kubelet does not SIGKILL the pod mid-drain), and `0s` in Hostport mode (grace 30s), where the terminating sentinel would otherwise keep holding the host port the game pod needs. ### Kubernetes RBAC @@ -251,7 +251,7 @@ Each accepted TCP connection runs as a *session* under its own context, separate 5. **No data duplication or loss.** The complete original stream is: `Consumed + remaining(bufio.Reader) + new-data-from-client`. Both classifiers and proxying preserve this boundary. -6. **Annotation patch coalesces bursts.** Callers share an in-flight wake operation, including its retries and final error, unless their own context expires first. Each operation makes at most four application-level PATCH calls, reusing one immutable merge patch/token. It retries recognizable transient API failures (service unavailable, throttling, API/server timeout, internal error) and transport failures (EOF, reset/refused connections, broken pipes, network timeouts). Permanent API statuses (including authorization, invalid requests and missing resources), context cancellation and unknown errors return without retry. Backoff starts at 100 ms, doubles per failure, uses +/-25% jitter and is capped at 800 ms after jitter. A Kubernetes suggested retry delay is a minimum and may exceed that cap. All PATCH calls and waits share the original five-second budget, further bounded by the owning caller's join context; client-go's own transport retries are also within that context. Errors retain the final underlying cause for `errors.Is/As`, including context expiry during backoff. Canceled callers never retry; live waiters may replace a canceled owner within their own remaining budget. Only successful operations start the `wakePatchInterval` cooldown; terminal failures release the operation for a later caller. One worker per UDP listener accepts a one-entry coalescing queue; packet reception and bounded source tracking continue while Kubernetes is stalled. Listener shutdown cancels and joins that worker. +6. **Annotation patch coalesces bursts.** Callers share an in-flight wake operation, including its retries and final error, unless their own context expires first. Each operation makes at most four application-level PATCH calls, reusing one immutable merge patch/token. It retries recognizable transient API failures (service unavailable, throttling, API/server timeout, internal error) and transport failures (EOF, reset/refused connections, broken pipes, network timeouts). Permanent API statuses (including authorization, invalid requests and missing resources), context cancellation and unknown errors return without retry. Backoff starts at 100 ms, doubles per failure, uses +/-25% jitter and is capped at 800 ms after jitter. A retry delay suggested by Kubernetes is a minimum and may exceed that cap. All PATCH calls and waits share the original five-second budget, further bounded by the owning caller's join context; client-go's own transport retries are also within that context. Errors retain the final underlying cause for `errors.Is/As`, including context expiry during backoff. Canceled callers never retry; live waiters may replace a canceled owner within their own remaining budget. Only successful operations start the `wakePatchInterval` cooldown; terminal failures release the operation for a later caller. One worker per UDP listener accepts a one-entry coalescing queue; packet reception and bounded source tracking continue while Kubernetes is stalled. Listener shutdown cancels and joins that worker. 7. **UDP sources are evicted fairly.** The least-recently-active source is evicted (LRU), not a random or oldest source, so brief bursts of scanners don't unfairly starve legitimate players trying to rejoin. diff --git a/specs/031-wake-retry-load/spec.md b/specs/032-wake-retry-load/spec.md similarity index 96% rename from specs/031-wake-retry-load/spec.md rename to specs/032-wake-retry-load/spec.md index 43d340f8a..cd3645a28 100644 --- a/specs/031-wake-retry-load/spec.md +++ b/specs/032-wake-retry-load/spec.md @@ -18,7 +18,7 @@ Generate one annotation token and immutable merge patch per shared operation. Implementation choices: at most four application-level PATCH calls in the existing five-second PATCH context (also bounded by the owner's join context). Use exponential backoff starting at 100 ms, capped at 800 ms, with +/-25% jitter -and a final 800 ms cap. A Kubernetes suggested retry delay is a minimum, subject +and a final 800 ms cap. A retry delay suggested by Kubernetes is a minimum, subject to context cancellation. Recognizable transient API/transport errors are eligible; permanent API errors and unknown errors return immediately. Preserve the last API error for `errors.Is/As`, including cancellation during backoff. Never renew the diff --git a/specs/031-wake-retry-load/tasks.md b/specs/032-wake-retry-load/tasks.md similarity index 100% rename from specs/031-wake-retry-load/tasks.md rename to specs/032-wake-retry-load/tasks.md From 8083a5c241041c146c64805d2bcca55b4d8e26cf Mon Sep 17 00:00:00 2001 From: Baylem Date: Tue, 10 Nov 2026 00:00:21 -0500 Subject: [PATCH 4/4] test(sentinel): cancel active wake on listener shutdown Signed-off-by: Baylem --- sentinel/specs.md | 1 + sentinel/wake_load_test.go | 111 +++++++++++++++++++++++++++++++++++++ 2 files changed, 112 insertions(+) diff --git a/sentinel/specs.md b/sentinel/specs.md index 9635156bc..47a2db9f5 100644 --- a/sentinel/specs.md +++ b/sentinel/specs.md @@ -299,6 +299,7 @@ Each accepted TCP connection runs as a *session* under its own context, separate - **Listener failure (F-180):** `TestRunSurfacesUDPListenFailureWhileTCPHealthy` verifies a UDP bind failure is returned promptly while a TCP listener is still healthy. - **Waker (annotation patching):** `TestWakerRequestWakeSetsAnnotation`, `TestWakerRequestWakePreservesOtherAnnotations`, `TestWakerRequestWakeCoalescesBursts` verify merge patch isolation, burst coalescing. - **Wake retries:** `wake_retry_test.go` covers transient failure then success, shared callers observing one retry sequence, immutable patch bytes, four-attempt exhaustion, permanent/unknown errors, cancellation/deadlines during backoff, retry-after and jitter bounds. Runtime red/green verification runs in CI only. +- **Active-wake shutdown:** `TestWakeListenerShutdownCancelsActivePatch` cancels a real UDP listener while its shared wake PATCH is held by an HTTP API fixture. It requires request cancellation and listener/worker completion within one bound shorter than the PATCH timeout, with no retry, active wake, retained shared operation or successful-PATCH cooldown. A separate cleanup release prevents a failed assertion from leaving the HTTP handler blocked. - **Mixed load:** `TestWakeMixedLoadOutageRecovery` uses real loopback listeners and an HTTP fake Kubernetes endpoint. A concurrent burst of 16 TCP players and 25 UDP sources joins a stalled shared PATCH; an outage response forces retry, followed by deliberately delayed game startup. Observable receive/waiter barriers assert UDP reception during the API stall, one shared PATCH at a time and one active plus one queued UDP wake. All TCP players exchange bytes after handoff, beyond their recorded join deadlines and during listener shutdown/session drain. The conditional verbose sentinel CI step logs actual attempts, handoffs, UDP receives, maximum concurrency, elapsed time and handoff latency, with race detection. These are measurements of the fixture, not a production capacity maximum; every player retains its own upstream connection. - **Wake availability:** `wake_budget_test.go` covers a hanging HTTP API, shared failure and retry, the total join deadline, worker cancellation/coalescing, successful gameplay beyond the join deadline, and receiving a 101-datagram burst while wake is stalled. CI logs elapsed receive time; this fixture does not establish a production capacity ceiling. - **Registry dispatch:** `TestRegistryProtocolDispatchMinecraftJoinWakes`, `TestRegistryProtocolDispatchMinecraftStatusPingNoWake`, `TestRegistryProtocolDispatchTerrariaJoinWakes`, `TestRegistryProtocolDispatchUnknownNoWake`, `TestRegistryProtocolReplayConsumedBytes` verify registry lookup, Classifier.Classify invocation, protocol-agnostic join/status/unknown handling, and consumed-byte replay. diff --git a/sentinel/wake_load_test.go b/sentinel/wake_load_test.go index 17b9f2121..aa848b4c8 100644 --- a/sentinel/wake_load_test.go +++ b/sentinel/wake_load_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "errors" "fmt" "io" "net" @@ -347,6 +348,116 @@ func TestWakeMixedLoadOutageRecovery(t *testing.T) { t.Logf("mixed load: TCP attempts=%d handoffs=%d UDP sources=%d received=%d API PATCH attempts=%d max_API_concurrency=%d UDP wake calls=%d max_UDP_wake_concurrency=%d UDP_receive_elapsed=%s handoff_mean=%s handoff_max=%s total_elapsed=%s", tcpPlayers, len(connected), len(udpClients), udpSources+2, patches.Load(), apiMaximum.Load(), udpWake.calls.Load(), udpWake.maximum.Load(), udpElapsed, totalLatency/tcpPlayers, maxLatency, time.Since(started)) } +func TestWakeListenerShutdownCancelsActivePatch(t *testing.T) { + listenerCtx, stopListeners := context.WithCancel(t.Context()) + requestStarted := make(chan struct{}) + requestDone := make(chan error, 1) + releaseAPI := make(chan struct{}) + var patches, apiActive atomic.Int32 + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if patches.Add(1) != 1 { + t.Error("listener shutdown issued another wake PATCH") + w.WriteHeader(http.StatusBadRequest) + return + } + apiActive.Add(1) + defer func() { + apiActive.Add(-1) + requestDone <- req.Context().Err() + }() + _, _ = io.Copy(io.Discard, req.Body) + if req.Method != http.MethodPatch { + t.Errorf("API method = %s, want PATCH", req.Method) + } + close(requestStarted) + select { + case <-req.Context().Done(): + case <-releaseAPI: + // Failure cleanup must unblock Server.Close even if cancellation regresses. + w.WriteHeader(http.StatusBadRequest) + } + })) + defer api.Close() + defer close(releaseAPI) + defer stopListeners() + client, err := dynamic.NewForConfig(&rest.Config{Host: api.URL}) + if err != nil { + t.Fatal(err) + } + shared := newWaker(client.Resource(gameServerGVR).Namespace("games"), "alpha", time.Minute) + wake := &loadWakeObserver{wake: shared, finished: make(chan error, 1)} + listener, err := (&net.ListenConfig{}).ListenPacket(listenerCtx, "udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + player, err := (&net.Dialer{}).DialContext(listenerCtx, "udp", listener.LocalAddr().String()) + if err != nil { + t.Fatal(err) + } + defer player.Close() + listenerDone := make(chan struct{}) + t.Cleanup(func() { + // Deferred cancellation, HTTP release and Server.Close run before this + // join, including when an assertion fails while the request is blocked. + select { + case <-listenerDone: + case <-time.After(loadSafetyTimeout): + t.Error("UDP listener or wake worker remained after fixture cleanup") + } + }) + errCh := make(chan error, 1) + go func() { + defer close(listenerDone) + serveUDPPackets(listenerCtx, listener, PortConfig{Protocol: "UDP"}, wake, Config{UDPThreshold: 1, UDPWindow: time.Minute, UDPMaxSources: 1}, errCh) + }() + if _, err := player.Write([]byte("wake before shutdown")); err != nil { + t.Fatal(err) + } + waitLoadEvent(t, requestStarted, "active wake PATCH") + if apiActive.Load() != 1 || wake.active.Load() != 1 { + t.Fatalf("wake not active before shutdown: API requests=%d wakes=%d", apiActive.Load(), wake.active.Load()) + } + + // One deadline covers every shutdown event and is shorter than the normal + // five-second PATCH budget, so timeout expiry cannot masquerade as cancellation. + shutdown := time.NewTimer(2 * time.Second) + defer shutdown.Stop() + stopListeners() + select { + case err := <-requestDone: + if !errors.Is(err, context.Canceled) { + t.Fatalf("HTTP PATCH ended with %v, want listener cancellation", err) + } + case <-shutdown.C: + t.Fatal("HTTP PATCH did not stop promptly after listener cancellation") + } + select { + case err := <-wake.finished: + if !errors.Is(err, context.Canceled) { + t.Fatalf("wake ended with %v, want listener cancellation", err) + } + case <-shutdown.C: + t.Fatal("wake did not stop promptly after listener cancellation") + } + select { + case <-listenerDone: + case <-shutdown.C: + t.Fatal("UDP listener and worker did not stop promptly") + } + shared.mu.Lock() + inFlight, lastPatch := shared.inFlight, shared.lastPatch + shared.mu.Unlock() + if patches.Load() != 1 || apiActive.Load() != 0 || wake.active.Load() != 0 || inFlight != nil || !lastPatch.IsZero() { + t.Fatalf("shutdown left patches=%d API requests=%d wakes=%d inFlight=%v lastPatch=%v", patches.Load(), apiActive.Load(), wake.active.Load(), inFlight != nil, lastPatch) + } + select { + case err := <-errCh: + t.Fatalf("listener shutdown reported failure: %v", err) + default: + } +} + func exchangeLoadPayload(conn net.Conn, payload string) error { if _, err := io.WriteString(conn, payload); err != nil { return fmt.Errorf("write payload: %w", err)