fix(resolve): resolve subscription updates concurrently in order#1602
Open
jensneuse wants to merge 3 commits into
Open
fix(resolve): resolve subscription updates concurrently in order#1602jensneuse wants to merge 3 commits into
jensneuse wants to merge 3 commits into
Conversation
Contributor
📝 WalkthroughWalkthroughSubscription resolution now admits updates concurrently across subscribers while preserving FIFO ordering per subscriber. Lifecycle operations wait for admitted updates, terminal frames suppress later writes, terminating triggers reject new subscribers, and tests cover concurrency, cleanup, cancellation, and failure paths. ChangesOrdered subscription updates
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Trigger
participant subscriptionUpdater
participant Resolver
participant subscriptionState
participant Writer
Trigger->>subscriptionUpdater: UpdateSubscription
subscriptionUpdater->>Resolver: resolve admitted subscriber update
Resolver->>subscriptionState: ordered write or terminal check
subscriptionState->>Writer: emit data or control frame
Trigger->>subscriptionUpdater: Complete or Error
subscriptionUpdater->>subscriptionUpdater: wait for admitted updates
subscriptionUpdater->>Resolver: deliver terminal lifecycle event
Resolver->>Writer: emit terminal frame
🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@v2/pkg/engine/resolve/resolve.go`:
- Around line 1074-1076: Synchronize the removal state with the downstream write
phase in resolve.go around the subscription write path guarded by sub.writeMu,
so Resolve or Flush cannot continue after unsubscribe returns; alternatively
stage output and atomically discard it if removal wins. Add a
subscription_updater_test.go case covering unsubscribe after the final write
section begins, and update the design spec and implementation plan at the cited
ranges to document this synchronization requirement and post-check race
coverage.
- Around line 980-987: Update subscription error-writing flows around
subscriptionState.writeError and the referenced send paths to return whether an
error frame was actually written, returning false when removed or terminal and
true after WriteError succeeds. Gate every SubscriptionUpdateSent() call on that
sent result, including the additional paths noted in the review, while
preserving existing locking and output behavior.
In `@v2/pkg/engine/resolve/subscription_updater_test.go`:
- Line 14: Fix the lint issues in subscription_updater_test.go by running the
repository’s gci formatter, inserting blank lines between embedded fields in the
affected declarations, and applying the rangeint-required loop fixes at the
referenced locations. Preserve the existing test behavior while updating all
indicated occurrences, including the additional reported ranges.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 3008777d-c8e4-43e0-874c-6f4688024c7d
📒 Files selected for processing (4)
docs/superpowers/plans/2026-07-17-ordered-concurrent-subscription-updates.mddocs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.mdv2/pkg/engine/resolve/resolve.gov2/pkg/engine/resolve/subscription_updater_test.go
Comment on lines
980
to
987
| 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) | ||
| } |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Count formatted errors only when a frame was written.
writeError now suppresses output after removal or termination, but these three paths still increment SubscriptionUpdateSent. Return a sent flag, as sendHeartbeat does, and gate the reporter increment.
Proposed fix
-func (s *subscriptionState) writeError(w AsyncErrorWriter, ctx *Context, err error, response *GraphQLResponse) {
+func (s *subscriptionState) writeError(w AsyncErrorWriter, ctx *Context, err error, response *GraphQLResponse) bool {
s.writeMu.Lock()
defer s.writeMu.Unlock()
if s.removed.Load() || s.terminal {
- return
+ return false
}
w.WriteError(ctx, err, response, s.writer)
+ return true
}Use the returned value to guard each SubscriptionUpdateSent() call.
Also applies to: 1031-1039, 1042-1050, 1059-1067
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@v2/pkg/engine/resolve/resolve.go` around lines 980 - 987, Update subscription
error-writing flows around subscriptionState.writeError and the referenced send
paths to return whether an error frame was actually written, returning false
when removed or terminal and true after WriteError succeeds. Gate every
SubscriptionUpdateSent() call on that sent result, including the additional
paths noted in the review, while preserving existing locking and output
behavior.
Comment on lines
+1074
to
+1076
| sub.writeMu.Lock() | ||
| defer sub.writeMu.Unlock() | ||
| if sub.removed.Load() || sub.terminal { |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Linearize unsubscribe with the downstream write phase.
removed can change after the one-time check at Line 1076, allowing Resolve or Flush to continue after unsubscribe returns.
v2/pkg/engine/resolve/resolve.go#L1074-L1076: synchronize removal withwriteMu, or stage output and atomically discard it when removal wins.v2/pkg/engine/resolve/subscription_updater_test.go#L1021-L1079: add a case that unsubscribes after the final write section has begun.docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md#L78-L78: document the required write/removal synchronization.docs/superpowers/plans/2026-07-17-ordered-concurrent-subscription-updates.md#L286-L286: require coverage of this post-check race window.
📍 Affects 4 files
v2/pkg/engine/resolve/resolve.go#L1074-L1076(this comment)v2/pkg/engine/resolve/subscription_updater_test.go#L1021-L1079docs/superpowers/specs/2026-07-17-ordered-concurrent-subscription-updates-design.md#L78-L78docs/superpowers/plans/2026-07-17-ordered-concurrent-subscription-updates.md#L286-L286
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@v2/pkg/engine/resolve/resolve.go` around lines 1074 - 1076, Synchronize the
removal state with the downstream write phase in resolve.go around the
subscription write path guarded by sub.writeMu, so Resolve or Flush cannot
continue after unsubscribe returns; alternatively stage output and atomically
discard it if removal wins. Add a subscription_updater_test.go case covering
unsubscribe after the final write section begins, and update the design spec and
implementation plan at the cited ranges to document this synchronization
requirement and post-check race coverage.
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fix the reported lint failures.
Run the repository’s gci formatter, separate embedded fields with blank lines, and apply the rangeint fixes at Lines 350 and 698.
type panickingSubscriptionWriter struct {
chronologicalSubscriptionWriter
+
writeErr error
}
type subscriptionSingleFlightWaitContext struct {
context.Context
+
waitObserved chan struct{}
}
-for i := 0; i < count; i++ {
+for i := range count {
-for i := 0; i < 32; i++ {
+for i := range 32 {Also applies to: 127-134, 170-174, 350-350, 698-698
🧰 Tools
🪛 GitHub Check: Linters (1.25)
[failure] 14-14:
File is not properly formatted (gci)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@v2/pkg/engine/resolve/subscription_updater_test.go` at line 14, Fix the lint
issues in subscription_updater_test.go by running the repository’s gci
formatter, inserting blank lines between embedded fields in the affected
declarations, and applying the rangeint-required loop fixes at the referenced
locations. Preserve the existing test behavior while updating all indicated
occurrences, including the additional reported ranges.
Source: Linters/SAST tools
This was referenced Jul 18, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
@coderabbitai summary
Summary
Motivation
UpdateSubscriptionheld a trigger-wide mutex across filtering, resolution, subgraph fetches, writes, and flushes. Fan-out updates for different subscribers therefore ran serially, causing head-of-line blocking and preventing identical in-flight subgraph fetches from sharing a response.This change replaces that trigger-wide resolution serialization with keyed per-subscription admission lanes. Different subscribers can resolve concurrently, while each subscriber retains deterministic update order and lifecycle operations cannot overtake admitted work.
Verification
go test ./... -count=1go test -race ./pkg/engine/resolve -count=1go vet ./...Checklist
Open Source AI Manifesto
This contribution follows the principles of the Open Source AI Manifesto.