Implement optional StreamMaxLifetime for server streams - #1142
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (4)
🚧 Files skipped from review as they are similar to previous changes (4)
WalkthroughAdds configurable ChangesStream Max Lifetime
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Config
participant GRPCService
participant StreamHandler
participant StreamContext
Client->>Config: Set STREAM_MAX_LIFETIME
Config->>GRPCService: Load StreamMaxLifetime
GRPCService->>StreamHandler: Construct handler with lifetime
Client->>StreamHandler: Start streaming RPC
StreamHandler->>StreamContext: Create bounded context
StreamContext-->>StreamHandler: Return context
StreamHandler-->>Client: End stream on timeout or cancellation
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
internal/interface/grpc/handlers/arkservice.go (1)
271-296: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBounded context wiring looks correct.
ctxis properly derived fromstream.Context()so client-side disconnects still propagate through the child context, andctx.Done()correctly replacesstream.Context().Done()in the select loop. Returningnilon expiry lets the client observeio.EOFand reconnect as intended.One gap: unlike
GetSubscriptioninindexer.go(which now hasTestGetSubscriptionMaxLifetime), there's no equivalent test here assertingGetEventStream/GetTransactionsStreamactually terminate oncemaxStreamLifetimeelapses. Consider adding analogous coverage for these two handlers.Also applies to: 467-491
🤖 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 `@internal/interface/grpc/handlers/arkservice.go` around lines 271 - 296, The bounded stream lifetime wiring in GetEventStream and GetTransactionsStream looks correct, but there is no test coverage proving they actually stop when maxStreamLifetime expires. Add analogous lifetime-expiry tests, similar to TestGetSubscriptionMaxLifetime in indexer.go, that exercise these handler paths and assert the stream terminates after the configured lifetime while still using the existing streamContext/ctx.Done flow.
🤖 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.
Nitpick comments:
In `@internal/interface/grpc/handlers/arkservice.go`:
- Around line 271-296: The bounded stream lifetime wiring in GetEventStream and
GetTransactionsStream looks correct, but there is no test coverage proving they
actually stop when maxStreamLifetime expires. Add analogous lifetime-expiry
tests, similar to TestGetSubscriptionMaxLifetime in indexer.go, that exercise
these handler paths and assert the stream terminates after the configured
lifetime while still using the existing streamContext/ctx.Done flow.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: d94c3fd3-7c58-4988-9a05-2db39187a3ee
📒 Files selected for processing (8)
README.mdcmd/arkd/main.gointernal/config/config.gointernal/interface/grpc/config.gointernal/interface/grpc/handlers/arkservice.gointernal/interface/grpc/handlers/indexer.gointernal/interface/grpc/handlers/indexer_test.gointernal/interface/grpc/service.go
Missed in a merge of `master`.
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Summary
Bounding server-streaming RPCs is a reasonable defense against phantom sessions that a proxy prevents us from noticing. The plumbing itself (config → grpc.Config → handler → streamContext helper) is clean, and the helper's contract (<=0 disables, positive bounds) is locked down by tests. Two behaviour concerns and a few cleanup items below; requesting changes on #1/#2.
1. GetSubscription new-flow: reap destroys the subscription → reconnect gets SUBSCRIPTION_NOT_FOUND
internal/interface/grpc/handlers/indexer.go:452-469 — for a new-flow stream (client sends empty subscription_id) reconnectWindow is forced to 0. When maxStreamLifetime fires, the deferred h.scriptSubsHandler.release(subscriptionId, att, 0) deletes the listener immediately (broker.go:258-264: reconnectWindow > 0 && hasFilters is required to schedule expiry). A client that reconnects with the ID it received in SubscriptionStartedEvent gets SUBSCRIPTION_NOT_FOUND and must re-issue SubscribeForScripts and re-apply its filters.
That contradicts the comment added at indexer.go:497-500 ("live clients reconnect transparently"). It only holds for the legacy flow where reconnectWindow = subscriptionTimeoutDuration (60s). New-flow clients that don't have SDK-level stale-ID recovery lose events silently every 30 min.
Either:
- Bind
reconnectWindow = h.subscriptionTimeoutDurationwhen release is triggered by the server-side lifetime (distinguish server-driven expiry from client-cancel), or - Update the docstring and README to state that new-flow clients must handle SUBSCRIPTION_NOT_FOUND on reconnect and re-apply their filters, and add a test that covers a new-flow stream getting reaped, reconnecting with the ID, and having to resubscribe.
2. Cross-repo consumer breakage: go-sdk listenForArkTxs returns on io.EOF
go-sdk/wallet.go:939-942:
if errors.Is(event.Err, io.EOF) {
closeFunc()
return
}There is no reconnect. When arkd 1142 ships with the default STREAM_MAX_LIFETIME=1800, any wallet built on the current go-sdk stops receiving ArkTx/CommitmentTx/SweepTx notifications 30 min after startup — settlements complete but the wallet-side VTXO store is never updated. This is a silent regression, no error is logged.
Options: (a) ship go-sdk with a reconnect loop first and default arkd to 0 for one release, then flip; (b) ship the two coordinated; (c) at minimum, call this out in the README with a migration note. The ts-sdk contractWatcher (contracts/contractWatcher.ts:684-688) already handles it; only the go-sdk tx listener needs work.
3. Non-blocking pre-select still watches stream.Context().Done()
indexer.go:519-527 (the "priority" non-blocking select) still uses stream.Context().Done() while the blocking select (now line 534) uses ctx.Done(). Not a correctness bug — a max-lifetime expiry is still caught on the next iteration of the blocking select — but it's inconsistent and means the priority optimization does not apply to the new exit path. Just switch it to ctx.Done() for consistency.
4. Thundering herd at expiry
Every stream opened within the same window (e.g. after an arkd restart) will be reaped at the same wall-clock instant. Add ±10% jitter to maxStreamLifetime in streamContext so reconnect load is smoothed.
5. Test coverage
TestGetSubscriptionMaxLifetime and TestStreamContext are good, but the equivalent path through GetEventStream / GetTransactionsStream in arkservice.go has no direct test. Since they use the same helper the risk is small, but a mock-stream test per handler is cheap and prevents future refactors from silently regressing.
Nits
- README.md:104 stray period at end of
ARKD_INDEXER_EXPOSURE.(pre-existing). - On reap the server just returns
nil(client sees OK/EOF). Considerstream.SetTrailer(md{"x-ark-stream-reaped": "true"})so clients can distinguish "reaped, reconnect now" from "server draining" and log accordingly.
|
Changes were requested 2+ days ago. @s373nZ need any help addressing the feedback? |
|
Changes were requested 8+ days ago. @s373nZ need any help addressing the feedback? |
|
Changes were requested 5 days ago. @s373nZ need any help addressing the feedback? |
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Arkana review — sha 28de8b5
Looks ready to merge.
StreamMaxLifetime is a straightforward and well-targeted defence against abandoned gRPC streams that survive because a proxy masks the client disconnect. The implementation is consistent across all three streaming handlers (GetEventStream, GetTransactionsStream, GetSubscription). The streamContext helper is a clean single-responsibility extraction.
Behaviour is exactly right: the stream returns nil on expiry so the client sees io.EOF and can reconnect, rather than an error that might be treated as permanent.
Default of 1800s (30min) is a sensible starting point; operators can tune or disable with ARKD_STREAM_MAX_LIFETIME=0.
The test TestGetSubscriptionMaxLifetime verifies the reaping at 150ms without relying on the production default, and TestStreamContext locks the disabled/bounded branches. Both pass. No concerns.
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Review — StreamMaxLifetime for server streams
Verdict: looks ready to merge.
Adds a bounded lifetime to all three server-streaming RPCs (GetEventStream, GetTransactionsStream, GetSubscription) so abandoned streams — clients that vanished behind a proxy that masks the TCP RST — are reaped after at most ARKD_STREAM_MAX_LIFETIME seconds (default 30 min).
Design observations:
streamContextis a clean, testable helper that abstracts the zero-disables-bound semantics. ✅- Returning
nilon timeout means the client seesio.EOFrather than a gRPC error, which is the correct signal for a reconnect. ✅ - The 30-minute default is reasonable for clients that reconnect on EOF; operators running behind L7 proxies with shorter keepalive timeouts can lower it.
- The bound uses
context.WithTimeoutfrom the stream's own context, so normal client disconnects still cancel promptly — the bound only fires when the parent context outlives the deadline. ✅
Tests: TestGetSubscriptionMaxLifetime and TestStreamContext are present and cover the key paths (zero disables, positive imposes deadline, handler exits gracefully). ✅
Minor note: the CHANGES_REQUESTED decision is from a prior reviewer — worth checking if their concerns have been addressed before merging.
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Changes were requested 48 days ago. @s373nZ need any help addressing the feedback on StreamMaxLifetime?
|
Changes were requested 52+ days ago. @s373nZ — need any help addressing the feedback? |
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Arkana review — #1142 (sha 28de8b5)
Implement optional StreamMaxLifetime for server streams
What's here
ARKD_STREAM_MAX_LIFETIME(default 1800s = 30min) bounds how longGetEventStreamandGetTransactionsStreammay stay openstreamContext()helper: returnsWithCancelwhen lifetime ≤ 0 (disabled),WithTimeoutotherwise- On expiry, handler returns
nil→ client receivesio.EOF→ reconnects transparently - Same logic applied to indexer
GetSubscription
Looks good
- The
maxLifetime <= 0guard correctly disables the bound - Returning
nil(not an error) on timeout is correct for gRPC server-streaming —io.EOFis the clean stream end that clients expect to reconnect on streamContextdefers properly —cancel()is always called
One question
When maxStreamLifetime > 0 and the stream times out, does the broker entry get cleaned up via the defer h.eventsListenerHandler.removeListener(listener.id) that's already in place? If the listener is removed on context expiry, a reconnecting client will register a fresh listener — confirm the broker's dedup check won't block a re-registration with the same ID.
Verdict: Clean, minimal change. Looks ready.
|
Changes were requested 7+ weeks ago. @s373nZ — need any help addressing the feedback? |
|
Changes were requested 56+ days ago. @s373nZ need any help addressing the feedback on StreamMaxLifetime? |
|
Changes were requested 2+ days ago. @s373nZ — StreamMaxLifetime: need any help addressing the feedback? |
|
Changes have been requested on this PR for 2+ months. @s373nZ need any help addressing the |
3 participants
Implements an optional
StreamMaxLifetimesetting which bounds the duration a stream should live for. When set, clients are expected to re-connect when the connection is reaped.Summary by CodeRabbit
STREAM_MAX_LIFETIMEconfiguration to cap maximum lifetime of server-streaming RPCs (default: 1800s).STREAM_MAX_LIFETIME=0to disable the cap.