feat: improved subscription heartbeats#1269
Conversation
endigma
requested review from
a team,
Noroth,
StarpTech,
devsergiy and
jensneuse
as code owners
WalkthroughAdds a Heartbeat() method to subscription writers and their test doubles, renames ResolverOptions.MultipartSubHeartbeatInterval to SubscriptionHeartbeatInterval, and changes resolver heartbeat flow to invoke writer.Heartbeat() with early-cancellation checks and a resettable ticker instead of writing multipart payloads. Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Tip 🔌 Remote MCP (Model Context Protocol) integration is now available!Pro plan users can now connect to remote MCP servers from the Integrations page. Connect with popular remote MCPs such as Notion and Linear to add more context to your reviews and chats. ✨ Finishing Touches
🧪 Generate unit tests
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR/Issue comments)Type Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🔭 Outside diff range comments (2)
v2/pkg/engine/resolve/response.go (1)
58-64: Clarify Heartbeat() contract (flush semantics, error signaling, concurrency).Right now it's unclear whether Heartbeat() must also flush, and which error to return on client disconnect. Given the new usage path depends on Heartbeat() to replace Write+Flush, please document and standardize:
- Implementations must write and flush a protocol-specific heartbeat immediately.
- On client disconnect, return an error for which errors.Is(err, context.Canceled) is true (or broaden resolver handling; see related comment).
- For protocols without heartbeats, it may be a no-op returning nil.
- Must be called from the writer goroutine only.
Apply this diff to document expectations:
type SubscriptionResponseWriter interface { ResponseWriter Flush() error Complete() - Heartbeat() error + // Heartbeat sends a protocol-specific heartbeat frame to keep the connection alive. + // Implementations should: + // - write the heartbeat frame and flush it immediately, + // - return an error comparable with context.Canceled (errors.Is(err, context.Canceled) == true) + // when the client connection is gone, + // - return nil for protocols where heartbeats are not applicable (no-op). + // Note: Heartbeat is invoked on the writer goroutine and must not block for long. + Heartbeat() error Close(kind SubscriptionCloseKind) }v2/pkg/engine/resolve/resolve.go (1)
499-525: Improve heartbeat error handling in handleHeartbeat
- Unsubscribe on common network‐closure errors, not just context.Canceled. For example: io.EOF, syscall.EPIPE, net.ErrClosed, etc.
- Avoid calling WriteError with a nil *GraphQLResponse (current test writers ignore it, but future implementations may dereference it). If you still want to report heartbeat errors, pass
sub.resolve.Responsewhen non-nil.- In many cases it’s better to simply unsubscribe and stop heartbeating rather than attempting to send an error over a broken connection.
Locations to update:
- v2/pkg/engine/resolve/resolve.go → func
(*Resolver) handleHeartbeat(around lines 499–525)Suggested diff sketch:
func (r *Resolver) handleHeartbeat(sub *sub) { … if err := sub.writer.Heartbeat(); err != nil { - if errors.Is(err, context.Canceled) { + // Treat common disconnect signals as end-of-subscription + if errors.Is(err, context.Canceled) || + errors.Is(err, io.EOF) || + errors.Is(err, syscall.EPIPE) /* add other errno errors as needed */ { _ = r.AsyncUnsubscribeSubscription(sub.id) return } - r.asyncErrorWriter.WriteError(sub.ctx, err, nil, sub.writer) + // Optionally report the error if you really need it—pass a non-nil response to avoid panics: + if sub.resolve != nil && sub.resolve.Response != nil { + r.asyncErrorWriter.WriteError(sub.ctx, err, sub.resolve.Response, sub.writer) + } } }
🧹 Nitpick comments (4)
execution/graphql/result_writer.go (1)
38-40: Document the no-op Heartbeat to avoid future misuse.This satisfies the interface but makes heartbeats a silent no-op for this writer. If EngineResultWriter is never used for streaming subscriptions that require heartbeats, this is fine; otherwise it will appear as if heartbeats are being sent when they aren’t.
Add a short comment to clarify intent:
func (e *EngineResultWriter) Heartbeat() error { + // No-op: EngineResultWriter is a buffer-based writer and does not stream heartbeats. + // Protocol-specific writers (e.g. multipart, SSE, WS) must implement the actual heartbeat behavior. return nil }v2/pkg/engine/resolve/resolve.go (3)
526-529: Misleading debug message: “flushed” though no explicit flush here.Since flush responsibility moved into writer.Heartbeat(), this log line can mislead when an implementation is a no-op. Consider renaming to “sent” or logging only after Heartbeat returns nil for implementations that actually flushed.
- fmt.Printf("resolver:heartbeat:subscription:flushed:%d\n", sub.id.SubscriptionID) + fmt.Printf("resolver:heartbeat:subscription:sent:%d\n", sub.id.SubscriptionID)
530-533: Heartbeats counted as “SubscriptionUpdateSent” may skew metrics.If metrics are intended to reflect data updates, counting heartbeats will inflate numbers. Consider a dedicated metric (e.g., SubscriptionHeartbeatSent) or skip reporter calls for heartbeats.
I can refactor Reporter to include a separate heartbeat counter and update call sites—want me to draft that?
312-341: Timer per subscription scalability: keep TODO visible and consider a shared scheduler.Acknowledged in the TODO. For many concurrent subscriptions, a ticker per sub is costly. A shared timer wheel or sharded scheduler can reduce overhead substantially.
Happy to draft a sharded ticker utility that feeds interested subs at interval ticks.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
execution/graphql/result_writer.go(1 hunks)v2/pkg/engine/resolve/resolve.go(3 hunks)v2/pkg/engine/resolve/response.go(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
🔇 Additional comments (2)
v2/pkg/engine/resolve/resolve.go (2)
329-339: LGTM: updated call site matches new signature.Switching to handleHeartbeat(s) aligns with the new writer-driven heartbeat mechanism.
500-516: Early-cancellation checks are good; consider combining with the select to avoid redundant checks.The select already listens to r.ctx.Done() and sub.ctx.Done() in the worker loop. These guards are fine, but redundant. Leaving as-is is acceptable due to negligible cost.
endigma
force-pushed
the
jesse/eng-7547-heartbeat-on-sse
branch
2 times, most recently
from
0f9f721 to
bf139fd
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
v2/pkg/engine/resolve/resolve_test.go (1)
4780-4782: Optional: instrument heartbeats for assertionsIf you want to assert that heartbeats were triggered without affecting payloads, increment a counter inside Heartbeat():
func (s *SubscriptionRecorder) Heartbeat() error { - return nil + s.mux.Lock() + defer s.mux.Unlock() + // track heartbeats for test assertions + if s.onFlush != nil { + // keep behavior no-op for payloads + } + return nil }Additions outside the selected range to support this (if desired):
// add to SubscriptionRecorder struct: heartbeats int // add getter for tests: func (s *SubscriptionRecorder) HeartbeatCount() int { s.mux.Lock() defer s.mux.Unlock() return s.heartbeats }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these settings in your CodeRabbit configuration.
📒 Files selected for processing (5)
execution/graphql/result_writer.go(1 hunks)v2/pkg/engine/resolve/event_loop_test.go(1 hunks)v2/pkg/engine/resolve/resolve.go(3 hunks)v2/pkg/engine/resolve/resolve_test.go(1 hunks)v2/pkg/engine/resolve/response.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (3)
- v2/pkg/engine/resolve/response.go
- execution/graphql/result_writer.go
- v2/pkg/engine/resolve/resolve.go
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
🔇 Additional comments (2)
v2/pkg/engine/resolve/resolve_test.go (1)
4780-4782: Add no-op Heartbeat to SubscriptionRecorder: LGTMNo-op Heartbeat() keeps tests from emitting heartbeat payloads while satisfying the updated SubscriptionResponseWriter interface. The method is concurrency-safe as implemented (does nothing) and aligns with tests that assert no heartbeats are written.
v2/pkg/engine/resolve/event_loop_test.go (1)
54-56: Add no-op Heartbeat to FakeSubscriptionWriter: LGTMThe no-op Heartbeat() maintains interface compliance and avoids introducing extra messages in tests. This is appropriate given the resolver heartbeat path now calls writer.Heartbeat() and tests don’t assert on heartbeat payloads.
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (2)
v2/pkg/engine/resolve/resolve.go (2)
201-202: Nit: field comment still says “Multipart heartbeat interval”Now that the option is generalized, update the field comment for accuracy.
Outside this range, update the comment at the field declaration:
// Subscription heartbeat interval heartbeatInterval time.Duration
309-339: Nit: outdated comment mentions “multipart” onlyHeartbeat is now delegated to the writer. Tweak the comment accordingly.
-// startWorkerWithHeartbeat is similar to startWorker but sends heartbeats to the client when -// subscription over multipart is used. It sends a heartbeat to the client every heartbeatInterval. +// startWorkerWithHeartbeat is similar to startWorker but sends heartbeats to the client when +// the writer supports heartbeats. It sends a heartbeat to the client every heartbeatInterval.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these settings in your CodeRabbit configuration.
📒 Files selected for processing (3)
v2/pkg/engine/resolve/event_loop_test.go(2 hunks)v2/pkg/engine/resolve/resolve.go(6 hunks)v2/pkg/engine/resolve/resolve_test.go(2 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- v2/pkg/engine/resolve/event_loop_test.go
- v2/pkg/engine/resolve/resolve_test.go
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Build and test (go 1.23 / windows-latest)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
🔇 Additional comments (2)
v2/pkg/engine/resolve/resolve.go (2)
505-511: LGTM: early-cancellation checks are in the right place and prevent unnecessary writesThe early returns on resolver and subscription contexts are correct and avoid unnecessary heartbeats on shutdown/client disconnect.
530-533: Confirm metric semantics: are heartbeats “updates”?SubscriptionUpdateSent increments on heartbeats too. If this metric is used to count actual data updates, this will inflate it. Either:
- use a dedicated HeartbeatSent metric, or
- pass a parameter/flag to distinguish heartbeat vs. data.
Please confirm desired behavior.
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (2)
v2/pkg/engine/resolve/resolve.go (2)
142-144: Breaking API: keep deprecated alias for ResolverOptions to avoid downstream breaksRenaming MultipartSubHeartbeatInterval to SubHeartbeatInterval is a breaking change for external callers. Keep the old exported field as deprecated and document the migration.
Apply this diff to restore backward compatibility:
- // SubHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) - SubHeartbeatInterval time.Duration + // Deprecated: use SubHeartbeatInterval. Kept for backward compatibility with older integrations. + MultipartSubHeartbeatInterval time.Duration + // SubHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) + SubHeartbeatInterval time.Duration
157-159: Honor deprecated MultipartSubHeartbeatInterval when defaultingIf users still set the deprecated field, use it before falling back to DefaultHeartbeatInterval.
Apply this diff:
- if options.SubHeartbeatInterval <= 0 { - options.SubHeartbeatInterval = DefaultHeartbeatInterval - } + if options.SubHeartbeatInterval <= 0 { + if options.MultipartSubHeartbeatInterval > 0 { + options.SubHeartbeatInterval = options.MultipartSubHeartbeatInterval + } else { + options.SubHeartbeatInterval = DefaultHeartbeatInterval + } + }
🧹 Nitpick comments (3)
v2/pkg/engine/resolve/resolve.go (3)
309-311: Heartbeat worker structure looks good; consider a one-shot timer to avoid tick backlogsThe design is clear and contained. Optional: replace the ticker with a time.Timer that you reset after each write/heartbeat to avoid accumulating ticks during long writes and to reduce resets per loop.
517-521: Differentiate common disconnects; optionally log non-disconnect errors onceCurrently any Heartbeat() error triggers unsubscribe, which is acceptable for SSE. Optional: treat context.Canceled and io.EOF as expected disconnects and only log other errors once in debug mode for easier diagnosis.
Apply this diff:
- if err := sub.writer.Heartbeat(); err != nil { - // If heartbeat fails (e.g. client disconnected), remove the subscription. - _ = r.AsyncUnsubscribeSubscription(sub.id) - return - } + if err := sub.writer.Heartbeat(); err != nil { + // Treat expected disconnects as terminal. + if err == context.Canceled || err == io.EOF { + _ = r.AsyncUnsubscribeSubscription(sub.id) + return + } + // For other errors, also unsubscribe but surface in debug logs. + if r.options.Debug { + fmt.Printf("resolver:heartbeat:subscription:error:%d:%v\n", sub.id.SubscriptionID, err) + } + _ = r.AsyncUnsubscribeSubscription(sub.id) + return + }
523-526: Nit: clarify debug text to reflect action“done” is ambiguous; “sent” makes the intent clearer.
Apply this diff:
- if r.options.Debug { - fmt.Printf("resolver:heartbeat:subscription:done:%d\n", sub.id.SubscriptionID) - } + if r.options.Debug { + fmt.Printf("resolver:heartbeat:subscription:sent:%d\n", sub.id.SubscriptionID) + }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (1)
v2/pkg/engine/resolve/resolve.go(8 hunks)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
🔇 Additional comments (5)
v2/pkg/engine/resolve/resolve.go (5)
68-72: LGTM: dedicated heartbeat interval on the resolverThe private field and placement are sensible; per-resolver configuration is appropriate.
201-203: LGTM: wire options into resolver stateUsing options.SubHeartbeatInterval to initialize the resolver’s heartbeatInterval is correct.
329-339: LGTM: reset heartbeat after each write to avoid immediate heartbeatsResetting the ticker after work keeps heartbeats strictly idle-time. Good choice.
500-516: LGTM: early-cancel checks before sending heartbeatsShort-circuiting on resolver/client cancellation prevents unnecessary writes.
527-529: Please review how heartbeats are emitted and whether they invokeSubscriptionUpdateSent. The reporter callback is wired to every send—including heartbeats—so heartbeat “pings” will increment your subscription‐update metric. If that is undesirable, you’ll need to filter out heartbeat sends (for example by checking the message payload or wrapping the report call) rather than unconditionally deleting the reporter invocation here.
endigma
force-pushed
the
jesse/eng-7547-heartbeat-on-sse
branch
from
5a1ebf7 to
0827865
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (3)
v2/pkg/engine/resolve/resolve_test.go (1)
93-99: API rename reflected in tests; keep a deprecated alias to avoid breaking external callersRenaming MultipartSubHeartbeatInterval to SubHeartbeatInterval is fine here, but it’s a breaking change for integrators using ResolverOptions. Recommend keeping a deprecated alias on ResolverOptions and honoring it during defaulting.
Apply in v2/pkg/engine/resolve/resolve.go (ResolverOptions):
+ // Deprecated: use SubHeartbeatInterval. Kept for backward compatibility with older integrations. + MultipartSubHeartbeatInterval time.Duration // SubHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) SubHeartbeatInterval time.DurationAnd defaulting:
- if options.SubHeartbeatInterval <= 0 { - options.SubHeartbeatInterval = DefaultHeartbeatInterval - } + if options.SubHeartbeatInterval <= 0 { + if options.MultipartSubHeartbeatInterval > 0 { + options.SubHeartbeatInterval = options.MultipartSubHeartbeatInterval + } else { + options.SubHeartbeatInterval = DefaultHeartbeatInterval + } + }v2/pkg/engine/resolve/resolve.go (2)
142-144: Preserve backward compatibility for renamed optionKeeping only SubHeartbeatInterval breaks existing users. Retain the old exported field as deprecated and map it during initialization.
Apply:
type ResolverOptions struct { @@ - // SubHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) - SubHeartbeatInterval time.Duration + // Deprecated: use SubHeartbeatInterval. Kept for backward compatibility with older integrations. + MultipartSubHeartbeatInterval time.Duration + // SubHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) + SubHeartbeatInterval time.Duration
157-159: Honor the deprecated field during defaulting to prevent silent behavior changesIf users still set MultipartSubHeartbeatInterval, prefer that value before falling back to DefaultHeartbeatInterval.
- if options.SubHeartbeatInterval <= 0 { - options.SubHeartbeatInterval = DefaultHeartbeatInterval - } + if options.SubHeartbeatInterval <= 0 { + if options.MultipartSubHeartbeatInterval > 0 { + options.SubHeartbeatInterval = options.MultipartSubHeartbeatInterval + } else { + options.SubHeartbeatInterval = DefaultHeartbeatInterval + } + }
🧹 Nitpick comments (3)
execution/graphql/result_writer.go (1)
38-40: No-op Heartbeat aligns with the new interface; consider documenting intent and returning a pointer from constructorsThis satisfies resolve.SubscriptionResponseWriter without changing behavior. If EngineResultWriter is ever used on transports that expect an actual heartbeat frame, document explicitly that this writer does nothing on Heartbeat.
Optional: both constructors return a value while all methods have pointer receivers. Returning a pointer eliminates ambiguity and avoids accidental copies.
Outside the selected lines, consider:
// Prefer returning a pointer to satisfy interfaces without taking the address later. func NewEngineResultWriter() *EngineResultWriter { return &EngineResultWriter{buf: &bytes.Buffer{}} } func NewEngineResultWriterFromBuffer(buf *bytes.Buffer) *EngineResultWriter { return &EngineResultWriter{buf: buf} }v2/pkg/engine/resolve/resolve_test.go (1)
4780-4785: Add an assertion to actually enforce “no heartbeats sent”Heartbeat() appends the literal "heartbeat" into messages. The test “should successfully get result from upstream” states that we shouldn’t see heartbeats, but doesn’t assert it. Add an explicit negative assertion to catch regressions.
Outside the selected lines, in the test block around Lines 5126–5135 after reading messages:
assert.NotContains(t, messages, "heartbeat")v2/pkg/engine/resolve/resolve.go (1)
499-530: Consider excluding heartbeats from “update sent” metricsCurrently, reporter.SubscriptionUpdateSent() increments for heartbeats as well as real updates. If this metric is intended to track client-visible payload updates, counting heartbeats will inflate it.
Do you intend to count heartbeats as “updates”? If not, remove the call in handleHeartbeat or introduce a dedicated Reporter hook (e.g., HeartbeatSent()).
Optionally, narrow error handling to treat common closure signals as disconnects (context.Canceled, io.EOF, io.ErrClosedPipe, net.ErrClosed) before unsubscribing, though unsubscribing on any error is acceptable.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (5)
execution/graphql/result_writer.go(1 hunks)v2/pkg/engine/resolve/event_loop_test.go(2 hunks)v2/pkg/engine/resolve/resolve.go(8 hunks)v2/pkg/engine/resolve/resolve_test.go(2 hunks)v2/pkg/engine/resolve/response.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- v2/pkg/engine/resolve/response.go
- v2/pkg/engine/resolve/event_loop_test.go
🧰 Additional context used
🧬 Code Graph Analysis (1)
v2/pkg/engine/resolve/resolve_test.go (1)
v2/pkg/engine/resolve/resolve.go (1)
AsyncErrorWriter(42-44)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
🔇 Additional comments (3)
v2/pkg/engine/resolve/resolve.go (3)
68-70: Good: dedicated heartbeat interval on the resolverClear separation of concerns; no issues.
201-203: LGTM: resolver uses the configured heartbeat intervalInitialization correctly wires the chosen interval.
309-341: Heartbeat worker flow is sound
- Dedicated ticker per subscription, stopped via defer.
- Early returns on resolver/client cancellation.
- Ticker reset after each write to avoid spurious heartbeats during activity.
No issues.
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (2)
v2/pkg/engine/resolve/resolve.go (2)
142-145: Breaking API: keep deprecated alias for external callers and document itRenaming the exported field breaks consumers initializing ResolverOptions. Keep a deprecated alias and prefer the new field.
Apply this diff to preserve backward compatibility:
- // SubscriptionHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) - SubscriptionHeartbeatInterval time.Duration + // Deprecated: use SubscriptionHeartbeatInterval. Kept for backward compatibility with older integrations. + MultipartSubHeartbeatInterval time.Duration + // SubscriptionHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) + SubscriptionHeartbeatInterval time.Duration
157-159: Honor deprecated field during defaulting to avoid silent behavior changesIf legacy integrations still set MultipartSubHeartbeatInterval, prefer it before falling back to DefaultHeartbeatInterval.
- if options.SubscriptionHeartbeatInterval <= 0 { - options.SubscriptionHeartbeatInterval = DefaultHeartbeatInterval - } + if options.SubscriptionHeartbeatInterval <= 0 { + if options.MultipartSubHeartbeatInterval > 0 { + options.SubscriptionHeartbeatInterval = options.MultipartSubHeartbeatInterval + } else { + options.SubscriptionHeartbeatInterval = DefaultHeartbeatInterval + } + }Would you like me to add a unit test that asserts the deprecated field is honored when the new field is unset?
🧹 Nitpick comments (4)
v2/pkg/engine/resolve/resolve.go (4)
68-72: Nit: clarify this is an internal, derived intervalMinor doc clarity: make it explicit that this is the internal interval, derived from options during construction.
- // Subscription heartbeat interval + // Internal subscription heartbeat interval (set from options during construction) heartbeatInterval time.Duration
309-311: Nit: align comment with actual behavior (ticker resets after writes)The worker resets the ticker after each data write to avoid immediate heartbeats. Capture that in the comment.
-// It sends a heartbeat to the client every heartbeatInterval. Heartbeats are handled by the SubscriptionResponseWriter interface. +// Sends a heartbeat every heartbeatInterval and resets the interval after each data write +// to avoid immediate heartbeats following updates. Heartbeats are handled by SubscriptionResponseWriter.
523-525: Nit: use ':sent' to better reflect the event“done” is a little ambiguous in logs; “sent” reads clearer.
- fmt.Printf("resolver:heartbeat:subscription:done:%d\n", sub.id.SubscriptionID) + fmt.Printf("resolver:heartbeat:subscription:sent:%d\n", sub.id.SubscriptionID)
527-529: Don’t treat heartbeats as “updates” in SubscriptionUpdateSentI confirmed that in
v2/pkg/engine/resolve/resolve.go, thehandleHeartbeatmethod (around line 548) callsif r.reporter != nil { r.reporter.SubscriptionUpdateSent() }which means every heartbeat is counted as an “update” in our metrics, inflating dashboard numbers.
To keep update metrics meaningful, I recommend one of the following optional refactors:
• Option A (preferred): stop counting heartbeats as updates. Remove the
SubscriptionUpdateSent()call fromhandleHeartbeat.func (r *Resolver) handleHeartbeat(sub *sub) { … if r.options.Debug { fmt.Printf("resolver:heartbeat:subscription:done:%d\n", sub.id.SubscriptionID) } - if r.reporter != nil { - r.reporter.SubscriptionUpdateSent() - } }• Option B: if you still want visibility into heartbeats but keep update-metrics clean, introduce a new hook via a type-asserted interface instead of changing
Reporter:// Define in the same package type HeartbeatReporter interface { HeartbeatSent() } // In handleHeartbeat: if r.reporter != nil { if hr, ok := r.reporter.(HeartbeatReporter); ok { hr.HeartbeatSent() } }This avoids breaking the existing
Reporterinterface and lets implementers opt into heartbeat metrics.Impacted locations:
v2/pkg/engine/resolve/resolve.go→handleHeartbeat- Tests in
v2/pkg/engine/resolve/event_loop_test.gomay need adjustments if they assert on heartbeat metrics.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (3)
v2/pkg/engine/resolve/event_loop_test.go(2 hunks)v2/pkg/engine/resolve/resolve.go(8 hunks)v2/pkg/engine/resolve/resolve_test.go(3 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- v2/pkg/engine/resolve/resolve_test.go
- v2/pkg/engine/resolve/event_loop_test.go
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
🔇 Additional comments (4)
v2/pkg/engine/resolve/resolve.go (4)
201-202: LGTM: resolver carries the configured heartbeat intervalUsing the normalized options.SubscriptionHeartbeatInterval here is correct.
329-329: LGTM: heartbeats executed on the writer goroutineCalling handleHeartbeat from the worker keeps writer access single-threaded.
499-516: LGTM: early-cancel checks avoid unnecessary workReturning early on resolver/client context cancellation before touching the writer is correct.
517-521: Heartbeat write errors: unsubscribe is acceptable; verify writer semantics across implementationsUnsubscribing on any
Heartbeat()error is a reasonable default given typical EOF or context-canceled cases. Please double-check that everySubscriptionResponseWriterimplementation you’ve registered returns a non-nil error when its underlying transport is closed—otherwise this loop branch could spin indefinitely:
- v2/pkg/engine/resolve/response.go (interface definition)
- v2/pkg/engine/resolve/event_loop_test.go →
FakeSubscriptionWriter.Heartbeat()- v2/pkg/engine/resolve/resolve_test.go →
SubscriptionRecorder.Heartbeat()- execution/graphql/result_writer.go →
EngineResultWriter.Heartbeat()If any of these stubs or real writers swallow transport-closed errors (for example by always returning
nil), please update them so that a closed connection surfaces as an error here.
endigma
force-pushed
the
jesse/eng-7547-heartbeat-on-sse
branch
from
8c167f3 to
fadc6bc
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
v2/pkg/engine/resolve/event_loop_test.go (1)
54-60: Remove redundant string conversion in Heartbeat().The cast string("heartbeat") is unnecessary; the literal is already a string. Minor cleanup.
Apply this diff:
func (f *FakeSubscriptionWriter) Heartbeat() error { f.mu.Lock() defer f.mu.Unlock() - f.writtenMessages = append(f.writtenMessages, string("heartbeat")) + f.writtenMessages = append(f.writtenMessages, "heartbeat") return nil }Optional: consider defining a shared const (e.g., heartbeatMsg = "heartbeat") across test writers to keep the sentinel consistent.
v2/pkg/engine/resolve/resolve_test.go (2)
89-89: Prefer const for the test heartbeat interval.This value is immutable and used as configuration; define it as a const to avoid accidental mutation.
Apply this diff:
-var subscriptionHeartbeatInterval = 100 * time.Millisecond +const subscriptionHeartbeatInterval = 100 * time.Millisecond
4780-4785: Heartbeat() test writer implementation is fine; consider two small hardening tweaks.
- Optional guard: avoid appending after Complete() or Close() to reduce flakiness in racy tests.
- Optional assertion: in tests where heartbeats must not appear (e.g., “should successfully get result from upstream”), add an explicit NotContains check for "heartbeat" to clearly encode intent.
Example guard (optional):
func (s *SubscriptionRecorder) Heartbeat() error { s.mux.Lock() defer s.mux.Unlock() + if s.complete.Load() || s.closed.Load() { + return nil + } s.messages = append(s.messages, "heartbeat") return nil }Example assertion addition in the “should successfully get result from upstream” test (outside the changed lines):
messages := recorder.Messages() assert.NotContains(t, messages, "heartbeat")
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (5)
execution/graphql/result_writer.go(1 hunks)v2/pkg/engine/resolve/event_loop_test.go(2 hunks)v2/pkg/engine/resolve/resolve.go(8 hunks)v2/pkg/engine/resolve/resolve_test.go(3 hunks)v2/pkg/engine/resolve/response.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (3)
- v2/pkg/engine/resolve/response.go
- execution/graphql/result_writer.go
- v2/pkg/engine/resolve/resolve.go
🧰 Additional context used
🧬 Code graph analysis (1)
v2/pkg/engine/resolve/event_loop_test.go (1)
v2/pkg/engine/resolve/resolve.go (1)
DefaultHeartbeatInterval(21-21)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Build and test (go 1.23 / windows-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
🔇 Additional comments (2)
v2/pkg/engine/resolve/event_loop_test.go (1)
134-134: Field rename to SubscriptionHeartbeatInterval looks correct.The test wiring matches the ResolverOptions rename and uses DefaultHeartbeatInterval as expected.
v2/pkg/engine/resolve/resolve_test.go (1)
98-98: API rename applied consistently.Using SubscriptionHeartbeatInterval in New(...) aligns with the public API change.
endigma
force-pushed
the
jesse/eng-7547-heartbeat-on-sse
branch
from
fadc6bc to
3733b12
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (2)
v2/pkg/engine/resolve/resolve.go (2)
142-144: Breaking API: field rename will break external callers — add deprecated aliasRenaming MultipartSubHeartbeatInterval → SubscriptionHeartbeatInterval breaks users instantiating ResolverOptions. Keep a deprecated alias to preserve source compatibility.
Apply in this struct:
// SubscriptionHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) + // Deprecated: use SubscriptionHeartbeatInterval. Honored if the new field is zero. + MultipartSubHeartbeatInterval time.Duration SubscriptionHeartbeatInterval time.Duration
157-159: Honor deprecated field during defaultingIf users still set MultipartSubHeartbeatInterval, prefer it when the new field is unset to avoid silent behavior changes.
- if options.SubscriptionHeartbeatInterval <= 0 { - options.SubscriptionHeartbeatInterval = DefaultHeartbeatInterval - } + if options.SubscriptionHeartbeatInterval <= 0 { + if options.MultipartSubHeartbeatInterval > 0 { + options.SubscriptionHeartbeatInterval = options.MultipartSubHeartbeatInterval + } else { + options.SubscriptionHeartbeatInterval = DefaultHeartbeatInterval + } + }
🧹 Nitpick comments (4)
v2/pkg/engine/resolve/resolve.go (4)
309-311: Clarify heartbeat worker docsNit: mention that heartbeats are enabled per subscription via ExecutionOptions.SendHeartbeat and implemented by the writer; actual action may be a no‑op.
-// startWorkerWithHeartbeat is similar to startWorker but sends heartbeats to the client when enabled. -// It sends a heartbeat to the client every heartbeatInterval. Heartbeats are handled by the SubscriptionResponseWriter interface. +// startWorkerWithHeartbeat is like startWorker but, when ExecutionOptions.SendHeartbeat is true, +// invokes writer.Heartbeat() every heartbeatInterval (writer may no-op). +// Heartbeat behavior is transport-specific and provided by SubscriptionResponseWriter.
517-521: On heartbeat error, immediate unsubscribe — acceptable; consider distinguishing metrics/loggingTreating any write error as disconnect is pragmatic. Optionally classify common closure errors (context.Canceled, io.EOF, net.ErrClosed) for cleaner logs, but not required.
523-525: Debug string: “done” → “sent” (optional)Minor: “sent” or “ok” better matches action semantics.
- fmt.Printf("resolver:heartbeat:subscription:done:%d\n", sub.id.SubscriptionID) + fmt.Printf("resolver:heartbeat:subscription:sent:%d\n", sub.id.SubscriptionID)
527-529: Don’t count heartbeats as “updates”SubscriptionUpdateSent likely tracks payload updates. Counting heartbeats may skew metrics. Consider a separate reporter hook, e.g., SubscriptionHeartbeatSent().
- if r.reporter != nil { - r.reporter.SubscriptionUpdateSent() - } + if r.reporter != nil { + // Optional: add a dedicated metric instead of reusing update counter. + // r.reporter.SubscriptionHeartbeatSent() + }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (5)
execution/graphql/result_writer.go(1 hunks)v2/pkg/engine/resolve/event_loop_test.go(2 hunks)v2/pkg/engine/resolve/resolve.go(8 hunks)v2/pkg/engine/resolve/resolve_test.go(3 hunks)v2/pkg/engine/resolve/response.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (3)
- execution/graphql/result_writer.go
- v2/pkg/engine/resolve/resolve_test.go
- v2/pkg/engine/resolve/event_loop_test.go
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
- GitHub Check: Build and test (go 1.23 / windows-latest)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
🔇 Additional comments (6)
v2/pkg/engine/resolve/response.go (1)
66-72: DocumentSubscriptionResponseWriter.Heartbeatsemantics & ensure implementations complyAdding the new
Heartbeat() errormethod to the exportedSubscriptionResponseWriterinterface is a breaking change for any third-party or internal types that implement it. At minimum, please:
Document the precise behavior
– It must send a transport-level keepalive frame only (no GraphQL message),
– Fallback implementations should returnnil(safe no-op),
– On underlying disconnects, returncontext.Canceled,io.EOF, or a similar error.Provide a migration aid
Consider defining a minimal “capability” interface—e.g.:type Heartbeater interface { // Heartbeat sends a transport-level keepalive; safe no-op if unsupported. Heartbeat() error }and wrapping or upgrading existing writers with a decorator that implements
Heartbeat()for them.Apply the following doc comment in
v2/pkg/engine/resolve/response.go:type SubscriptionResponseWriter interface { ResponseWriter Flush() error Complete() + // Heartbeat sends a transport-level keepalive without emitting a GraphQL payload. + // Implementations that don't support heartbeats should return nil (no-op). + // Disconnects should surface as context.Canceled, io.EOF, or similar. Heartbeat() error Close(kind SubscriptionCloseKind) }• Verify implementations
To catch any broken types at compile time, simply rebuild or test the code:# this will fail if any type now lacks Heartbeat() go test ./v2/... ./pkg/...Additionally, you can add a compile-time assertion in your own packages:
// ensure MyWriter still implements the interface var _ SubscriptionResponseWriter = (*MyWriter)(nil)This ensures that all existing and future implementers are updated and avoids runtime surprises.
v2/pkg/engine/resolve/resolve.go (5)
68-71: Internal heartbeatInterval field — LGTMScoped, private cache of options.SubscriptionHeartbeatInterval used by workers. No issues.
201-203: Propagating interval into resolver — LGTMResolver caches options.SubscriptionHeartbeatInterval into heartbeatInterval.
329-329: Heartbeat called from the writer goroutine — LGTMPreserves the single-writer invariant; good.
499-516: Early-cancel checks before heartbeat — LGTMAvoids futile writes when resolver or client context is done.
338-339: Confirmed Go ≥1.15 — no changes neededThe
time.Ticker.Resetcall is safe on active tickers as of Go 1.15, and your module requires Go 1.23.0 per go.mod, so no draining or additional handling is necessary.
endigma
force-pushed
the
jesse/eng-7547-heartbeat-on-sse
branch
from
3733b12 to
62e42e6
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
v2/pkg/engine/resolve/resolve.go (1)
499-520: Gate heartbeats via optional interface to avoid hard dependency on new methodIf you keep SubscriptionResponseWriter unchanged (as suggested), gate heartbeat calls via a type assertion to SubscriptionHeartbeater. Writers without heartbeat support will safely no-op.
-func (r *Resolver) handleHeartbeat(sub *sub) { +func (r *Resolver) handleHeartbeat(sub *sub) { @@ - if err := sub.writer.Heartbeat(); err != nil { + hb, ok := sub.writer.(SubscriptionHeartbeater) + if !ok { + // Writer doesn't support heartbeats; nothing to do. + return + } + if err := hb.Heartbeat(); err != nil { // If heartbeat fails (e.g. client disconnected), remove the subscription. _ = r.AsyncUnsubscribeSubscription(sub.id) return }
♻️ Duplicate comments (2)
v2/pkg/engine/resolve/resolve.go (2)
157-159: Honor the deprecated option during defaultingIf callers still set MultipartSubHeartbeatInterval, prefer it before falling back to DefaultHeartbeatInterval.
- if options.SubscriptionHeartbeatInterval <= 0 { - options.SubscriptionHeartbeatInterval = DefaultHeartbeatInterval - } + if options.SubscriptionHeartbeatInterval <= 0 { + if options.MultipartSubHeartbeatInterval > 0 { + options.SubscriptionHeartbeatInterval = options.MultipartSubHeartbeatInterval + } else { + options.SubscriptionHeartbeatInterval = DefaultHeartbeatInterval + } + }
142-145: Preserve backward compatibility: reintroduce deprecated field and docsRenaming MultipartSubHeartbeatInterval to SubscriptionHeartbeatInterval breaks external callers constructing ResolverOptions. Keep the deprecated field and document it.
type ResolverOptions struct { @@ - // SubscriptionHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) - SubscriptionHeartbeatInterval time.Duration + // Deprecated: use SubscriptionHeartbeatInterval. Kept for backward compatibility. + MultipartSubHeartbeatInterval time.Duration + // SubscriptionHeartbeatInterval defines the interval in which a heartbeat is sent to all subscriptions (whether or not this does anything is determined by the subscription response writer) + SubscriptionHeartbeatInterval time.Duration
🧹 Nitpick comments (3)
execution/graphql/result_writer.go (1)
38-40: Make Heartbeat useful by delegating to Flush()A no-op heartbeat means upstream keepalives won’t propagate. Delegate to Flush() so transports using the flushCallback can emit a keepalive frame.
-func (e *EngineResultWriter) Heartbeat() error { - return nil -} +func (e *EngineResultWriter) Heartbeat() error { + return e.Flush() +}v2/pkg/engine/resolve/resolve.go (2)
313-321: Avoid per-sub ticker when writer can’t heartbeatSmall optimization: if the writer doesn’t implement heartbeats, fall back to the non-heartbeat worker to avoid creating/resetting a ticker unnecessarily.
func (s *sub) startWorkerWithHeartbeat() { + if _, ok := s.writer.(SubscriptionHeartbeater); !ok { + s.startWorkerWithoutHeartbeat() + return + } heartbeatTicker := time.NewTicker(s.resolver.heartbeatInterval) defer heartbeatTicker.Stop()
527-529: Confirm metric semantics: counting heartbeats as “updates”SubscriptionUpdateSent() is incremented on heartbeats. If dashboards interpret this as data updates, consider adding a separate Reporter hook (e.g., SubscriptionHeartbeatSent) to avoid skew.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (5)
execution/graphql/result_writer.go(1 hunks)v2/pkg/engine/resolve/event_loop_test.go(2 hunks)v2/pkg/engine/resolve/resolve.go(8 hunks)v2/pkg/engine/resolve/resolve_test.go(3 hunks)v2/pkg/engine/resolve/response.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- v2/pkg/engine/resolve/event_loop_test.go
- v2/pkg/engine/resolve/resolve_test.go
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
- GitHub Check: Build and test (go 1.23 / ubuntu-latest)
- GitHub Check: Build and test (go 1.23 / windows-latest)
🔇 Additional comments (1)
v2/pkg/engine/resolve/response.go (1)
66-72: Separate Heartbeat into an optional interface to avoid breaking changesI ran a search for internal types implementing Heartbeat() or the full SubscriptionResponseWriter interface and found no matches, but that doesn’t guarantee there are no external implementers. To prevent breaking any third-party writers, keep the core interface unchanged and introduce a new, narrow interface for heartbeats:
• File needing update:
- v2/pkg/engine/resolve/response.go
• Proposed diff:
type SubscriptionResponseWriter interface { ResponseWriter Flush() error Complete() - Heartbeat() error Close(kind SubscriptionCloseKind) } +// SubscriptionHeartbeater is optionally implemented by writers that can emit transport-level keepalive frames. +type SubscriptionHeartbeater interface { + Heartbeat() error +}• Usage note:
In the resolver, emit a heartbeat only when the writer satisfies SubscriptionHeartbeater, e.g.:if hb, ok := writer.(SubscriptionHeartbeater); ok { hb.Heartbeat() }Please verify whether any external packages implement the old interface (and would thus break) before merging.
🤖 I have created a release *beep* *boop* --- ## [2.0.0-rc.223](v2.0.0-rc.222...v2.0.0-rc.223) (2025-08-28) ### Features * improved subscription heartbeats ([#1269](#1269)) ([4423d60](4423d60)) --- This PR was generated with [Release Please](https://github.com/googleapis/release-please). See [documentation](https://github.com/googleapis/release-please#release-please). <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * Documentation * Added changelog entry for 2.0.0-rc.223 (2025-08-28), highlighting improved subscription heartbeats with references to the related issue and commit. * Existing changelog entries remain intact; changes are additive. * Chores * Updated the public release manifest to reference v2 version 2.0.0-rc.223 (execution version unchanged). * No code, UI, or behavior changes included in this PR. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
2 participants
Summary by CodeRabbit
New Features
Bug Fixes
Performance
Refactor
Tests
Checklist