From e6c8c552a436e907f70dfd69f664bec982a5ec42 Mon Sep 17 00:00:00 2001 From: Jens Neuse Date: Fri, 17 Jul 2026 21:20:09 +0200 Subject: [PATCH 1/3] docs: design ordered concurrent subscription updates --- ...-concurrent-subscription-updates-design.md | 115 ++++++++++++++++++ 1 file changed, 115 insertions(+) create mode 100644 docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md diff --git a/docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md b/docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md new file mode 100644 index 0000000000..5cd2a7ff81 --- /dev/null +++ b/docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md @@ -0,0 +1,115 @@ +# Ordered Concurrent Subscription Updates Design + +## Goal + +Restore concurrent resolution for per-subscriber subscription updates without allowing updates for one subscriber, filter errors, or terminal frames to be delivered out of order. + +The change must be local to `graphql-go-tools`, retain the synchronous completion semantics of `SubscriptionUpdater.UpdateSubscription`, and avoid reinstating one long-lived worker goroutine per subscription. + +## Background + +`subscriptionUpdater.UpdateSubscription` currently holds one trigger-wide mutex while it filters, resolves, fetches subgraph data, writes, and flushes a subscriber update. A stream hook fan-out invokes this method concurrently for different subscribers, but the mutex serializes the complete operation. Identical subgraph fetches therefore never overlap and cannot share an in-flight response. + +Simply releasing the mutex before resolution is insufficient. If ordering is established later with a mutex on `subscriptionState`, concurrent calls can reach that mutex in a different order from the order in which the updater admitted them. Filter errors and lifecycle operations can also bypass that mutex. + +## Required Behavior + +1. Updates admitted for distinct `SubscriptionIdentifier` values may resolve concurrently. +2. Updates admitted for the same `SubscriptionIdentifier` resolve and write in updater admission order. +3. `UpdateSubscription` returns only after its update has resolved or has been skipped. +4. Operations admitted through `subscriptionUpdater`—broadcast updates, explicit heartbeats, completion, terminal errors, per-subscription close, and `Done`—do not overtake already admitted per-subscriber updates. +5. The first completion or terminal error wins. Once admitted, later data, heartbeat, completion, and error frames are suppressed so nothing is delivered after a terminal frame. +6. `subscriptionUpdater.Done` waits for admitted updates before detaching the trigger. Direct resolver unsubscribe and shutdown paths retain their existing removal-based behavior described below. +7. A panic in one admitted update releases its lane and in-flight accounting before propagating, so later lifecycle operations cannot deadlock. + +## Architecture + +### Admission and per-subscriber lanes + +`subscriptionUpdater` remains the ordering boundary. While holding its existing lifecycle mutex, `UpdateSubscription` records a new lane entry for the target subscription and increments an in-flight wait group. Recording the entry before releasing the mutex gives each call a deterministic admission position relative to every other updater operation. + +Each lane entry contains a completion channel and references the previous entry's completion channel. After admission, the call releases the lifecycle mutex, waits for only its predecessor on the same lane, and invokes `handleUpdateSubscription`. Calls for other identifiers have independent predecessors and can proceed immediately. + +`UpdateSubscription` itself does not spawn a goroutine. Concurrency comes from callers that already invoke different subscriber updates concurrently, and each caller continues to block until its own resolution completes. + +### Lane bookkeeping + +A separate lane mutex protects the map of current lane tails. The lock order is lifecycle mutex followed by lane mutex. Completion only needs the lane mutex, so lifecycle operations may safely hold the lifecycle mutex while waiting for all in-flight entries. + +When an entry finishes, it closes its completion channel, removes itself from the tail map if it is still the current tail, and decrements the in-flight wait group. Cleanup runs in a defer so it also executes on panic. + +Removing completed tails prevents the map from growing when subscriptions churn on a long-lived trigger. + +### Lifecycle barriers + +The updater's existing mutex continues to serialize admission of all public operations. Operations that previously relied on holding that mutex across synchronous resolution first wait for the per-subscriber in-flight wait group while still holding the mutex: + +- `Update` waits, then performs its synchronous broadcast fan-out. +- `Heartbeat` waits, then sends heartbeats. +- `Complete` and `Error` mark the updater terminal, wait, then deliver the terminal signal. +- `Done` marks the updater done, waits, then detaches the trigger. +- `CloseSubscription` waits before removing the subscription. + +Holding the lifecycle mutex while waiting is safe because update completion never reacquires it. It also prevents a concurrent `UpdateSubscription` from incrementing the wait group while a lifecycle operation is waiting. + +The wait-group invariant is: every positive `Add` occurs while holding the lifecycle mutex and before publishing the lane entry; every operation that calls `Wait` holds the same mutex. Consequently, no `Add` can race with a zero-count `Wait` or begin until that wait and its lifecycle operation have returned. + +The terminal state is distinct from `done`: `Complete` and `Error` stop new data admissions but leave cleanup to the required final `Done` call. + +`CloseSubscription` deliberately waits for all currently admitted per-subscriber updates rather than only the target lane. This preserves the existing trigger-wide ordering of public updater calls and keeps this short-term change small. Closing a subscription is exceptional, so an unrelated slow subscriber delaying close is accepted here; a keyed lifecycle barrier belongs in a larger rewrite. + +### Terminal operation matrix + +- The first `Complete` or `Error` marks the updater terminal, waits for admitted updates, and emits its terminal frame. +- Later `Update`, `UpdateSubscription`, `Heartbeat`, `Complete`, and `Error` calls are no-ops. +- `CloseSubscription` remains allowed before `Done`; it performs subscription cleanup but cannot emit data. +- `Done` remains required and performs trigger cleanup exactly once. +- After `Done`, all updater methods except the read-only `Subscriptions` call are no-ops. + +Each `subscriptionState` also records whether its downstream terminal frame has been written. `complete` and `error` set this state while holding `writeMu`; data, formatted errors, and heartbeats check it while holding the same mutex before writing. This closes the window between an updater terminal call and `Done`, including writes originating from resolver paths that do not pass through the updater. + +### Resolver paths outside the updater + +The resolver heartbeat loop calls `heartbeatTriggerSubscriptions` directly. It does not participate in updater admission and may run while an update is fetching, as it does today. `subscriptionState.writeMu` continues to make the eventual heartbeat or data write atomic, and the per-subscription terminal state prevents a heartbeat after a terminal frame. A heartbeat may still precede an update whose fetch is in flight; heartbeat-versus-in-flight-data ordering before termination is not part of this fix's data-ordering guarantee. + +Direct unsubscribe, resolver shutdown, startup failure, and unsubscribe-on-flush-failure also bypass the updater barrier. They retain the existing removal protocol: removal marks `subscriptionState.removed`, detaches it from resolver indexes, and prevents an in-flight resolution from writing after removal. They must not wait on the updater in-flight group because unsubscribe-on-flush-failure can originate inside an admitted update and would deadlock by waiting for itself. + +## Ordering Guarantee and Limitation + +The engine guarantees updater admission order. It cannot reconstruct original broker order if an upstream hook runner invokes a newer event before an older event. In particular, a router timeout that abandons an older hook invocation can allow the next event's hook to call `UpdateSubscription` first. + +Guaranteeing broker order across such timeouts requires the router to assign event sequence information before hook execution and carry it through to delivery, or to prevent overlapping event batches. That router-level change is outside this short-term fix. + +## Error Handling + +Filtering, resolution errors, authorization errors, and writes all execute inside the admitted lane entry, so they retain the same per-subscriber order as successful data frames. Existing error formatting and unsubscribe-on-flush-failure behavior remain unchanged. + +While a queued entry waits for its lane predecessor, it also observes the updater context. If that context is cancelled first, the entry skips resolution and completes its lane bookkeeping, unblocking its successor. Once an entry starts resolving, cancellation continues to be handled by the existing resolution timeout and subscription contexts. + +An admitted entry always completes its lane bookkeeping if resolution returns early or panics. A panic is not swallowed: cleanup runs and the panic continues to the caller, preserving existing panic behavior while preventing a stuck lane or lifecycle deadlock. + +## Anonymous Regression Coverage + +Tests will use generic subscription identifiers, event payloads, and a blocking in-memory data source. No customer names, schemas, headers, message-provider details, or measured production values will be included. + +The tests will verify: + +1. Two subscribers on one trigger both enter their subgraph loads before either load is released. This is the red regression test against the current trigger-wide serialization. +2. Two updates admitted for one subscriber cannot resolve or write in reverse order. +3. A filter error and a successful update for one subscriber retain their admission order. +4. The first terminal operation waits for admitted updates, later competing terminal operations are ignored, and later data is not admitted. +5. Periodic and explicit heartbeats cannot write after a terminal frame. +6. `Update`, `Heartbeat`, `Complete`, `Error`, `Done`, and `CloseSubscription` each wait for already admitted per-subscriber updates; dedicated assertions verify `Done` detachment and `CloseSubscription` removal. +7. A queued update cleans up promptly when the updater context is cancelled. +8. A panic releases its lane and in-flight accounting before propagating. +9. Direct unsubscribe during an in-flight update prevents the late write without waiting on the updater barrier. +10. Completed lane tails are removed after both single and overlapping same-subscriber updates. +11. Focused tests pass under the Go race detector. + +## Non-Goals + +- Rewriting the router subscription pipeline. +- Restoring per-subscription worker goroutines. +- Deduplicating non-identical subgraph requests. +- Guaranteeing broker order after the router has already reordered hook invocations. +- Changing public interfaces or configuration. From 0f27d41b5e7411cd4392434ab271244437eaca30 Mon Sep 17 00:00:00 2001 From: Jens Neuse Date: Fri, 17 Jul 2026 21:49:51 +0200 Subject: [PATCH 2/3] docs: plan ordered concurrent subscription updates --- ...ordered-concurrent-subscription-updates.md | 382 ++++++++++++++++++ 1 file changed, 382 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-17-ordered-concurrent-subscription-updates.md diff --git a/docs/superpowers/plans/2026-07-17-ordered-concurrent-subscription-updates.md b/docs/superpowers/plans/2026-07-17-ordered-concurrent-subscription-updates.md new file mode 100644 index 0000000000..b3a39b3d11 --- /dev/null +++ b/docs/superpowers/plans/2026-07-17-ordered-concurrent-subscription-updates.md @@ -0,0 +1,382 @@ +# Ordered Concurrent Subscription Updates Implementation Plan + +> **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Resolve per-subscriber updates concurrently across subscribers while preserving updater admission order within each subscriber and preventing writes after terminal frames. + +**Architecture:** Add keyed predecessor/completion lanes at the `subscriptionUpdater` admission boundary. Existing callers remain synchronous, but concurrent callers for different subscription identifiers no longer share the trigger-wide mutex during resolution. Lifecycle methods use an in-flight barrier, while `subscriptionState` gains terminal-frame protection for write paths that bypass the updater. + +**Tech Stack:** Go, `sync.Mutex`, `sync.WaitGroup`, channels, existing resolver subscription tests, Go race detector. + +**Design:** `docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md` + +--- + +## File Structure + +- Create `v2/pkg/engine/resolve/subscription_updater_test.go`: focused anonymous regression tests and small test-only harnesses for updater admission, ordering, lifecycle, cancellation, and panic behavior. +- Modify `v2/pkg/engine/resolve/resolve.go`: keyed update lanes, updater lifecycle barriers, terminal updater state, and per-subscription terminal write protection. + +No public interface, router code, configuration, or customer-specific fixture changes are required. + +Command convention: run `go test` and final verification commands from `v2/`. Run `git`, implementation `gofmt`, and commit commands that use `v2/...` paths from the repository root. + +## Chunk 1: Concurrent and Ordered Admission + +### Task 1: Add the anonymous cross-subscriber regression test + +**Files:** +- Create: `v2/pkg/engine/resolve/subscription_updater_test.go` + +- [ ] **Step 1: Add a blocking test data source and resolver harness** + +Create a `concurrentSubscriptionDataSource` that counts entries into `Load`, closes `allStarted` after the configured number of calls, and blocks on `release`. Implement both `Load` and `LoadWithFiles` using the same helper. Its `Release` method must use `sync.Once`, making explicit release, deferred failure cleanup, and harness cleanup safely idempotent. + +Create `newSubscriptionUpdaterHarness(t, count, dataSource)` which: + +1. Creates a real resolver with a long heartbeat interval and gives each subscription context `ExecutionOptions.DisableSubgraphRequestDeduplication = true`, ensuring the test observes two physical loads rather than one leader and one single-flight follower. +2. Builds one trigger and one `subscriptionUpdater`. +3. Registers `count` anonymous subscription states under distinct identifiers. +4. Gives every state a response whose event post-processing selects `data`, whose `SingleFetch` has static input `{}`, query operation metadata, data-source post-processing that selects `data`, and whose rendered object projects both event field `value` and fetched field `resolved`. +5. Registers the trigger and subscriptions in the resolver indexes. +6. Returns the updater, identifiers, recorders, and a cleanup function. + +Use generic payloads such as `{"data":{"value":"event"}}` and `{"data":{"resolved":"value"}}`. + +The cleanup function must release every blocking primitive, cancel the resolver, and wait with a timeout for all started updater calls. Recorder assertions must require the rendered output `{"data":{"value":"event","resolved":"value"}}`, proving the real fetch tree and render path ran. + +- [ ] **Step 2: Write the failing concurrency test** + +Add: + +```go +func TestSubscriptionUpdater_UpdateSubscription_ResolvesDistinctSubscribersConcurrently(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(2) + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 2, dataSource) + defer cleanup() + defer dataSource.Release() + + var wg sync.WaitGroup + for _, id := range ids { + wg.Go(func() { + updater.UpdateSubscription(id, []byte(`{"data":{"value":"event"}}`)) + }) + } + + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + t.Fatal("distinct subscribers did not enter their fetches concurrently") + } + + dataSource.Release() + wg.Wait() +} +``` + +- [ ] **Step 3: Run the test and verify RED** + +Run: + +```bash +go test ./pkg/engine/resolve -run TestSubscriptionUpdater_UpdateSubscription_ResolvesDistinctSubscribersConcurrently -count=1 +``` + +from `v2/`. + +Expected: FAIL with `distinct subscribers did not enter their fetches concurrently`. The defer must release the blocked first fetch so no goroutine leaks. + +### Task 2: Implement keyed admission lanes + +**Files:** +- Modify: `v2/pkg/engine/resolve/resolve.go` near `subscriptionUpdater` +- Test: `v2/pkg/engine/resolve/subscription_updater_test.go` + +- [ ] **Step 1: Add lane state to `subscriptionUpdater`** + +Add the following private state: + +```go +updateMu sync.Mutex +updateTails map[SubscriptionIdentifier]chan struct{} +updateWG sync.WaitGroup +terminal bool +``` + +`mu` remains the lifecycle/admission mutex. `updateMu` only protects tail publication and completion cleanup. + +- [ ] **Step 2: Add admission and completion helpers** + +Implement helpers equivalent to: + +```go +func (s *subscriptionUpdater) admitSubscriptionUpdate(id SubscriptionIdentifier) (previous <-chan struct{}, current chan struct{}) { + s.updateMu.Lock() + defer s.updateMu.Unlock() + if s.updateTails == nil { + s.updateTails = make(map[SubscriptionIdentifier]chan struct{}) + } + previous = s.updateTails[id] + current = make(chan struct{}) + s.updateTails[id] = current + s.updateWG.Add(1) + return previous, current +} + +func (s *subscriptionUpdater) finishSubscriptionUpdate(id SubscriptionIdentifier, current chan struct{}) { + s.updateMu.Lock() + close(current) + if s.updateTails[id] == current { + delete(s.updateTails, id) + } + s.updateMu.Unlock() + s.updateWG.Done() +} +``` + +Document the lock order `mu -> updateMu`, and that `finishSubscriptionUpdate` never acquires `mu`. + +- [ ] **Step 3: Change `UpdateSubscription` to admit under `mu` and resolve outside it** + +The method must: + +1. Lock `mu`. +2. Return if done, terminal, or the updater context is cancelled. +3. Admit the keyed entry and unlock `mu`. +4. Defer lane completion immediately. +5. Wait for the same-subscription predecessor, also selecting on updater-context cancellation. +6. Recheck updater-context cancellation. +7. Call `handleUpdateSubscription` synchronously. + +Do not create a goroutine in the engine and do not add a mutex to `subscriptionState` for resolution. + +- [ ] **Step 4: Run the focused test and verify GREEN** + +Run the Task 1 command again. + +Expected: PASS; both data-source calls enter before release. + +### Task 3: Lock down same-subscriber order and lane cleanup + +**Files:** +- Modify: `v2/pkg/engine/resolve/subscription_updater_test.go` +- Modify if needed: `v2/pkg/engine/resolve/resolve.go` + +- [ ] **Step 1: Add a same-subscriber FIFO test** + +Use a first fetch that blocks and signals entry. Once it is active, lock `updateMu` and capture its current `updateTails[id]` channel. Admit a second update for the same identifier, then poll under `updateMu` with a one-second deadline until `updateTails[id]` differs from the captured first tail; this proves the second call was admitted rather than merely unscheduled. Assert the second fetch does not begin before the first is released, then assert recorder messages are exactly `first` followed by `second`. + +Configure the response so `value` is projected from each event payload while the fetched `resolved` field is constant. Use payloads `{"data":{"value":"first"}}` and `{"data":{"value":"second"}}` so FIFO is visible at the writer. + +- [ ] **Step 2: Add lane-tail cleanup assertions** + +After one update and after two overlapping same-subscriber updates complete, lock `updateMu` in the test and require `updateTails` to be empty. + +- [ ] **Step 3: Add cancellation and panic cleanup tests** + +For cancellation, queue a second same-subscriber update behind a blocked first update, prove it was admitted by observing the replaced lane tail, cancel the updater context, and require the queued call to return within one second. Retain a deferred idempotent release for failure paths, but explicitly call `Release()` before joining the first call with a one-second deadline. Only after that join may the test verify final lane and in-flight cleanup. + +For panic, use a test data source that panics from `Load`. Start a goroutine that invokes `UpdateSubscription` inside a closure with a deferred recover and sends the recovered value over a buffered channel; require the unchanged value within one second. Start a separate waiter goroutine for `updateWG.Wait()` and require it to return within one second before asserting the lane-tail map is empty. No test may call `Wait` directly on the test goroutine without a timeout. + +- [ ] **Step 4: Add filter-error ordering coverage** + +Add a synchronized test-only `VariableRenderer` that emits a matching static JSON value on its first invocation and returns a sentinel error on its second invocation. Filter templates receive the subscription context rather than event data, so invocation count—not payload inspection—selects the error. Use a chronological subscription writer whose single log records flushed data, formatted errors, terminal frames, and heartbeats in actual call order. + +Admit a blocked successful update first, prove a filter-error update for the same subscriber has replaced the lane tail, and require the log to remain empty before releasing the first update. After release, require the log to contain the successful data frame followed by the formatted filter error. + +- [ ] **Step 5: Run the admission test group** + +Run: + +```bash +go test ./pkg/engine/resolve -run 'TestSubscriptionUpdater_UpdateSubscription_' -count=1 +``` + +Expected: PASS. + +Name the FIFO, cancellation, panic, cleanup, and filter-error tests with the exact `TestSubscriptionUpdater_UpdateSubscription_` prefix so this command selects all of them. + +- [ ] **Step 6: Commit the admission behavior** + +```bash +gofmt -w v2/pkg/engine/resolve/resolve.go v2/pkg/engine/resolve/subscription_updater_test.go +git add v2/pkg/engine/resolve/resolve.go v2/pkg/engine/resolve/subscription_updater_test.go +git commit -m "fix(resolve): resolve subscriber updates concurrently" +``` + +## Chunk 2: Lifecycle and Terminal Ordering + +### Task 4: Add updater lifecycle barriers + +**Files:** +- Modify: `v2/pkg/engine/resolve/resolve.go` +- Modify: `v2/pkg/engine/resolve/subscription_updater_test.go` + +- [ ] **Step 1: Add failing lifecycle-order tests** + +Create table-driven subtests with isolated harnesses. Each subtest admits and blocks a per-subscriber update, defers the data source's idempotent release for failure cleanup, and exercises one public updater operation from another goroutine: + +- `Update` +- `Heartbeat` +- `Complete` +- `Error` +- `CloseSubscription` +- `Done` + +Before releasing the update, assert both that the lifecycle call has not returned and that its operation-specific effect has not occurred: + +- `Update`: the broadcast's second fetch has not started. +- `Heartbeat`: no heartbeat is present in the chronological writer log. +- `Complete` and `Error`: no terminal frame is present. +- `CloseSubscription`: the target remains registered. +- `Done`: the trigger remains registered and the subscription completion channel remains open. + +Explicitly release the blocked update, then require both the update goroutine and lifecycle goroutine to finish using bounded selects with a shared multi-second test deadline. After release, require the operation-specific effect. A lifecycle test must never use an unbounded `Wait` or rely only on a goroutine-return assertion. + +For the heartbeat subtest, set `subscriptionState.heartbeat = true` and set the test resolver's `heartbeatInterval` to zero after construction. This makes the subscription eligible even though the completed update refreshes `lastWriteTime`, so the post-release assertion must observe exactly one heartbeat. + +Add table-driven first-terminal-wins cases for both `Complete`-first and `Error`-first, each with at least two subscriptions. After the winning terminal call, calls to `Error`, `Complete`, `Update`, `UpdateSubscription`, and `Heartbeat` must not change either writer. `CloseSubscription` must remove one target after terminal state while leaving the trigger and second subscription registered. `Done` must then detach the trigger and complete the remaining subscription. After `Done`, invoke every mutating updater method and require all resolver and writer state to remain unchanged; `Subscriptions` remains read-only and callable. + +Name these tests with the exact prefixes `TestSubscriptionUpdater_Lifecycle_` and `TestSubscriptionUpdater_Terminal_`. + +- [ ] **Step 2: Verify the lifecycle tests fail against the partial implementation** + +Run: + +```bash +go test ./pkg/engine/resolve -run 'TestSubscriptionUpdater_(Lifecycle|Terminal)' -count=1 +``` + +Expected: FAIL because methods other than `UpdateSubscription` do not yet wait on `updateWG` or enforce terminal state. + +- [ ] **Step 3: Add lifecycle barriers** + +While holding `subscriptionUpdater.mu`, call `updateWG.Wait()` before executing `Update`, `Heartbeat`, `Complete`, `Error`, `CloseSubscription`, or `Done`. + +Apply these state rules: + +- `Update`, `UpdateSubscription`, and `Heartbeat` return when `done`, `terminal`, or context-cancelled. +- `Complete` and `Error` return when `done`, `terminal`, or context-cancelled; the first accepted call sets `terminal = true` before waiting. +- `CloseSubscription` is allowed while terminal but returns when done or context-cancelled. +- `Done` sets `done = true` before waiting and always remains available after terminal. + +Add a comment recording the `WaitGroup` invariant: every `Add` occurs under `mu`, and every `Wait` holds `mu`, so `Add` cannot race a zero-count `Wait`. + +- [ ] **Step 4: Run the lifecycle tests and verify GREEN** + +Run the Step 2 command again. + +Expected: PASS. + +### Task 5: Prevent writes after terminal frames + +**Files:** +- Modify: `v2/pkg/engine/resolve/resolve.go` near `subscriptionState` and its write helpers +- Modify: `v2/pkg/engine/resolve/subscription_updater_test.go` + +- [ ] **Step 1: Add failing terminal-writer tests** + +Test `subscriptionState` directly with a recording writer: + +1. Call `complete`, then require `sendHeartbeat` and `writeError` to produce no frames. +2. Call `error`, then require the same suppression. +3. Call completion/error repeatedly and require exactly one terminal frame. +4. Exercise a real late data path: start `executeSubscriptionUpdate` with its data source blocked, deliver `complete` or `error` while the load is blocked, release the load, join it with a bounded select, and require no data frame after the terminal frame. This directly tests the terminal check in the resolver's final write section. +5. Mark a heartbeat-enabled subscription terminal, then invoke `resolver.heartbeatTriggerSubscriptions` directly and require no heartbeat. This covers the periodic resolver path that bypasses `subscriptionUpdater`. +6. Start direct unsubscribe during an in-flight resolve and require that removal returns without waiting for the updater barrier and that the late resolve does not write. Use deferred idempotent release for failure cleanup, explicit release on the success path, and bounded joins. + +Name state tests with `TestSubscriptionState_Terminal_`, the periodic path `TestResolver_TerminalHeartbeat_`, and unsubscribe `TestSubscriptionUpdater_DirectUnsubscribe_`. + +- [ ] **Step 2: Verify RED** + +Run: + +```bash +go test ./pkg/engine/resolve -run 'TestSubscriptionState_Terminal_|TestResolver_TerminalHeartbeat_|TestSubscriptionUpdater_DirectUnsubscribe_' -count=1 +``` + +Expected: FAIL because `subscriptionState` currently tracks only removal, not terminal delivery. + +- [ ] **Step 3: Add per-subscription terminal state** + +Add `terminal bool` to `subscriptionState`, guarded by `writeMu`. + +- `complete` and `error` check `removed || terminal`, set terminal, then write the terminal frame. +- `writeError` and `sendHeartbeat` return without writing when removed or terminal. +- The final write section of `executeSubscriptionUpdate` returns when removed or terminal. + +Keep all terminal reads and writes under `writeMu`; no new atomic or lock is needed. + +- [ ] **Step 4: Run terminal and lifecycle tests** + +Run the Step 2 command, followed by the Task 4 lifecycle command. + +Expected: PASS. + +- [ ] **Step 5: Commit lifecycle and terminal ordering** + +```bash +gofmt -w v2/pkg/engine/resolve/resolve.go v2/pkg/engine/resolve/subscription_updater_test.go +git add v2/pkg/engine/resolve/resolve.go v2/pkg/engine/resolve/subscription_updater_test.go +git commit -m "fix(resolve): preserve subscription update ordering" +``` + +### Task 6: Full verification + +**Files:** +- Verify: `v2/pkg/engine/resolve/resolve.go` +- Verify: `v2/pkg/engine/resolve/subscription_updater_test.go` + +- [ ] **Step 1: Check formatting and inspect all task changes** + +Run: + +```bash +test -z "$(gofmt -d pkg/engine/resolve/resolve.go pkg/engine/resolve/subscription_updater_test.go)" +task_base=$(git merge-base HEAD origin/master) +git diff --check "$task_base"..HEAD +git diff --stat "$task_base"..HEAD +``` + +Expected: no formatting or whitespace errors; only the planned resolver, test, spec, and plan files differ from the original branch base. + +- [ ] **Step 2: Run focused tests repeatedly** + +```bash +go test ./pkg/engine/resolve -run 'TestSubscriptionUpdater_|TestSubscriptionState_Terminal_|TestResolver_TerminalHeartbeat_' -count=20 +``` + +Expected: PASS across all repetitions. + +- [ ] **Step 3: Run the focused race suite** + +```bash +go test -race ./pkg/engine/resolve -run 'TestSubscriptionUpdater_|TestSubscriptionState_Terminal_|TestResolver_TerminalHeartbeat_' -count=1 +``` + +Expected: PASS with no race reports. + +- [ ] **Step 4: Run the complete resolve package** + +```bash +go test ./pkg/engine/resolve -count=1 +``` + +Expected: PASS. + +- [ ] **Step 5: Run the complete resolve package under the race detector** + +```bash +go test -race ./pkg/engine/resolve -count=1 +``` + +Expected: PASS with no race reports. + +- [ ] **Step 6: Review repository status and commits** + +```bash +git status --short +git log --oneline -4 +``` + +Expected: a clean worktree with the design commit, plan commit, and two implementation commits. Do not push or mutate any remote service without separate explicit approval. From 7ef3a87ce17bf6344ead7ef2b500c5a92abeae61 Mon Sep 17 00:00:00 2001 From: Jens Neuse Date: Sat, 18 Jul 2026 00:13:01 +0200 Subject: [PATCH 3/3] fix(resolve): resolve subscription updates concurrently in order --- ...-concurrent-subscription-updates-design.md | 10 +- v2/pkg/engine/resolve/resolve.go | 286 ++- .../resolve/subscription_updater_test.go | 2112 +++++++++++++++++ 3 files changed, 2337 insertions(+), 71 deletions(-) create mode 100644 v2/pkg/engine/resolve/subscription_updater_test.go diff --git a/docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md b/docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md index 5cd2a7ff81..3f4b5e28a7 100644 --- a/docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md +++ b/docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md @@ -21,6 +21,7 @@ Simply releasing the mutex before resolution is insufficient. If ordering is est 5. The first completion or terminal error wins. Once admitted, later data, heartbeat, completion, and error frames are suppressed so nothing is delivered after a terminal frame. 6. `subscriptionUpdater.Done` waits for admitted updates before detaching the trigger. Direct resolver unsubscribe and shutdown paths retain their existing removal-based behavior described below. 7. A panic in one admitted update releases its lane and in-flight accounting before propagating, so later lifecycle operations cannot deadlock. +8. Once `Complete`, `Error`, or `Done` begins, the trigger rejects new subscribers. A subscriber registered before `Complete` or `Error` is included in terminal-frame delivery; one registered before `Done` is included in cleanup and detachment. A subscriber arriving after any of those transitions is not attached. ## Architecture @@ -56,6 +57,8 @@ The wait-group invariant is: every positive `Add` occurs while holding the lifec The terminal state is distinct from `done`: `Complete` and `Error` stop new data admissions but leave cleanup to the required final `Done` call. +Terminal operations also atomically publish that the trigger is no longer accepting subscribers before taking a terminal snapshot or detaching it. Existing-trigger subscription admission observes that state while holding the trigger subscription lock at the registration linearization point. Admission that wins first is visible to the later terminal snapshot; terminal publication that wins first causes the add to return an internal error without changing trigger or resolver indexes, counters, or startup hooks. This observation does not acquire the updater lifecycle mutex while the resolver mutex is held, preserving lock order. + `CloseSubscription` deliberately waits for all currently admitted per-subscriber updates rather than only the target lane. This preserves the existing trigger-wide ordering of public updater calls and keeps this short-term change small. Closing a subscription is exceptional, so an unrelated slow subscriber delaying close is accepted here; a keyed lifecycle barrier belongs in a larger rewrite. ### Terminal operation matrix @@ -74,6 +77,8 @@ The resolver heartbeat loop calls `heartbeatTriggerSubscriptions` directly. It d Direct unsubscribe, resolver shutdown, startup failure, and unsubscribe-on-flush-failure also bypass the updater barrier. They retain the existing removal protocol: removal marks `subscriptionState.removed`, detaches it from resolver indexes, and prevents an in-flight resolution from writing after removal. They must not wait on the updater in-flight group because unsubscribe-on-flush-failure can originate inside an admitted update and would deadlock by waiting for itself. +The post-load write phase holds `subscriptionState.writeMu` only inside a defer-unlock critical section. Resolve, writer, flush, and error-formatter panics therefore release the write lock while continuing to propagate. Resolve-arena ownership uses an idempotent release function with a deferred panic fallback, while normal paths return the arena before error formatting or flushing so slow downstream writers do not retain it. Flush-failure unsubscribe remains outside the write lock. + ## Ordering Guarantee and Limitation The engine guarantees updater admission order. It cannot reconstruct original broker order if an upstream hook runner invokes a newer event before an older event. In particular, a router timeout that abandons an older hook invocation can allow the next event's hook to call `UpdateSubscription` first. @@ -104,7 +109,10 @@ The tests will verify: 8. A panic releases its lane and in-flight accounting before propagating. 9. Direct unsubscribe during an in-flight update prevents the late write without waiting on the updater barrier. 10. Completed lane tails are removed after both single and overlapping same-subscriber updates. -11. Focused tests pass under the Go race detector. +11. Late subscriber admission is rejected after `Complete`, `Error`, or `Done` publication without launching startup hooks or changing indexes and counters. Existing subscribers receive the terminal frame for `Complete` or `Error`, while `Done` detaches them and completes cleanup. +12. Panics from writing, flushing, or formatting a writer error release the write critical section so `Done` can complete. +13. Concurrent subscribers with identical query fetches share one physical single-flight load while rendering their own logical responses. +14. Focused tests pass under the Go race detector. ## Non-Goals diff --git a/v2/pkg/engine/resolve/resolve.go b/v2/pkg/engine/resolve/resolve.go index 8af64b561f..6292406b1e 100644 --- a/v2/pkg/engine/resolve/resolve.go +++ b/v2/pkg/engine/resolve/resolve.go @@ -29,6 +29,11 @@ const ( DefaultHeartbeatInterval = 5 * time.Second ) +// ErrSubscriptionTriggerTerminating indicates that a matching subscription +// trigger is draining and cannot accept another subscriber. Callers may retry +// after the terminating trigger has been detached. +var ErrSubscriptionTriggerTerminating = errors.New("subscription trigger is no longer accepting subscriptions") + // ConnectionID identifies a client connection for subscription routing. type ConnectionID int64 @@ -58,8 +63,10 @@ type AsyncErrorWriter interface { } // Resolver manages GraphQL subscriptions using a mutex-protected trigger registry. -// All trigger/subscription state is guarded by mu. Long-running I/O (writes to clients) -// is performed outside the lock using a snapshot-and-release pattern. +// Resolver.mu protects the registry and subscription indexes. Each trigger, +// subscription updater, and subscription writer has narrower synchronization for +// membership, lifecycle/admission, and downstream writes respectively, as documented +// on those types. Long-running I/O is performed outside Resolver.mu. type Resolver struct { ctx context.Context options ResolverOptions @@ -308,7 +315,8 @@ func New(ctx context.Context, options ResolverOptions) *Resolver { resolver.maxConcurrency <- struct{}{} } - go resolver.heartbeatLoop() + heartbeatTicker := time.NewTicker(resolver.heartbeatInterval) + go resolver.heartbeatLoop(heartbeatTicker) context.AfterFunc(resolver.ctx, func() { resolver.shutdownResolver() }) @@ -919,8 +927,10 @@ type subscriptionState struct { heartbeat bool completed chan struct{} // writeMu protects all writes to writer (Complete, Error, Write, Flush, Heartbeat). - // Paired with the removed atomic to prevent writes after removal. + // Paired with removed and terminal to prevent writes after removal or a terminal frame. writeMu sync.Mutex + // terminal is guarded by writeMu and suppresses all writes after the first Complete or Error. + terminal bool // removed guards against writes after the subscription has been removed. // Uses CompareAndSwap to prevent double-close of the completed channel. removed atomic.Bool @@ -947,6 +957,10 @@ func (s *subscriptionState) done() { func (s *subscriptionState) complete() { s.writeMu.Lock() defer s.writeMu.Unlock() + if s.removed.Load() || s.terminal { + return + } + s.terminal = true s.writer.Complete() } @@ -955,6 +969,10 @@ func (s *subscriptionState) complete() { func (s *subscriptionState) error(data []byte) { s.writeMu.Lock() defer s.writeMu.Unlock() + if s.removed.Load() || s.terminal { + return + } + s.terminal = true s.writer.Error(data) } @@ -962,7 +980,7 @@ func (s *subscriptionState) error(data []byte) { func (s *subscriptionState) writeError(w AsyncErrorWriter, ctx *Context, err error, response *GraphQLResponse) { s.writeMu.Lock() defer s.writeMu.Unlock() - if s.removed.Load() { + if s.removed.Load() || s.terminal { return } w.WriteError(ctx, err, response, s.writer) @@ -970,13 +988,16 @@ func (s *subscriptionState) writeError(w AsyncErrorWriter, ctx *Context, err err // sendHeartbeat sends a keep-alive frame to the downstream writer under writeMu. // @TODO: this is bad, see ENG-9356 -func (s *subscriptionState) sendHeartbeat() error { +func (s *subscriptionState) sendHeartbeat() (sent bool, err error) { s.writeMu.Lock() defer s.writeMu.Unlock() - if s.removed.Load() { - return nil + if s.removed.Load() || s.terminal { + return false, nil } - return s.writer.Heartbeat() + if err := s.writer.Heartbeat(); err != nil { + return false, err + } + return true, nil } func (r *Resolver) executeSubscriptionUpdate(resolveCtx *Context, sub *subscriptionState, sharedInput []byte) { @@ -994,12 +1015,21 @@ func (r *Resolver) executeSubscriptionUpdate(resolveCtx *Context, sub *subscript copy(input, sharedInput) resolveArena := r.resolveArenaPool.Acquire(resolveCtx.Request.ID) + arenaReleased := false + releaseArena := func() { + if arenaReleased { + return + } + arenaReleased = true + r.resolveArenaPool.Release(resolveArena) + } + defer releaseArena() resolvable := NewResolvable(resolveArena.Arena, r.options.ResolvableOptions) authorization := NewFieldAuthorization(resolveCtx) resolvable.SetFieldAuthorization(authorization) if err := resolvable.InitSubscription(resolveCtx, input, sub.resolve.Trigger.PostProcessing); err != nil { - r.resolveArenaPool.Release(resolveArena) + releaseArena() sub.writeError(r.errorFormatter, resolveCtx, err, sub.resolve.Response) if r.options.Debug { fmt.Printf("resolver:trigger:subscription:init:failed:%d\n", sub.id.SubscriptionID) @@ -1010,7 +1040,7 @@ func (r *Resolver) executeSubscriptionUpdate(resolveCtx *Context, sub *subscript return } if err := authorization.authorizePreFetch(sub.resolve.Response); err != nil { - r.resolveArenaPool.Release(resolveArena) + releaseArena() sub.writeError(r.errorFormatter, resolveCtx, err, sub.resolve.Response) if r.options.Debug { fmt.Printf("resolver:trigger:subscription:authorization:failed:%d\n", sub.id.SubscriptionID) @@ -1027,7 +1057,7 @@ func (r *Resolver) executeSubscriptionUpdate(resolveCtx *Context, sub *subscript loader := NewLoader(r.options, r.allowedErrorExtensionFields, r.allowedErrorFields, r.subgraphRequestSingleFlight, resolveArena.Arena, db, authorization) if err := loader.LoadGraphQLResponseData(resolveCtx, sub.resolve.Response); err != nil { - r.resolveArenaPool.Release(resolveArena) + releaseArena() sub.writeError(r.errorFormatter, resolveCtx, err, sub.resolve.Response) if r.options.Debug { fmt.Printf("resolver:trigger:subscription:load:failed:%d\n", sub.id.SubscriptionID) @@ -1038,31 +1068,48 @@ func (r *Resolver) executeSubscriptionUpdate(resolveCtx *Context, sub *subscript return } - sub.writeMu.Lock() - if sub.removed.Load() { - sub.writeMu.Unlock() - r.resolveArenaPool.Release(resolveArena) - return - } + var resolveErr, flushErr error + skipped := false + func() { + sub.writeMu.Lock() + defer sub.writeMu.Unlock() + if sub.removed.Load() || sub.terminal { + skipped = true + return + } - // Inject loader output into Resolvable before rendering. InitSubscription may - // have already set resolvable.errors from the event payload, so append the - // loader's fetch errors rather than overwrite them. - resolvable.data = loader.dataBuffer.Get() - resolvable.subgraphExtensions = loader.subgraphExtensions - if loader.errors != nil { - if resolvable.errors == nil { - resolvable.errors = loader.errors - } else { - resolvable.errors.AppendArrayItems(resolveArena.Arena, loader.errors) + // Inject loader output into Resolvable before rendering. InitSubscription may + // have already set resolvable.errors from the event payload, so append the + // loader's fetch errors rather than overwrite them. + resolvable.data = loader.dataBuffer.Get() + resolvable.subgraphExtensions = loader.subgraphExtensions + if loader.errors != nil { + if resolvable.errors == nil { + resolvable.errors = loader.errors + } else { + resolvable.errors.AppendArrayItems(resolveArena.Arena, loader.errors) + } } - } - resolvable.skipValueCompletion = loader.skipValueCompletion + resolvable.skipValueCompletion = loader.skipValueCompletion - if err := resolvable.Resolve(resolveCtx.ctx, sub.resolve.Response.Data, sub.resolve.Response.Fetches, sub.writer); err != nil { - r.resolveArenaPool.Release(resolveArena) - r.errorFormatter.WriteError(resolveCtx, err, sub.resolve.Response, sub.writer) - sub.writeMu.Unlock() + resolveErr = resolvable.Resolve(resolveCtx.ctx, sub.resolve.Response.Data, sub.resolve.Response.Fetches, sub.writer) + releaseArena() + if resolveErr != nil { + r.errorFormatter.WriteError(resolveCtx, resolveErr, sub.resolve.Response, sub.writer) + return + } + + flushErr = sub.writer.Flush() + if flushErr != nil { + return + } + sub.lastWriteTime.Store(time.Now().UnixNano()) + }() + + if skipped { + return + } + if resolveErr != nil { if r.options.Debug { fmt.Printf("resolver:trigger:subscription:resolve:failed:%d\n", sub.id.SubscriptionID) } @@ -1071,17 +1118,11 @@ func (r *Resolver) executeSubscriptionUpdate(resolveCtx *Context, sub *subscript } return } - - r.resolveArenaPool.Release(resolveArena) - - if err := sub.writer.Flush(); err != nil { - sub.writeMu.Unlock() + if flushErr != nil { // If flush fails (e.g. client disconnected), remove the subscription. _ = r.UnsubscribeSubscription(sub.id) return } - sub.lastWriteTime.Store(time.Now().UnixNano()) - sub.writeMu.Unlock() if r.options.Debug { fmt.Printf("resolver:trigger:subscription:flushed:%d\n", sub.id.SubscriptionID) @@ -1104,12 +1145,13 @@ func (r *Resolver) executeSubscriptionHeartbeat(sub *subscriptionState) { return } - if err := sub.sendHeartbeat(); err != nil { + sent, err := sub.sendHeartbeat() + if err != nil { _ = r.UnsubscribeSubscription(sub.id) return } - if r.reporter != nil { + if sent && r.reporter != nil { r.reporter.SubscriptionUpdateSent() } } @@ -1137,11 +1179,35 @@ func (r *Resolver) executeStartupHooks(add *addSubscription, updater *subscripti return err } -// registerSubscriptionLocked updates the by-ID and by-connection indexes. +// registerSubscriptionLocked updates the trigger and resolver subscription indexes. +// r.mu must be held by the caller. This is used for a newly created trigger, +// which cannot be terminating before its first subscription is registered. func (r *Resolver) registerSubscriptionLocked(trig *trigger, s *subscriptionState) { trig.mu.Lock() trig.subscriptions[s.id] = s trig.mu.Unlock() + r.registerSubscriptionIndexesLocked(s) +} + +// tryRegisterSubscriptionLocked registers on an existing trigger if it is still +// accepting subscribers. Admission is linearized under trig.mu so a terminal +// state published first rejects the subscription, while a subscription registered +// first is included in the terminal snapshot. r.mu must be held by the caller. +func (r *Resolver) tryRegisterSubscriptionLocked(trig *trigger, s *subscriptionState) bool { + trig.mu.Lock() + if trig.updater.terminating.Load() { + trig.mu.Unlock() + return false + } + trig.subscriptions[s.id] = s + trig.mu.Unlock() + r.registerSubscriptionIndexesLocked(s) + return true +} + +// registerSubscriptionIndexesLocked updates the resolver's by-ID and +// by-connection indexes. r.mu must be held by the caller. +func (r *Resolver) registerSubscriptionIndexesLocked(s *subscriptionState) { id := s.id r.subscriptionsByID[id] = s byConn, ok := r.subscriptionsByConnection[id.ConnectionID] @@ -1189,14 +1255,15 @@ func (r *Resolver) addSubscription(triggerID uint64, add *addSubscription) error trig, ok := r.triggers[triggerID] if ok { + if !r.tryRegisterSubscriptionLocked(trig, s) { + return ErrSubscriptionTriggerTerminating + } if r.reporter != nil { r.reporter.SubscriptionCountInc(1) } if r.options.Debug { fmt.Printf("resolver:trigger:subscription:added:%d:%d\n", triggerID, add.id.SubscriptionID) } - // Register first so startup hooks can deliver initial data via UpdateSubscription. - r.registerSubscriptionLocked(trig, s) // Execute the startup hooks in a goroutine to avoid holding the lock. go func() { if err := r.executeStartupHooks(add, trig.updater); err != nil { @@ -1608,8 +1675,7 @@ func (r *Resolver) shutdownResolver() { } } -func (r *Resolver) heartbeatLoop() { - ticker := time.NewTicker(r.heartbeatInterval) +func (r *Resolver) heartbeatLoop(ticker *time.Ticker) { defer ticker.Stop() for { select { @@ -1908,25 +1974,70 @@ type subscriptionUpdater struct { // mu serves two roles: // // 1. Event serialization gate -- held across the entire Update() call including - // wg.Wait(), ensuring event A fully completes before event B begins. + // updateWG.Wait(), ensuring lifecycle operations cannot overtake an admitted + // per-subscription update. // - // 2. Lifecycle guard -- the done flag prevents callbacks after Done() has torn down - // the trigger. Every method checks done || ctx.Err() under the lock before proceeding. - mu sync.Mutex - done bool - debug bool - triggerID uint64 - resolver *Resolver - ctx context.Context - subsFn func() map[context.Context]SubscriptionIdentifier + // 2. Lifecycle guard -- terminal prevents data/control frames after the first + // Complete or Error, while done prevents callbacks after Done tears down the trigger. + // Lock ordering: mu must be acquired before updateMu when both are needed. + mu sync.Mutex + done bool + terminal bool + // terminating is published while mu is held and observed under trigger.mu to + // linearize late subscription admission against Complete, Error, and Done. + terminating atomic.Bool + debug bool + triggerID uint64 + resolver *Resolver + ctx context.Context + subsFn func() map[context.Context]SubscriptionIdentifier + + // updateMu protects the per-subscription lane tails. An admitted update publishes + // a new tail, waits for its predecessor, resolves synchronously, then closes its + // tail so the next update in the same lane can proceed. + updateMu sync.Mutex + updateTails map[SubscriptionIdentifier]chan struct{} + // Every updateWG.Add occurs while mu is held, and every lifecycle updateWG.Wait + // also holds mu, so Add cannot race a zero-count Wait. Update completion never + // needs mu, which lets lifecycle operations wait without deadlocking. + updateWG sync.WaitGroup +} + +// admitSubscriptionUpdate's caller must hold mu so admission cannot race a +// lifecycle operation waiting on updateWG. +func (s *subscriptionUpdater) admitSubscriptionUpdate(id SubscriptionIdentifier) (previous <-chan struct{}, current chan struct{}) { + s.updateMu.Lock() + defer s.updateMu.Unlock() + + if s.updateTails == nil { + s.updateTails = make(map[SubscriptionIdentifier]chan struct{}) + } + previous = s.updateTails[id] + current = make(chan struct{}) + s.updateWG.Add(1) + s.updateTails[id] = current + return previous, current +} + +func (s *subscriptionUpdater) finishSubscriptionUpdate(id SubscriptionIdentifier, current chan struct{}) { + // Completion deliberately never acquires mu: lifecycle code may hold mu while + // waiting for all admitted updates to finish. + s.updateMu.Lock() + close(current) + if s.updateTails[id] == current { + delete(s.updateTails, id) + } + s.updateMu.Unlock() + s.updateWG.Done() } func (s *subscriptionUpdater) Update(data []byte) { s.mu.Lock() defer s.mu.Unlock() - if s.done || s.ctx.Err() != nil { + if s.done || s.terminal || s.ctx.Err() != nil { return } + s.updateWG.Wait() if s.debug { fmt.Printf("resolver:subscription_updater:update:%d\n", s.triggerID) } @@ -1936,16 +2047,31 @@ func (s *subscriptionUpdater) Update(data []byte) { func (s *subscriptionUpdater) Heartbeat() { s.mu.Lock() defer s.mu.Unlock() - if s.done || s.ctx.Err() != nil { + if s.done || s.terminal || s.ctx.Err() != nil { return } + s.updateWG.Wait() s.resolver.heartbeatTriggerSubscriptions(s.triggerID) } func (s *subscriptionUpdater) UpdateSubscription(id SubscriptionIdentifier, data []byte) { s.mu.Lock() - defer s.mu.Unlock() - if s.done || s.ctx.Err() != nil { + if s.done || s.terminal || s.ctx.Err() != nil { + s.mu.Unlock() + return + } + previous, current := s.admitSubscriptionUpdate(id) + s.mu.Unlock() + defer s.finishSubscriptionUpdate(id, current) + + if previous != nil { + select { + case <-previous: + case <-s.ctx.Done(): + return + } + } + if s.ctx.Err() != nil { return } if s.debug { @@ -1961,12 +2087,15 @@ func (s *subscriptionUpdater) Subscriptions() map[context.Context]SubscriptionId func (s *subscriptionUpdater) Complete() { s.mu.Lock() defer s.mu.Unlock() - if s.done || s.ctx.Err() != nil { + s.terminating.Store(true) + if s.done || s.terminal || s.ctx.Err() != nil { if s.debug { fmt.Printf("resolver:subscription_updater:complete:skip:%d\n", s.triggerID) } return } + s.terminal = true + s.updateWG.Wait() if s.debug { fmt.Printf("resolver:subscription_updater:complete:%d\n", s.triggerID) } @@ -1976,12 +2105,15 @@ func (s *subscriptionUpdater) Complete() { func (s *subscriptionUpdater) Error(data []byte) { s.mu.Lock() defer s.mu.Unlock() - if s.done || s.ctx.Err() != nil { + s.terminating.Store(true) + if s.done || s.terminal || s.ctx.Err() != nil { if s.debug { fmt.Printf("resolver:subscription_updater:error:skip:%d\n", s.triggerID) } return } + s.terminal = true + s.updateWG.Wait() if s.debug { fmt.Printf("resolver:subscription_updater:error:%d\n", s.triggerID) } @@ -1991,10 +2123,12 @@ func (s *subscriptionUpdater) Error(data []byte) { func (s *subscriptionUpdater) Done() { s.mu.Lock() defer s.mu.Unlock() + s.terminating.Store(true) if s.done { return } s.done = true + s.updateWG.Wait() if s.debug { fmt.Printf("resolver:subscription_updater:done:%d\n", s.triggerID) } @@ -2010,6 +2144,7 @@ func (s *subscriptionUpdater) CloseSubscription(id SubscriptionIdentifier) { } return } + s.updateWG.Wait() if s.debug { fmt.Printf("resolver:subscription_updater:close:%d\n", s.triggerID) } @@ -2028,22 +2163,33 @@ type addSubscription struct { headers http.Header } +// SubscriptionUpdater receives events and lifecycle callbacks for one trigger. +// Implementations must be safe for concurrent calls. Targeted updates for different +// subscription identifiers may resolve concurrently, while targeted updates for the +// same identifier execute in updater admission order. Lifecycle methods wait for +// already admitted targeted updates before taking effect. type SubscriptionUpdater interface { // Update sends an update to the client. It is not guaranteed that the update is sent immediately. Update(data []byte) - // UpdateSubscription sends an update to a single subscription. It is not guaranteed that the update is sent immediately. + // UpdateSubscription synchronously resolves an update for one subscription. The + // call returns after the update resolves or is skipped. Calls for the same identifier + // execute in updater admission order; calls for different identifiers may overlap. UpdateSubscription(id SubscriptionIdentifier, data []byte) // Complete delivers a "subscription done" signal to all subscriptions on the trigger. - // Does not perform cleanup — call Done() after Complete(). + // The first Complete or Error call wins. Complete waits for admitted targeted + // updates and does not perform cleanup — call Done after Complete. Complete() // Error delivers a terminal error to all subscriptions on the trigger, bypassing the resolve pipeline. - // Does not perform cleanup — call Done() after Error(). + // The first Complete or Error call wins. Error waits for admitted targeted + // updates and does not perform cleanup — call Done after Error. Error(data []byte) // Done performs internal cleanup: detaches the trigger, closes completed channels. - // Must always be the final call. Does not send any downstream messages. + // It waits for admitted targeted updates, must always be the final call, and is + // still required after Complete or Error. Done does not send downstream messages. Done() - // CloseSubscription closes a single subscription. No more updates should be sent to that subscription after calling CloseSubscription. + // CloseSubscription waits for admitted targeted updates, then closes a single + // subscription. No more updates should be sent to it after the call returns. CloseSubscription(id SubscriptionIdentifier) - // Subscriptions return all the subscriptions associated to this Updater + // Subscriptions returns a snapshot of subscriptions associated with this updater. Subscriptions() map[context.Context]SubscriptionIdentifier } diff --git a/v2/pkg/engine/resolve/subscription_updater_test.go b/v2/pkg/engine/resolve/subscription_updater_test.go new file mode 100644 index 0000000000..204f243375 --- /dev/null +++ b/v2/pkg/engine/resolve/subscription_updater_test.go @@ -0,0 +1,2112 @@ +package resolve + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/wundergraph/astjson" + + "github.com/wundergraph/graphql-go-tools/v2/pkg/ast" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/httpclient" +) + +type concurrentSubscriptionDataSource struct { + data []byte + + mu sync.Mutex + started int + startedEvents chan int + expected int + allStarted chan struct{} + allStartedOnce sync.Once + release chan struct{} + releaseOnce sync.Once +} + +type panickingSubscriptionDataSource struct { + panicValue any +} + +type subscriptionUpdatePanic struct { + label string +} + +type sequencedSubscriptionFilterRenderer struct { + mu sync.Mutex + expectedContext context.Context + calls int + sawUnexpectedCtx bool + sawNonNilEventData bool + secondErr error +} + +func (r *sequencedSubscriptionFilterRenderer) GetKind() string { + return VariableRendererKindJson +} + +func (r *sequencedSubscriptionFilterRenderer) RenderVariable(ctx context.Context, data *astjson.Value, out io.Writer) error { + r.mu.Lock() + defer r.mu.Unlock() + + r.calls++ + if ctx != r.expectedContext { + r.sawUnexpectedCtx = true + } + if data != nil { + r.sawNonNilEventData = true + } + if r.calls == 2 { + return r.secondErr + } + _, err := out.Write([]byte(`"match"`)) + return err +} + +func (r *sequencedSubscriptionFilterRenderer) results() (calls int, sawUnexpectedContext, sawNonNilEventData bool) { + r.mu.Lock() + defer r.mu.Unlock() + return r.calls, r.sawUnexpectedCtx, r.sawNonNilEventData +} + +type chronologicalSubscriptionWriter struct { + mu sync.Mutex + buf bytes.Buffer + entries []string +} + +func (w *chronologicalSubscriptionWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.Write(p) +} + +func (w *chronologicalSubscriptionWriter) Flush() error { + w.mu.Lock() + defer w.mu.Unlock() + w.entries = append(w.entries, w.buf.String()) + w.buf.Reset() + return nil +} + +func (w *chronologicalSubscriptionWriter) Complete() { + w.mu.Lock() + defer w.mu.Unlock() + w.entries = append(w.entries, "") +} + +func (w *chronologicalSubscriptionWriter) Heartbeat() error { + w.mu.Lock() + defer w.mu.Unlock() + w.entries = append(w.entries, "") + return nil +} + +func (w *chronologicalSubscriptionWriter) Error(data []byte) { + w.mu.Lock() + defer w.mu.Unlock() + w.entries = append(w.entries, string(data)) +} + +func (w *chronologicalSubscriptionWriter) Entries() []string { + w.mu.Lock() + defer w.mu.Unlock() + return append([]string(nil), w.entries...) +} + +type chronologicalSubscriptionErrorWriter struct{} + +type panickingSubscriptionWriter struct { + chronologicalSubscriptionWriter + writeErr error + flushErr error + panicWrite any + panicFlush any + flushStarted chan struct{} + flushRelease <-chan struct{} +} + +func (w *panickingSubscriptionWriter) Write(p []byte) (int, error) { + if w.panicWrite != nil { + panic(w.panicWrite) + } + if w.writeErr != nil { + return 0, w.writeErr + } + return w.chronologicalSubscriptionWriter.Write(p) +} + +func (w *panickingSubscriptionWriter) Flush() error { + if w.panicFlush != nil { + panic(w.panicFlush) + } + if w.flushStarted != nil { + close(w.flushStarted) + } + if w.flushRelease != nil { + <-w.flushRelease + } + if w.flushErr != nil { + return w.flushErr + } + return w.chronologicalSubscriptionWriter.Flush() +} + +type panickingSubscriptionErrorWriter struct { + panicValue any +} + +func (w *panickingSubscriptionErrorWriter) WriteError(*Context, error, *GraphQLResponse, io.Writer) { + panic(w.panicValue) +} + +type subscriptionSingleFlightWaitContext struct { + context.Context + waitObserved chan struct{} + once sync.Once +} + +func (c *subscriptionSingleFlightWaitContext) Done() <-chan struct{} { + c.once.Do(func() { + close(c.waitObserved) + }) + return c.Context.Done() +} + +type subscriptionSingleFlightLoaderHook struct { + waitObserved chan struct{} +} + +func (h *subscriptionSingleFlightLoaderHook) OnLoad(ctx context.Context, _ DataSourceInfo) context.Context { + return &subscriptionSingleFlightWaitContext{Context: ctx, waitObserved: h.waitObserved} +} + +func (h *subscriptionSingleFlightLoaderHook) OnFinished(context.Context, DataSourceInfo, *ResponseInfo) { +} + +type subscriptionUpdateCountingReporter struct { + mu sync.Mutex + updates int + subscriptions int + triggers int +} + +type subscriptionUpdateReporterSnapshot struct { + updates int + subscriptions int + triggers int +} + +func (r *subscriptionUpdateCountingReporter) SubscriptionUpdateSent() { + r.mu.Lock() + defer r.mu.Unlock() + r.updates++ +} + +func (r *subscriptionUpdateCountingReporter) SubscriptionCountInc(count int) { + r.mu.Lock() + defer r.mu.Unlock() + r.subscriptions += count +} + +func (r *subscriptionUpdateCountingReporter) SubscriptionCountDec(count int) { + r.mu.Lock() + defer r.mu.Unlock() + r.subscriptions -= count +} + +func (r *subscriptionUpdateCountingReporter) TriggerCountInc(count int) { + r.mu.Lock() + defer r.mu.Unlock() + r.triggers += count +} + +func (r *subscriptionUpdateCountingReporter) TriggerCountDec(count int) { + r.mu.Lock() + defer r.mu.Unlock() + r.triggers -= count +} + +func (r *subscriptionUpdateCountingReporter) snapshot() subscriptionUpdateReporterSnapshot { + r.mu.Lock() + defer r.mu.Unlock() + return subscriptionUpdateReporterSnapshot{ + updates: r.updates, + subscriptions: r.subscriptions, + triggers: r.triggers, + } +} + +func (w *chronologicalSubscriptionErrorWriter) WriteError(_ *Context, err error, _ *GraphQLResponse, out io.Writer) { + _, _ = fmt.Fprintf(out, `{"errors":[{"message":%q}]}`, err.Error()) + if flusher, ok := out.(interface{ Flush() error }); ok { + _ = flusher.Flush() + } +} + +func (d *panickingSubscriptionDataSource) Load(context.Context, http.Header, []byte) ([]byte, error) { + panic(d.panicValue) +} + +func (d *panickingSubscriptionDataSource) LoadWithFiles(context.Context, http.Header, []byte, []*httpclient.FileUpload) ([]byte, error) { + panic(d.panicValue) +} + +func newConcurrentSubscriptionDataSource(expected int) *concurrentSubscriptionDataSource { + return &concurrentSubscriptionDataSource{ + data: []byte(`{"data":{"resolved":"value"}}`), + expected: expected, + startedEvents: make(chan int, expected+1), + allStarted: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (d *concurrentSubscriptionDataSource) load() ([]byte, error) { + d.mu.Lock() + d.started++ + started := d.started + if d.started == d.expected { + d.allStartedOnce.Do(func() { + close(d.allStarted) + }) + } + d.mu.Unlock() + select { + case d.startedEvents <- started: + default: + } + + <-d.release + return d.data, nil +} + +func (d *concurrentSubscriptionDataSource) Load(context.Context, http.Header, []byte) ([]byte, error) { + return d.load() +} + +func (d *concurrentSubscriptionDataSource) LoadWithFiles(context.Context, http.Header, []byte, []*httpclient.FileUpload) ([]byte, error) { + return d.load() +} + +func (d *concurrentSubscriptionDataSource) AllStarted() <-chan struct{} { + return d.allStarted +} + +func (d *concurrentSubscriptionDataSource) StartedEvents() <-chan int { + return d.startedEvents +} + +func (d *concurrentSubscriptionDataSource) Started() int { + d.mu.Lock() + defer d.mu.Unlock() + return d.started +} + +func (d *concurrentSubscriptionDataSource) Release() { + d.releaseOnce.Do(func() { + close(d.release) + }) +} + +func newSubscriptionUpdaterHarness(t *testing.T, count int, dataSource DataSource) (*subscriptionUpdater, []SubscriptionIdentifier, []*SubscriptionRecorder, func()) { + t.Helper() + + resolverCtx, cancelResolver := context.WithCancel(t.Context()) + resolver := New(resolverCtx, ResolverOptions{ + MaxConcurrency: count, + AsyncErrorWriter: &FakeErrorWriter{}, + SubscriptionHeartbeatInterval: time.Hour, + }) + + const triggerID uint64 = 1 + triggerCtx, cancelTrigger := context.WithCancel(resolverCtx) + updater := &subscriptionUpdater{ + triggerID: triggerID, + resolver: resolver, + ctx: triggerCtx, + } + trig := &trigger{ + id: triggerID, + cancel: cancelTrigger, + subscriptions: make(map[SubscriptionIdentifier]*subscriptionState, count), + updater: updater, + } + updater.subsFn = trig.subscriptionIds + + ids := make([]SubscriptionIdentifier, 0, count) + recorders := make([]*SubscriptionRecorder, 0, count) + completed := make([]<-chan struct{}, 0, count) + + resolver.mu.Lock() + resolver.triggers[triggerID] = trig + for i := 0; i < count; i++ { + id := SubscriptionIdentifier{ + ConnectionID: ConnectionID(i + 1), + SubscriptionID: int64(i + 1), + } + recorder := &SubscriptionRecorder{buf: &bytes.Buffer{}} + subscriptionCtx := NewContext(resolverCtx) + subscriptionCtx.ExecutionOptions.DisableSubgraphRequestDeduplication = true + completion := make(chan struct{}) + + state := &subscriptionState{ + triggerID: triggerID, + resolve: &GraphQLSubscription{ + Trigger: GraphQLSubscriptionTrigger{ + PostProcessing: PostProcessingConfiguration{ + SelectResponseDataPath: []string{"data"}, + }, + }, + Response: &GraphQLResponse{ + Info: &GraphQLResponseInfo{OperationType: ast.OperationTypeQuery}, + Fetches: Single(&SingleFetch{ + FetchConfiguration: FetchConfiguration{ + DataSource: dataSource, + PostProcessing: PostProcessingConfiguration{ + SelectResponseDataPath: []string{"data"}, + }, + }, + InputTemplate: InputTemplate{ + Segments: []TemplateSegment{{ + SegmentType: StaticSegmentType, + Data: []byte(`{}`), + }}, + }, + Info: &FetchInfo{ + DataSourceID: "source", + DataSourceName: "source", + OperationType: ast.OperationTypeQuery, + QueryPlan: &QueryPlan{ + Query: "query { resolved }", + }, + }, + }), + Data: &Object{ + Fields: []*Field{ + { + Name: []byte("value"), + Value: &String{ + Path: []string{"value"}, + }, + }, + { + Name: []byte("resolved"), + Value: &String{ + Path: []string{"resolved"}, + }, + }, + }, + }, + }, + }, + ctx: subscriptionCtx, + writer: recorder, + id: id, + completed: completion, + } + + resolver.registerSubscriptionLocked(trig, state) + ids = append(ids, id) + recorders = append(recorders, recorder) + completed = append(completed, completion) + } + resolver.mu.Unlock() + + cleanup := func() { + if releaser, ok := dataSource.(interface{ Release() }); ok { + releaser.Release() + } + cancelResolver() + resolver.shutdownResolver() + + deadline := time.NewTimer(time.Second) + defer deadline.Stop() + for _, done := range completed { + select { + case <-done: + case <-deadline.C: + t.Errorf("timed out waiting for subscription updater harness cleanup") + return + } + } + } + + return updater, ids, recorders, cleanup +} + +func subscriptionUpdateTail(updater *subscriptionUpdater, id SubscriptionIdentifier) chan struct{} { + updater.updateMu.Lock() + defer updater.updateMu.Unlock() + return updater.updateTails[id] +} + +func waitForSubscriptionUpdateTailChange(t *testing.T, updater *subscriptionUpdater, id SubscriptionIdentifier, previous chan struct{}) chan struct{} { + t.Helper() + + deadline := time.Now().Add(time.Second) + for { + current := subscriptionUpdateTail(updater, id) + if current != nil && current != previous { + return current + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for subscription update tail to change") + } + time.Sleep(time.Millisecond) + } +} + +func waitForSubscriptionUpdateCall(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(time.Second): + t.Error("timed out waiting for subscription update call") + } +} + +func requireNoSubscriptionUpdateTails(t *testing.T, updater *subscriptionUpdater) { + t.Helper() + updater.updateMu.Lock() + defer updater.updateMu.Unlock() + require.Empty(t, updater.updateTails) +} + +func waitForSubscriptionUpdates(t *testing.T, updater *subscriptionUpdater) { + t.Helper() + done := make(chan struct{}) + go func() { + updater.mu.Lock() + defer updater.mu.Unlock() + updater.updateWG.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Error("timed out waiting for admitted subscription updates") + } +} + +func startSubscriptionUpdaterCall(call func()) (started, done <-chan struct{}) { + startedCh := make(chan struct{}) + doneCh := make(chan struct{}) + go func() { + close(startedCh) + defer close(doneCh) + call() + }() + return startedCh, doneCh +} + +func waitForSubscriptionUpdaterCallStart(t *testing.T, started <-chan struct{}, deadline time.Time) { + t.Helper() + select { + case <-started: + case <-time.After(time.Until(deadline)): + t.Fatal("timed out waiting for subscription updater call to start") + } +} + +func waitForSubscriptionUpdaterCallToEnter(t *testing.T, updater *subscriptionUpdater, done <-chan struct{}, deadline time.Time) { + t.Helper() + for { + if !updater.mu.TryLock() { + return + } + updater.mu.Unlock() + select { + case <-done: + t.Error("subscription updater call returned before reaching the lifecycle barrier") + return + default: + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for subscription updater call to reach the lifecycle barrier") + } + time.Sleep(time.Millisecond) + } +} + +func waitForSubscriptionUpdaterTerminating(t *testing.T, updater *subscriptionUpdater, done <-chan struct{}, deadline time.Time) { + t.Helper() + for !updater.terminating.Load() { + select { + case <-done: + t.Error("subscription updater call returned before publishing termination") + return + default: + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for subscription updater to publish termination") + } + time.Sleep(time.Millisecond) + } +} + +func checkSubscriptionUpdaterCallBlocked(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + t.Error("subscription updater call returned before admitted update was released") + case <-time.After(50 * time.Millisecond): + } +} + +func waitForSubscriptionUpdaterCall(t *testing.T, done <-chan struct{}, deadline time.Time) { + t.Helper() + select { + case <-done: + case <-time.After(time.Until(deadline)): + t.Error("timed out waiting for subscription updater call") + } +} + +func requireSubscriptionUpdaterCallShortCircuitsWithoutResolver(t *testing.T, updater *subscriptionUpdater, name string, call func()) { + t.Helper() + resolver := updater.resolver + updater.resolver = nil + defer func() { + updater.resolver = resolver + }() + require.NotPanics(t, call, "%s reached the resolver instead of short-circuiting", name) +} + +func subscriptionUpdaterHasID(subscriptions map[context.Context]SubscriptionIdentifier, id SubscriptionIdentifier) bool { + for _, subscriptionID := range subscriptions { + if subscriptionID == id { + return true + } + } + return false +} + +func cancelSubscriptionUpdater(t *testing.T, updater *subscriptionUpdater) { + t.Helper() + trig, ok := updater.resolver.getTrigger(updater.triggerID) + require.True(t, ok) + trig.cancel() +} + +func configureSubscriptionUpdaterState(t *testing.T, updater *subscriptionUpdater, id SubscriptionIdentifier, configure func(*subscriptionState)) { + t.Helper() + trig, ok := updater.resolver.getTrigger(updater.triggerID) + require.True(t, ok) + trig.mu.Lock() + defer trig.mu.Unlock() + state, ok := trig.subscriptions[id] + require.True(t, ok) + configure(state) +} + +func TestSubscriptionState_Terminal_CompleteSuppressesLaterFrames(t *testing.T) { + writer := &chronologicalSubscriptionWriter{} + ctx := NewContext(t.Context()) + state := &subscriptionState{ + ctx: ctx, + writer: writer, + completed: make(chan struct{}), + } + + state.complete() + sent, err := state.sendHeartbeat() + require.NoError(t, err) + require.False(t, sent) + state.writeError(&chronologicalSubscriptionErrorWriter{}, ctx, errors.New("late write error"), &GraphQLResponse{}) + state.error([]byte(`{"errors":[{"message":"late terminal error"}]}`)) + + require.Equal(t, []string{""}, writer.Entries()) +} + +func TestSubscriptionState_Terminal_ErrorSuppressesLaterFrames(t *testing.T) { + const terminalError = `{"errors":[{"message":"terminal"}]}` + writer := &chronologicalSubscriptionWriter{} + ctx := NewContext(t.Context()) + state := &subscriptionState{ + ctx: ctx, + writer: writer, + completed: make(chan struct{}), + } + + state.error([]byte(terminalError)) + sent, err := state.sendHeartbeat() + require.NoError(t, err) + require.False(t, sent) + state.writeError(&chronologicalSubscriptionErrorWriter{}, ctx, errors.New("late write error"), &GraphQLResponse{}) + state.complete() + + require.Equal(t, []string{terminalError}, writer.Entries()) +} + +func TestSubscriptionState_RemovedSuppressesTerminalFrames(t *testing.T) { + const terminalError = `{"errors":[{"message":"terminal"}]}` + tests := []struct { + name string + deliver func(*subscriptionState) + }{ + { + name: "Complete", + deliver: func(state *subscriptionState) { + state.complete() + }, + }, + { + name: "Error", + deliver: func(state *subscriptionState) { + state.error([]byte(terminalError)) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + writer := &chronologicalSubscriptionWriter{} + state := &subscriptionState{ + ctx: NewContext(t.Context()), + writer: writer, + completed: make(chan struct{}), + } + state.removed.Store(true) + + test.deliver(state) + + require.Empty(t, writer.Entries()) + require.False(t, state.terminal) + }) + } +} + +func TestSubscriptionState_Terminal_CompetingTerminalFramesEmitExactlyOnce(t *testing.T) { + const terminalError = `{"errors":[{"message":"terminal"}]}` + writer := &chronologicalSubscriptionWriter{} + state := &subscriptionState{ + ctx: NewContext(t.Context()), + writer: writer, + completed: make(chan struct{}), + } + + start := make(chan struct{}) + var calls sync.WaitGroup + for i := 0; i < 32; i++ { + calls.Add(1) + go func(i int) { + defer calls.Done() + <-start + if i%2 == 0 { + state.complete() + return + } + state.error([]byte(terminalError)) + }(i) + } + close(start) + + done := make(chan struct{}) + go func() { + calls.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timed out waiting for competing terminal frames") + } + + entries := writer.Entries() + require.Len(t, entries, 1) + require.True(t, entries[0] == "" || entries[0] == terminalError, + "unexpected terminal frame %q", entries[0]) +} + +func TestSubscriptionState_Terminal_LateDataDropsResolvedData(t *testing.T) { + const terminalError = `{"errors":[{"message":"terminal"}]}` + tests := []struct { + name string + terminal string + deliver func(*Resolver, uint64) + }{ + { + name: "Complete", + terminal: "", + deliver: func(resolver *Resolver, triggerID uint64) { + resolver.handleTriggerComplete(triggerID) + }, + }, + { + name: "Error", + terminal: terminalError, + deliver: func(resolver *Resolver, triggerID uint64) { + resolver.handleTriggerError(triggerID, []byte(terminalError)) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + writer := &chronologicalSubscriptionWriter{} + var state *subscriptionState + configureSubscriptionUpdaterState(t, updater, ids[0], func(subscription *subscriptionState) { + subscription.writer = writer + state = subscription + }) + + updateDone := make(chan struct{}) + go func() { + defer close(updateDone) + updater.resolver.executeSubscriptionUpdate(state.ctx, state, []byte(`{"data":{"value":"late"}}`)) + }() + defer func() { + dataSource.Release() + select { + case <-updateDone: + case <-time.After(time.Second): + t.Error("timed out joining late subscription update") + } + }() + + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + t.Fatal("timed out waiting for late subscription update to enter the data source") + } + + test.deliver(updater.resolver, updater.triggerID) + require.Equal(t, []string{test.terminal}, writer.Entries()) + + dataSource.Release() + select { + case <-updateDone: + case <-time.After(time.Second): + t.Fatal("timed out waiting for late subscription update to return") + } + require.Equal(t, []string{test.terminal}, writer.Entries()) + }) + } +} + +func TestSubscriptionState_WritePanic_ReleasesLifecycleLock(t *testing.T) { + tests := []struct { + name string + configure func(*Resolver, *panickingSubscriptionWriter, any) + }{ + { + name: "Write", + configure: func(_ *Resolver, writer *panickingSubscriptionWriter, panicValue any) { + writer.panicWrite = panicValue + }, + }, + { + name: "Flush", + configure: func(_ *Resolver, writer *panickingSubscriptionWriter, panicValue any) { + writer.panicFlush = panicValue + }, + }, + { + name: "ErrorFormatter", + configure: func(resolver *Resolver, writer *panickingSubscriptionWriter, panicValue any) { + writer.writeErr = errors.New("subscription writer failed") + resolver.SetAsyncErrorWriter(&panickingSubscriptionErrorWriter{panicValue: panicValue}) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + + panicValue := &subscriptionUpdatePanic{label: test.name} + writer := &panickingSubscriptionWriter{} + test.configure(updater.resolver, writer, panicValue) + var state *subscriptionState + var completed <-chan struct{} + configureSubscriptionUpdaterState(t, updater, ids[0], func(subscription *subscriptionState) { + subscription.writer = writer + state = subscription + completed = subscription.completed + }) + arenaKey := state.ctx.Request.ID + expectedArena := updater.resolver.resolveArenaPool.Acquire(arenaKey) + updater.resolver.resolveArenaPool.Release(expectedArena) + + var recovered any + func() { + defer func() { + recovered = recover() + }() + updater.resolver.executeSubscriptionUpdate(state.ctx, state, []byte(`{"data":{"value":"event"}}`)) + }() + require.Same(t, panicValue, recovered) + observedArena := updater.resolver.resolveArenaPool.Acquire(arenaKey) + require.Same(t, expectedArena, observedArena, "resolve arena was not returned after panic") + updater.resolver.resolveArenaPool.Release(observedArena) + + done := make(chan struct{}) + go func() { + updater.Done() + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("Done blocked on writeMu after recovered writer panic") + } + select { + case <-completed: + case <-time.After(time.Second): + t.Fatal("Done did not close completion after recovered writer panic") + } + cleanup() + }) + } +} + +func TestSubscriptionState_ArenaReleasedAfterInitializationError(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + writer := &chronologicalSubscriptionWriter{} + var state *subscriptionState + configureSubscriptionUpdaterState(t, updater, ids[0], func(subscription *subscriptionState) { + subscription.writer = writer + state = subscription + }) + + arenaKey := state.ctx.Request.ID + expectedArena := updater.resolver.resolveArenaPool.Acquire(arenaKey) + updater.resolver.resolveArenaPool.Release(expectedArena) + + updater.resolver.executeSubscriptionUpdate(state.ctx, state, []byte(`{`)) + + observedArena := updater.resolver.resolveArenaPool.Acquire(arenaKey) + require.Same(t, expectedArena, observedArena, "resolve arena was not returned after initialization error") + updater.resolver.resolveArenaPool.Release(observedArena) + require.Zero(t, dataSource.Started()) +} + +func TestSubscriptionState_ArenaReleasedBeforeBlockingFlush(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + flushStarted := make(chan struct{}) + flushRelease := make(chan struct{}) + writer := &panickingSubscriptionWriter{ + flushStarted: flushStarted, + flushRelease: flushRelease, + } + var state *subscriptionState + configureSubscriptionUpdaterState(t, updater, ids[0], func(subscription *subscriptionState) { + subscription.writer = writer + state = subscription + }) + + arenaKey := state.ctx.Request.ID + expectedArena := updater.resolver.resolveArenaPool.Acquire(arenaKey) + updater.resolver.resolveArenaPool.Release(expectedArena) + + updateDone := make(chan struct{}) + go func() { + defer close(updateDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"event"}}`)) + }() + select { + case <-flushStarted: + case <-time.After(time.Second): + close(flushRelease) + t.Fatal("timed out waiting for blocking Flush") + } + + observedArena := updater.resolver.resolveArenaPool.Acquire(arenaKey) + reusedBeforeFlushReturned := observedArena == expectedArena + updater.resolver.resolveArenaPool.Release(observedArena) + close(flushRelease) + waitForSubscriptionUpdateCall(t, updateDone) + + require.True(t, reusedBeforeFlushReturned, "resolve arena remained checked out while Flush was blocked") +} + +func TestResolver_TerminalHeartbeat_DoesNotWriteAfterTerminal(t *testing.T) { + const terminalError = `{"errors":[{"message":"terminal"}]}` + tests := []struct { + name string + terminal string + deliver func(*subscriptionState) + }{ + { + name: "Complete", + terminal: "", + deliver: func(state *subscriptionState) { + state.complete() + }, + }, + { + name: "Error", + terminal: terminalError, + deliver: func(state *subscriptionState) { + state.error([]byte(terminalError)) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + updater.resolver.heartbeatInterval = 0 + + writer := &chronologicalSubscriptionWriter{} + var state *subscriptionState + configureSubscriptionUpdaterState(t, updater, ids[0], func(subscription *subscriptionState) { + subscription.writer = writer + subscription.heartbeat = true + state = subscription + }) + + test.deliver(state) + updater.resolver.heartbeatTriggerSubscriptions(updater.triggerID) + + require.Equal(t, []string{test.terminal}, writer.Entries()) + }) + } +} + +func TestResolver_TerminalHeartbeat_ReportsOnlySentFrames(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 2, dataSource) + defer cleanup() + + reporter := &subscriptionUpdateCountingReporter{} + updater.resolver.reporter = reporter + writers := make([]*chronologicalSubscriptionWriter, len(ids)) + states := make([]*subscriptionState, len(ids)) + for i, id := range ids { + writers[i] = &chronologicalSubscriptionWriter{} + configureSubscriptionUpdaterState(t, updater, id, func(state *subscriptionState) { + state.writer = writers[i] + state.heartbeat = true + states[i] = state + }) + } + + states[0].complete() + updater.resolver.executeSubscriptionHeartbeat(states[0]) + require.Zero(t, reporter.snapshot().updates, "suppressed terminal heartbeat was reported as sent") + require.Equal(t, []string{""}, writers[0].Entries()) + + updater.resolver.executeSubscriptionHeartbeat(states[1]) + require.Equal(t, 1, reporter.snapshot().updates) + require.Equal(t, []string{""}, writers[1].Entries()) +} + +func TestSubscriptionUpdater_DirectUnsubscribe_DoesNotWaitForInFlightUpdate(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + defer dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + writer := &chronologicalSubscriptionWriter{} + var completed <-chan struct{} + configureSubscriptionUpdaterState(t, updater, ids[0], func(state *subscriptionState) { + state.writer = writer + completed = state.completed + }) + + updateDone := make(chan struct{}) + go func() { + defer close(updateDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"late"}}`)) + }() + defer func() { + dataSource.Release() + select { + case <-updateDone: + case <-time.After(time.Second): + t.Error("timed out joining unsubscribed subscription update") + } + }() + + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + t.Fatal("timed out waiting for subscription update to enter the data source") + } + + unsubscribeDone := make(chan error, 1) + go func() { + unsubscribeDone <- updater.resolver.UnsubscribeSubscription(ids[0]) + }() + select { + case err := <-unsubscribeDone: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("direct unsubscribe waited for the in-flight updater barrier") + } + select { + case <-completed: + case <-time.After(time.Second): + t.Fatal("direct unsubscribe did not close the subscription completion channel") + } + + dataSource.Release() + select { + case <-updateDone: + case <-time.After(time.Second): + t.Fatal("timed out waiting for late resolve after direct unsubscribe") + } + require.Empty(t, writer.Entries()) + waitForSubscriptionUpdates(t, updater) + requireNoSubscriptionUpdateTails(t, updater) +} + +func TestSubscriptionUpdater_Lifecycle_WaitsForAdmittedSubscriptionUpdate(t *testing.T) { + const ( + admittedData = `{"data":{"value":"admitted"}}` + broadcastData = `{"data":{"value":"broadcast"}}` + terminalError = `{"errors":[{"message":"terminal"}]}` + ) + + tests := []struct { + name string + invoke func(*subscriptionUpdater, SubscriptionIdentifier) + assertAbsent func(*testing.T, *subscriptionUpdater, SubscriptionIdentifier, *concurrentSubscriptionDataSource, *chronologicalSubscriptionWriter, <-chan struct{}) + assertEffect func(*testing.T, *subscriptionUpdater, SubscriptionIdentifier, *concurrentSubscriptionDataSource, *chronologicalSubscriptionWriter, <-chan struct{}, time.Time) + }{ + { + name: "Update", + invoke: func(updater *subscriptionUpdater, _ SubscriptionIdentifier) { + updater.Update([]byte(broadcastData)) + }, + assertAbsent: func(t *testing.T, _ *subscriptionUpdater, _ SubscriptionIdentifier, dataSource *concurrentSubscriptionDataSource, _ *chronologicalSubscriptionWriter, _ <-chan struct{}) { + require.Equal(t, 1, dataSource.Started(), "broadcast fetch started before admitted update completed") + }, + assertEffect: func(t *testing.T, _ *subscriptionUpdater, _ SubscriptionIdentifier, dataSource *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, _ <-chan struct{}, _ time.Time) { + require.Equal(t, 2, dataSource.Started()) + require.Equal(t, []string{ + `{"data":{"value":"admitted","resolved":"value"}}`, + `{"data":{"value":"broadcast","resolved":"value"}}`, + }, writer.Entries()) + }, + }, + { + name: "Heartbeat", + invoke: func(updater *subscriptionUpdater, _ SubscriptionIdentifier) { + updater.Heartbeat() + }, + assertAbsent: func(t *testing.T, _ *subscriptionUpdater, _ SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, _ <-chan struct{}) { + require.Empty(t, writer.Entries(), "heartbeat was written before admitted update completed") + }, + assertEffect: func(t *testing.T, _ *subscriptionUpdater, _ SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, _ <-chan struct{}, _ time.Time) { + require.Equal(t, []string{ + `{"data":{"value":"admitted","resolved":"value"}}`, + "", + }, writer.Entries()) + }, + }, + { + name: "Complete", + invoke: func(updater *subscriptionUpdater, _ SubscriptionIdentifier) { + updater.Complete() + }, + assertAbsent: func(t *testing.T, _ *subscriptionUpdater, _ SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, _ <-chan struct{}) { + require.Empty(t, writer.Entries(), "complete frame was written before admitted update completed") + }, + assertEffect: func(t *testing.T, _ *subscriptionUpdater, _ SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, _ <-chan struct{}, _ time.Time) { + require.Equal(t, []string{ + `{"data":{"value":"admitted","resolved":"value"}}`, + "", + }, writer.Entries()) + }, + }, + { + name: "Error", + invoke: func(updater *subscriptionUpdater, _ SubscriptionIdentifier) { + updater.Error([]byte(terminalError)) + }, + assertAbsent: func(t *testing.T, _ *subscriptionUpdater, _ SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, _ <-chan struct{}) { + require.Empty(t, writer.Entries(), "error frame was written before admitted update completed") + }, + assertEffect: func(t *testing.T, _ *subscriptionUpdater, _ SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, _ <-chan struct{}, _ time.Time) { + require.Equal(t, []string{ + `{"data":{"value":"admitted","resolved":"value"}}`, + terminalError, + }, writer.Entries()) + }, + }, + { + name: "CloseSubscription", + invoke: func(updater *subscriptionUpdater, id SubscriptionIdentifier) { + updater.CloseSubscription(id) + }, + assertAbsent: func(t *testing.T, updater *subscriptionUpdater, id SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, _ *chronologicalSubscriptionWriter, completed <-chan struct{}) { + require.True(t, subscriptionUpdaterHasID(updater.Subscriptions(), id), "target subscription was removed before admitted update completed") + select { + case <-completed: + t.Error("target subscription completed before admitted update completed") + default: + } + }, + assertEffect: func(t *testing.T, updater *subscriptionUpdater, id SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, completed <-chan struct{}, deadline time.Time) { + require.Equal(t, []string{`{"data":{"value":"admitted","resolved":"value"}}`}, writer.Entries()) + require.False(t, subscriptionUpdaterHasID(updater.Subscriptions(), id)) + select { + case <-completed: + case <-time.After(time.Until(deadline)): + t.Fatal("timed out waiting for closed subscription completion") + } + }, + }, + { + name: "Done", + invoke: func(updater *subscriptionUpdater, _ SubscriptionIdentifier) { + updater.Done() + }, + assertAbsent: func(t *testing.T, updater *subscriptionUpdater, _ SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, _ *chronologicalSubscriptionWriter, completed <-chan struct{}) { + _, registered := updater.resolver.getTrigger(updater.triggerID) + require.True(t, registered, "trigger detached before admitted update completed") + select { + case <-completed: + t.Error("subscription completed before admitted update completed") + default: + } + }, + assertEffect: func(t *testing.T, updater *subscriptionUpdater, _ SubscriptionIdentifier, _ *concurrentSubscriptionDataSource, writer *chronologicalSubscriptionWriter, completed <-chan struct{}, deadline time.Time) { + require.Equal(t, []string{`{"data":{"value":"admitted","resolved":"value"}}`}, writer.Entries()) + _, registered := updater.resolver.getTrigger(updater.triggerID) + require.False(t, registered) + select { + case <-completed: + case <-time.After(time.Until(deadline)): + t.Fatal("timed out waiting for done subscription completion") + } + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + deadline := time.Now().Add(5 * time.Second) + dataSource := newConcurrentSubscriptionDataSource(1) + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + updater.resolver.heartbeatInterval = 0 + + writer := &chronologicalSubscriptionWriter{} + var completed <-chan struct{} + configureSubscriptionUpdaterState(t, updater, ids[0], func(state *subscriptionState) { + state.writer = writer + state.heartbeat = true + completed = state.completed + }) + _, updateDone := startSubscriptionUpdaterCall(func() { + updater.UpdateSubscription(ids[0], []byte(admittedData)) + }) + var operationDone <-chan struct{} + defer func() { + dataSource.Release() + waitForSubscriptionUpdaterCall(t, updateDone, deadline) + if operationDone != nil { + waitForSubscriptionUpdaterCall(t, operationDone, deadline) + } + }() + + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Until(deadline)): + t.Fatal("timed out waiting for admitted subscription update") + } + + operationStarted, done := startSubscriptionUpdaterCall(func() { + test.invoke(updater, ids[0]) + }) + operationDone = done + waitForSubscriptionUpdaterCallStart(t, operationStarted, deadline) + waitForSubscriptionUpdaterCallToEnter(t, updater, operationDone, deadline) + checkSubscriptionUpdaterCallBlocked(t, operationDone) + test.assertAbsent(t, updater, ids[0], dataSource, writer, completed) + + dataSource.Release() + waitForSubscriptionUpdaterCall(t, updateDone, deadline) + waitForSubscriptionUpdaterCall(t, operationDone, deadline) + test.assertEffect(t, updater, ids[0], dataSource, writer, completed, deadline) + }) + } +} + +func TestSubscriptionUpdater_Terminal_FirstTerminalWinsAndDoneRemainsAvailable(t *testing.T) { + const terminalError = `{"errors":[{"message":"terminal"}]}` + + tests := []struct { + name string + win func(*subscriptionUpdater) + winningTerminal string + }{ + { + name: "CompleteFirst", + win: func(updater *subscriptionUpdater) { + updater.Complete() + }, + winningTerminal: "", + }, + { + name: "ErrorFirst", + win: func(updater *subscriptionUpdater) { + updater.Error([]byte(terminalError)) + }, + winningTerminal: terminalError, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + deadline := time.Now().Add(5 * time.Second) + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 2, dataSource) + defer cleanup() + updater.resolver.heartbeatInterval = 0 + + writers := make([]*chronologicalSubscriptionWriter, len(ids)) + completed := make([]<-chan struct{}, len(ids)) + for i, id := range ids { + writers[i] = &chronologicalSubscriptionWriter{} + configureSubscriptionUpdaterState(t, updater, id, func(state *subscriptionState) { + state.writer = writers[i] + state.heartbeat = true + completed[i] = state.completed + }) + } + + test.win(updater) + for _, writer := range writers { + require.Equal(t, []string{test.winningTerminal}, writer.Entries()) + } + + fetchesBeforeSuppressedCalls := dataSource.Started() + terminalCalls := []struct { + name string + call func() + }{ + {name: "Error", call: func() { updater.Error([]byte(`{"errors":[{"message":"later"}]}`)) }}, + {name: "Complete", call: updater.Complete}, + {name: "Update", call: func() { updater.Update([]byte(`{"data":{"value":"broadcast"}}`)) }}, + {name: "UpdateSubscription", call: func() { updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"targeted"}}`)) }}, + {name: "Heartbeat", call: updater.Heartbeat}, + } + for _, terminalCall := range terminalCalls { + requireSubscriptionUpdaterCallShortCircuitsWithoutResolver(t, updater, terminalCall.name, terminalCall.call) + } + require.Equal(t, fetchesBeforeSuppressedCalls, dataSource.Started(), "terminal Update or UpdateSubscription reached the data source") + for _, writer := range writers { + require.Equal(t, []string{test.winningTerminal}, writer.Entries(), "later updater calls changed a terminal writer") + } + + updater.CloseSubscription(ids[0]) + subscriptions := updater.Subscriptions() + require.Len(t, subscriptions, 1) + require.True(t, subscriptionUpdaterHasID(subscriptions, ids[1])) + _, triggerRegistered := updater.resolver.getTrigger(updater.triggerID) + require.True(t, triggerRegistered) + select { + case <-completed[0]: + case <-time.After(time.Until(deadline)): + t.Fatal("timed out waiting for terminal close completion") + } + select { + case <-completed[1]: + t.Fatal("closing one terminal subscription completed the remaining subscription") + default: + } + + updater.Done() + _, triggerRegistered = updater.resolver.getTrigger(updater.triggerID) + require.False(t, triggerRegistered) + select { + case <-completed[1]: + case <-time.After(time.Until(deadline)): + t.Fatal("timed out waiting for remaining terminal subscription completion") + } + + updater.terminal = false + fetchesBeforeDoneCalls := dataSource.Started() + postDoneCalls := []struct { + name string + call func() + }{ + {name: "Update", call: func() { updater.Update([]byte(`{"data":{"value":"after-done"}}`)) }}, + {name: "UpdateSubscription", call: func() { updater.UpdateSubscription(ids[1], []byte(`{"data":{"value":"after-done-targeted"}}`)) }}, + {name: "Heartbeat", call: updater.Heartbeat}, + {name: "Complete", call: updater.Complete}, + {name: "Error", call: func() { updater.Error([]byte(`{"errors":[{"message":"after done"}]}`)) }}, + {name: "CloseSubscription", call: func() { updater.CloseSubscription(ids[1]) }}, + {name: "Done", call: updater.Done}, + } + for _, postDoneCall := range postDoneCalls { + requireSubscriptionUpdaterCallShortCircuitsWithoutResolver(t, updater, postDoneCall.name, postDoneCall.call) + } + require.Equal(t, fetchesBeforeDoneCalls, dataSource.Started(), "post-Done Update or UpdateSubscription reached the data source") + for _, writer := range writers { + require.Equal(t, []string{test.winningTerminal}, writer.Entries(), "post-Done updater call changed a writer") + } + + readOnlySnapshot := updater.Subscriptions() + require.Empty(t, readOnlySnapshot) + readOnlySnapshot[context.Background()] = SubscriptionIdentifier{ConnectionID: 99, SubscriptionID: 99} + require.Empty(t, updater.Subscriptions(), "mutating the returned subscription snapshot changed updater state") + }) + } +} + +func TestSubscriptionUpdater_Terminal_RejectsLateSubscriber(t *testing.T) { + const terminalError = `{"errors":[{"message":"terminal"}]}` + tests := []struct { + name string + expectedEntries []string + detaches bool + deliver func(*subscriptionUpdater) + }{ + { + name: "Complete", + expectedEntries: []string{""}, + deliver: func(updater *subscriptionUpdater) { + updater.Complete() + }, + }, + { + name: "Error", + expectedEntries: []string{terminalError}, + deliver: func(updater *subscriptionUpdater) { + updater.Error([]byte(terminalError)) + }, + }, + { + name: "Done", + detaches: true, + deliver: func(updater *subscriptionUpdater) { + updater.Done() + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + deadline := time.Now().Add(5 * time.Second) + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + reporter := &subscriptionUpdateCountingReporter{} + reporter.SubscriptionCountInc(1) + updater.resolver.reporter = reporter + reporterBeforeAdd := reporter.snapshot() + existingWriter := &chronologicalSubscriptionWriter{} + var existingCompleted <-chan struct{} + configureSubscriptionUpdaterState(t, updater, ids[0], func(state *subscriptionState) { + state.writer = existingWriter + existingCompleted = state.completed + }) + + updater.updateWG.Add(1) + barrierReleased := false + defer func() { + if !barrierReleased { + updater.updateWG.Done() + } + }() + _, terminalDone := startSubscriptionUpdaterCall(func() { + test.deliver(updater) + }) + waitForSubscriptionUpdaterTerminating(t, updater, terminalDone, deadline) + + hookStarted := make(chan struct{}, 1) + source := createFakeStream(nil, 0, nil, func(StartupHookContext, []byte) error { + hookStarted <- struct{}{} + return nil + }) + lateID := SubscriptionIdentifier{ConnectionID: 99, SubscriptionID: 99} + lateCompleted := make(chan struct{}) + addErr := updater.resolver.addSubscription(updater.triggerID, &addSubscription{ + ctx: NewContext(t.Context()), + resolve: &GraphQLSubscription{ + Trigger: GraphQLSubscriptionTrigger{Source: source}, + Response: &GraphQLResponse{}, + }, + writer: &chronologicalSubscriptionWriter{}, + id: lateID, + completed: lateCompleted, + }) + + require.ErrorIs(t, addErr, ErrSubscriptionTriggerTerminating) + require.EqualError(t, addErr, "subscription trigger is no longer accepting subscriptions") + require.Equal(t, reporterBeforeAdd, reporter.snapshot(), "rejected subscriber changed reporter state") + require.False(t, subscriptionUpdaterHasID(updater.Subscriptions(), lateID)) + updater.resolver.mu.Lock() + _, indexedByID := updater.resolver.subscriptionsByID[lateID] + _, indexedByConnection := updater.resolver.subscriptionsByConnection[lateID.ConnectionID] + updater.resolver.mu.Unlock() + require.False(t, indexedByID) + require.False(t, indexedByConnection) + select { + case <-hookStarted: + t.Fatal("startup hook launched for rejected subscriber") + default: + } + select { + case <-lateCompleted: + t.Fatal("rejected subscriber was attached before lifecycle release") + default: + } + + updater.updateWG.Done() + barrierReleased = true + waitForSubscriptionUpdaterCall(t, terminalDone, deadline) + select { + case <-hookStarted: + t.Fatal("startup hook launched for rejected subscriber") + case <-time.After(50 * time.Millisecond): + } + + require.Equal(t, test.expectedEntries, existingWriter.Entries()) + if !test.detaches { + _, triggerRegistered := updater.resolver.getTrigger(updater.triggerID) + require.True(t, triggerRegistered) + select { + case <-existingCompleted: + t.Fatal("terminal delivery completed the existing subscriber before Done") + default: + } + updater.Done() + } else { + _, triggerRegistered := updater.resolver.getTrigger(updater.triggerID) + require.False(t, triggerRegistered) + } + select { + case <-existingCompleted: + case <-time.After(time.Until(deadline)): + t.Fatal("timed out waiting for existing subscriber completion") + } + select { + case <-lateCompleted: + t.Fatal("rejected subscriber was attached and completed by lifecycle cleanup") + default: + } + require.Zero(t, reporter.snapshot().subscriptions) + }) + } +} + +func TestSubscriptionUpdater_Terminal_IncludesSubscriberRegisteredFirst(t *testing.T) { + const terminalError = `{"errors":[{"message":"terminal"}]}` + tests := []struct { + name string + expectedEntries []string + detaches bool + deliver func(*subscriptionUpdater) + }{ + { + name: "Complete", + expectedEntries: []string{""}, + deliver: func(updater *subscriptionUpdater) { + updater.Complete() + }, + }, + { + name: "Error", + expectedEntries: []string{terminalError}, + deliver: func(updater *subscriptionUpdater) { + updater.Error([]byte(terminalError)) + }, + }, + { + name: "Done", + detaches: true, + deliver: func(updater *subscriptionUpdater) { + updater.Done() + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, _, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + hookStarted := make(chan struct{}) + source := createFakeStream(nil, 0, nil, func(StartupHookContext, []byte) error { + close(hookStarted) + return nil + }) + registeredID := SubscriptionIdentifier{ConnectionID: 99, SubscriptionID: 99} + registeredWriter := &chronologicalSubscriptionWriter{} + registeredCompleted := make(chan struct{}) + err := updater.resolver.addSubscription(updater.triggerID, &addSubscription{ + ctx: NewContext(t.Context()), + resolve: &GraphQLSubscription{ + Trigger: GraphQLSubscriptionTrigger{Source: source}, + Response: &GraphQLResponse{}, + }, + writer: registeredWriter, + id: registeredID, + completed: registeredCompleted, + }) + require.NoError(t, err) + select { + case <-hookStarted: + case <-time.After(time.Second): + t.Fatal("timed out waiting for registered subscriber startup hook") + } + require.True(t, subscriptionUpdaterHasID(updater.Subscriptions(), registeredID)) + + test.deliver(updater) + + require.Equal(t, test.expectedEntries, registeredWriter.Entries()) + if !test.detaches { + select { + case <-registeredCompleted: + t.Fatal("terminal delivery completed the registered subscriber before Done") + default: + } + updater.Done() + } + select { + case <-registeredCompleted: + case <-time.After(time.Second): + t.Fatal("registered subscriber was not included in lifecycle cleanup") + } + }) + } +} + +func TestSubscriptionUpdater_Lifecycle_CanceledTriggerContextGuardsCallbacks(t *testing.T) { + deadline := time.Now().Add(5 * time.Second) + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 2, dataSource) + defer cleanup() + updater.resolver.heartbeatInterval = 0 + + writers := make([]*chronologicalSubscriptionWriter, len(ids)) + completed := make([]<-chan struct{}, len(ids)) + for i, id := range ids { + writers[i] = &chronologicalSubscriptionWriter{} + configureSubscriptionUpdaterState(t, updater, id, func(state *subscriptionState) { + state.writer = writers[i] + state.heartbeat = true + completed[i] = state.completed + }) + } + + cancelSubscriptionUpdater(t, updater) + canceledContextCalls := []struct { + name string + call func() + }{ + {name: "Update", call: func() { updater.Update([]byte(`{"data":{"value":"canceled"}}`)) }}, + {name: "UpdateSubscription", call: func() { updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"canceled-targeted"}}`)) }}, + {name: "Heartbeat", call: updater.Heartbeat}, + {name: "Complete", call: updater.Complete}, + {name: "Error", call: func() { updater.Error([]byte(`{"errors":[{"message":"canceled"}]}`)) }}, + {name: "CloseSubscription", call: func() { updater.CloseSubscription(ids[0]) }}, + } + for _, canceledContextCall := range canceledContextCalls { + requireSubscriptionUpdaterCallShortCircuitsWithoutResolver(t, updater, canceledContextCall.name, canceledContextCall.call) + } + + require.Zero(t, dataSource.Started()) + trig, triggerRegistered := updater.resolver.getTrigger(updater.triggerID) + require.True(t, triggerRegistered) + require.Len(t, trig.snapshotSubscriptions(), len(ids), "canceled-context callback changed registrations") + for i, writer := range writers { + require.Empty(t, writer.Entries()) + select { + case <-completed[i]: + t.Fatalf("canceled-context callback completed subscription %d", i) + default: + } + } + + updater.Done() + _, triggerRegistered = updater.resolver.getTrigger(updater.triggerID) + require.False(t, triggerRegistered) + for i, completion := range completed { + select { + case <-completion: + case <-time.After(time.Until(deadline)): + t.Fatalf("timed out waiting for Done to complete canceled-context subscription %d", i) + } + } +} + +func TestSubscriptionUpdater_UpdateSubscription_ChronologicalWriterRecordsControlFrames(t *testing.T) { + writer := &chronologicalSubscriptionWriter{} + + require.NoError(t, writer.Heartbeat()) + writer.Complete() + writer.Error([]byte(`{"errors":[{"message":"terminal"}]}`)) + + require.Equal(t, []string{ + "", + "", + `{"errors":[{"message":"terminal"}]}`, + }, writer.Entries()) +} + +func TestSubscriptionUpdater_UpdateSubscription_CancellationReleasesQueuedSameSubscriberUpdate(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + var callsDone []<-chan struct{} + defer func() { + dataSource.Release() + for _, done := range callsDone { + waitForSubscriptionUpdateCall(t, done) + } + }() + + firstDone := make(chan struct{}) + callsDone = append(callsDone, firstDone) + go func() { + defer close(firstDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"first"}}`)) + }() + + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + t.Fatal("timed out waiting for first subscription fetch") + } + firstTail := subscriptionUpdateTail(updater, ids[0]) + require.NotNil(t, firstTail) + + secondDone := make(chan struct{}) + callsDone = append(callsDone, secondDone) + go func() { + defer close(secondDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"second"}}`)) + }() + waitForSubscriptionUpdateTailChange(t, updater, ids[0], firstTail) + + cancelSubscriptionUpdater(t, updater) + select { + case <-secondDone: + case <-time.After(time.Second): + t.Fatal("queued subscription update did not return after cancellation") + } + require.Equal(t, 1, dataSource.Started(), "queued update entered its fetch after cancellation") + + dataSource.Release() + waitForSubscriptionUpdateCall(t, firstDone) + waitForSubscriptionUpdates(t, updater) + requireNoSubscriptionUpdateTails(t, updater) +} + +func TestSubscriptionUpdater_UpdateSubscription_PropagatesPanicAndCleansTail(t *testing.T) { + panicValue := &subscriptionUpdatePanic{label: "subscription fetch panic"} + dataSource := &panickingSubscriptionDataSource{panicValue: panicValue} + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + recovered := make(chan any, 1) + go func() { + defer func() { + recovered <- recover() + }() + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"event"}}`)) + }() + + select { + case actual := <-recovered: + require.Same(t, panicValue, actual) + case <-time.After(time.Second): + t.Fatal("timed out waiting for subscription update panic") + } + waitForSubscriptionUpdates(t, updater) + requireNoSubscriptionUpdateTails(t, updater) +} + +func TestSubscriptionUpdater_UpdateSubscription_FlushErrorUnsubscribesWithoutDeadlock(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + dataSource.Release() + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + + flushErr := errors.New("subscription flush failed") + writer := &panickingSubscriptionWriter{flushErr: flushErr} + var completed <-chan struct{} + configureSubscriptionUpdaterState(t, updater, ids[0], func(state *subscriptionState) { + state.writer = writer + completed = state.completed + }) + + updateDone := make(chan struct{}) + go func() { + defer close(updateDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"event"}}`)) + }() + waitForSubscriptionUpdateCall(t, updateDone) + select { + case <-completed: + case <-time.After(time.Second): + t.Fatal("flush failure did not complete the removed subscription") + } + + _, triggerRegistered := updater.resolver.getTrigger(updater.triggerID) + require.False(t, triggerRegistered) + updater.resolver.mu.Lock() + _, indexedByID := updater.resolver.subscriptionsByID[ids[0]] + _, indexedByConnection := updater.resolver.subscriptionsByConnection[ids[0].ConnectionID] + updater.resolver.mu.Unlock() + require.False(t, indexedByID) + require.False(t, indexedByConnection) + waitForSubscriptionUpdates(t, updater) + requireNoSubscriptionUpdateTails(t, updater) + require.Empty(t, writer.Entries()) + + cleanup() +} + +func TestSubscriptionUpdater_UpdateSubscription_OrdersFilterErrorAfterPrecedingUpdate(t *testing.T) { + filterErr := errors.New("subscription filter render failed") + dataSource := newConcurrentSubscriptionDataSource(1) + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + writer := &chronologicalSubscriptionWriter{} + renderer := &sequencedSubscriptionFilterRenderer{secondErr: filterErr} + configureSubscriptionUpdaterState(t, updater, ids[0], func(state *subscriptionState) { + renderer.expectedContext = state.ctx.Context() + state.ctx.Variables = astjson.MustParseBytes([]byte(`{"filterValue":"match"}`)) + state.writer = writer + state.resolve.Filter = &SubscriptionFilter{ + In: &SubscriptionFieldFilter{ + FieldPath: []string{"filter"}, + Values: []InputTemplate{{ + Segments: []TemplateSegment{{ + SegmentType: VariableSegmentType, + VariableKind: ResolvableObjectVariableKind, + VariableSourcePath: []string{"filterValue"}, + Renderer: renderer, + }}, + }}, + }, + } + }) + updater.resolver.SetAsyncErrorWriter(&chronologicalSubscriptionErrorWriter{}) + + var callsDone []<-chan struct{} + defer func() { + dataSource.Release() + for _, done := range callsDone { + waitForSubscriptionUpdateCall(t, done) + } + }() + + firstDone := make(chan struct{}) + callsDone = append(callsDone, firstDone) + go func() { + defer close(firstDone) + updater.UpdateSubscription(ids[0], []byte(`{"filter":"match","data":{"value":"first"}}`)) + }() + + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + t.Fatal("timed out waiting for first subscription fetch") + } + firstTail := subscriptionUpdateTail(updater, ids[0]) + require.NotNil(t, firstTail) + + secondDone := make(chan struct{}) + callsDone = append(callsDone, secondDone) + go func() { + defer close(secondDone) + updater.UpdateSubscription(ids[0], []byte(`{"filter":"match","data":{"value":"second"}}`)) + }() + waitForSubscriptionUpdateTailChange(t, updater, ids[0], firstTail) + require.Empty(t, writer.Entries(), "queued filter error was written before the preceding update completed") + + dataSource.Release() + waitForSubscriptionUpdateCall(t, firstDone) + waitForSubscriptionUpdateCall(t, secondDone) + require.Equal(t, []string{ + `{"data":{"value":"first","resolved":"value"}}`, + `{"errors":[{"message":"subscription filter render failed"}]}`, + }, writer.Entries()) + calls, sawUnexpectedContext, sawNonNilEventData := renderer.results() + require.Equal(t, 2, calls) + require.False(t, sawUnexpectedContext) + require.False(t, sawNonNilEventData) +} + +func TestSubscriptionUpdater_UpdateSubscription_CleansTailAfterSingleUpdate(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + done := make(chan struct{}) + defer func() { + dataSource.Release() + waitForSubscriptionUpdateCall(t, done) + }() + go func() { + defer close(done) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"single"}}`)) + }() + + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + t.Fatal("timed out waiting for subscription fetch") + } + require.NotNil(t, subscriptionUpdateTail(updater, ids[0])) + + dataSource.Release() + waitForSubscriptionUpdateCall(t, done) + requireNoSubscriptionUpdateTails(t, updater) +} + +func TestSubscriptionUpdater_UpdateSubscription_CleansTailAfterOverlappingSameSubscriberUpdates(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + updater, ids, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + var callsDone []<-chan struct{} + defer func() { + dataSource.Release() + for _, done := range callsDone { + waitForSubscriptionUpdateCall(t, done) + } + }() + + firstDone := make(chan struct{}) + callsDone = append(callsDone, firstDone) + go func() { + defer close(firstDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"first"}}`)) + }() + + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + t.Fatal("timed out waiting for first subscription fetch") + } + firstTail := subscriptionUpdateTail(updater, ids[0]) + require.NotNil(t, firstTail) + + secondDone := make(chan struct{}) + callsDone = append(callsDone, secondDone) + go func() { + defer close(secondDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"second"}}`)) + }() + waitForSubscriptionUpdateTailChange(t, updater, ids[0], firstTail) + + dataSource.Release() + waitForSubscriptionUpdateCall(t, firstDone) + waitForSubscriptionUpdateCall(t, secondDone) + requireNoSubscriptionUpdateTails(t, updater) +} + +func TestSubscriptionUpdater_UpdateSubscription_PreservesSameSubscriberFIFO(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + updater, ids, recorders, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + var callsDone []<-chan struct{} + defer func() { + dataSource.Release() + for _, done := range callsDone { + waitForSubscriptionUpdateCall(t, done) + } + }() + + firstDone := make(chan struct{}) + callsDone = append(callsDone, firstDone) + go func() { + defer close(firstDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"first"}}`)) + }() + + select { + case started := <-dataSource.StartedEvents(): + require.Equal(t, 1, started) + case <-time.After(time.Second): + t.Fatal("timed out waiting for first subscription fetch") + } + firstTail := subscriptionUpdateTail(updater, ids[0]) + require.NotNil(t, firstTail) + + secondDone := make(chan struct{}) + callsDone = append(callsDone, secondDone) + go func() { + defer close(secondDone) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"second"}}`)) + }() + + waitForSubscriptionUpdateTailChange(t, updater, ids[0], firstTail) + select { + case started := <-dataSource.StartedEvents(): + t.Fatalf("fetch %d began before the preceding update was released", started) + case <-time.After(250 * time.Millisecond): + } + + dataSource.Release() + select { + case started := <-dataSource.StartedEvents(): + require.Equal(t, 2, started) + case <-time.After(time.Second): + t.Fatal("timed out waiting for second subscription fetch after releasing the first") + } + waitForSubscriptionUpdateCall(t, firstDone) + waitForSubscriptionUpdateCall(t, secondDone) + + require.Equal(t, []string{ + `{"data":{"value":"first","resolved":"value"}}`, + `{"data":{"value":"second","resolved":"value"}}`, + }, recorders[0].Messages()) +} + +func TestSubscriptionUpdater_UpdateSubscription_UnknownSubscriberIsNoOp(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + updater, _, _, cleanup := newSubscriptionUpdaterHarness(t, 1, dataSource) + defer cleanup() + + unknownID := SubscriptionIdentifier{ConnectionID: 99, SubscriptionID: 99} + updater.UpdateSubscription(unknownID, []byte(`{"data":{"value":"ignored"}}`)) + + require.Zero(t, dataSource.Started()) + requireNoSubscriptionUpdateTails(t, updater) +} + +func TestSubscriptionUpdater_UpdateSubscription_ResolvesDistinctSubscribersConcurrently(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(2) + updater, ids, recorders, cleanup := newSubscriptionUpdaterHarness(t, 2, dataSource) + defer cleanup() + + var wg sync.WaitGroup + wg.Add(len(ids)) + for _, id := range ids { + go func(id SubscriptionIdentifier) { + defer wg.Done() + updater.UpdateSubscription(id, []byte(`{"data":{"value":"event"}}`)) + }(id) + } + callsDone := make(chan struct{}) + go func() { + wg.Wait() + close(callsDone) + }() + var callsCleanupOnce sync.Once + callsJoined := false + cleanupCalls := func() { + callsCleanupOnce.Do(func() { + dataSource.Release() + select { + case <-callsDone: + callsJoined = true + case <-time.After(time.Second): + t.Error("timed out waiting for subscription updates to finish") + } + }) + } + defer cleanupCalls() + + startedConcurrently := true + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + startedConcurrently = false + } + + cleanupCalls() + if !callsJoined { + return + } + + for _, recorder := range recorders { + require.Equal(t, []string{`{"data":{"value":"event","resolved":"value"}}`}, recorder.Messages()) + } + + if !startedConcurrently { + t.Fatal("distinct subscribers did not enter their fetches concurrently") + } +} + +func TestSubscriptionUpdater_UpdateSubscription_DeduplicatesConcurrentSubscriberFetches(t *testing.T) { + dataSource := newConcurrentSubscriptionDataSource(1) + updater, ids, recorders, cleanup := newSubscriptionUpdaterHarness(t, 2, dataSource) + defer cleanup() + + waitObserved := make([]chan struct{}, len(ids)) + for i, id := range ids { + waitObserved[i] = make(chan struct{}) + configureSubscriptionUpdaterState(t, updater, id, func(state *subscriptionState) { + state.ctx.ExecutionOptions.DisableSubgraphRequestDeduplication = false + state.ctx.LoaderHooks = &subscriptionSingleFlightLoaderHook{waitObserved: waitObserved[i]} + }) + } + + updateDone := make([]chan struct{}, len(ids)) + for i := range updateDone { + updateDone[i] = make(chan struct{}) + } + defer func() { + dataSource.Release() + for _, done := range updateDone { + waitForSubscriptionUpdateCall(t, done) + } + }() + + go func() { + defer close(updateDone[0]) + updater.UpdateSubscription(ids[0], []byte(`{"data":{"value":"first"}}`)) + }() + waitForSubscriptionUpdateTailChange(t, updater, ids[0], nil) + select { + case <-dataSource.AllStarted(): + case <-time.After(time.Second): + t.Fatal("timed out waiting for single-flight leader fetch") + } + + go func() { + defer close(updateDone[1]) + updater.UpdateSubscription(ids[1], []byte(`{"data":{"value":"second"}}`)) + }() + waitForSubscriptionUpdateTailChange(t, updater, ids[1], nil) + select { + case <-waitObserved[1]: + case <-time.After(time.Second): + t.Fatal("timed out waiting for second subscriber to join the single-flight fetch") + } + + require.Equal(t, 1, dataSource.Started()) + dataSource.Release() + for _, done := range updateDone { + waitForSubscriptionUpdateCall(t, done) + } + require.Equal(t, 1, dataSource.Started()) + require.Equal(t, []string{`{"data":{"value":"first","resolved":"value"}}`}, recorders[0].Messages()) + require.Equal(t, []string{`{"data":{"value":"second","resolved":"value"}}`}, recorders[1].Messages()) +}