feat(executor): let a plugin's Filter actually reach arkd - #15
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (3)
WalkthroughThe solver now opens a separate arkd gRPC connection and configures arkd subscriptions. ChangesArkade subscription integration
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Service
participant Source
participant ArkdIndexer
participant Solver
Service->>ArkdIndexer: Create arkd gRPC connection
Service->>Source: Attach subscription client
Solver->>Source: Subscribe with CEL filter
Source->>ArkdIndexer: Request filtered subscription
ArkdIndexer-->>Source: Stream transaction events
Source-->>Solver: Deliver decoded PSBT packets
Source->>ArkdIndexer: Reopen stream after failure
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 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 |
executor.Run has passed p.Filter() to source.Subscribe since it was written, and arkdsource has thrown it away just as long — the doc comment says so outright. Filter() is dead code today and every plugin reads the whole tx stream. It cannot be honoured on the RPC arkdsource uses. GetTransactionsStream is on the Ark service and takes no filter; CEL expressions live on the indexer's GetSubscription, in SubscriptionFilter.expressions. Nor can client-lib carry one — its NewSubscription takes a script list and nothing else, and it opens the stream with a subscription id already assigned, which is exactly the case where arkd documents the filter as ignored. So the filtered path talks to the generated indexer stub, with an empty subscription id so arkd creates the subscription and applies the expression itself. api-spec was already an indirect dependency at a build that has SubscriptionFilter, so it only moves to direct. An empty filter still means the full stream, and deliberately does not become an empty subscription: one with no expressions and no scripts matches nothing, so the swap plugin and anything else that sets no filter would have gone quiet rather than unfiltered. When a plugin does set one and no subscription client is configured, Subscribe says so at Warn and falls back — the plugin still filters in Match, but silence there is what let this go unnoticed. solverd opens its own arkd connection for it. The go-sdk keeps its conn private and grpc.NewClient is lazy, so an unused one costs nothing.
bdf8eb4 to
9df949d
Compare
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Arkana review — #15
Verdict: approve with minor nits. No protocol-critical paths touched.
This change lives entirely in the tx sourcing layer — it wires up the server-side CEL filter that was already plumbed through executor.Plugin.Filter() but silently dropped in arkdsource. VTXO handling, signing, forfeit paths, round lifecycle, and unilateral exit are unaffected.
Correctness
GetSubscriptionRequest semantics confirmed. The proto comment at api-spec/protobuf/gen/ark/v1/indexer.pb.go matches the PR description: subscription_id empty → arkd creates the subscription and applies filter; subscription_id non-empty → filter is ignored. Sending an empty id to have arkd assign one and apply the expression is the right call.
Empty-filter fast path is correct. Routing an empty filter to subscribeAll rather than a zero-expression subscription is necessary: an empty SubscriptionFilter is a no-op on creation (confirmed in the UpdateSubscriptionRequest proto docs), which for the subscription path means the subscription would match nothing. Sending the swap plugin — and any future plugin with Filter() == "" — down that path would silently starve it. The behaviour correctly diverges here.
Context cancellation in subscribeFiltered. The context is passed to GetSubscription, so cancellation propagates into the gRPC stream and Recv() returns a context error. The ctx.Err() != nil guard catches it before the Warn path. Clean.
stop() not needed in subscribeFiltered. subscribeAll defers stop() (the stopper returned by GetTransactionsStream); subscribeFiltered uses a gRPC stream whose lifetime is tied to the context, so no separate stopper is needed. Correct.
Minor issues
service.go:171–173 — nil guard is always true.
if s.arkdConn != nil {
src = src.WithSubscriptions(arkv1.NewIndexerServiceClient(s.arkdConn))
}New() either sets arkdConn to a non-nil *grpc.ClientConn or returns an error, so this condition is always satisfied on a live Service. The guard reads as if there's a meaningful nil case, which there isn't. grpc.NewClient being lazy is correctly called out in the comment, but the laziness applies to dialling, not to object creation. No functional impact; just a readability nit.
No reconnect on stream error (subscribeFiltered). When Recv() returns a non-EOF, non-context error, the goroutine logs at Warn and exits, closing out. The executor sees the closed channel and returns ErrAllStreamsClosed. This is pre-existing behaviour from subscribeAll and documented implicitly in the executor contract; noting it here only so future callers know: a transient arkd blip currently terminates the plugin loop until the process restarts.
Cross-repo impact
covclaimd (current head). Both preimage.plugin.Filter() and RevealPlugin.Filter() return "", so both take the subscribeAll path regardless. No behaviour change and no breakage; the nil-subscription-client Warn path will not fire for existing covclaimd deployments. The new WithSubscriptions option is additive.
covclaimd#5 (pending). covclaimd/cmd/covclaimd/main.go constructs arkdsource.New(arkClient, logger) without calling WithSubscriptions. If covclaimd#5 adds a non-empty Filter() to one of its plugins without also passing a subscription client, Subscribe will Warn and fall back to the full stream — which is safe but noisy. The PR description acknowledges this; it's a covclaimd#5 concern, not a blocker here.
solver/pkg/swap — behaviour change for existing deployments. swap.plugin.Filter() returns a non-empty CEL expression. With this PR, solverd dials the indexer subscription path for the swap plugin instead of the full tx stream. This is the intended improvement, but operators running solverd should be aware: the first RPC after startup will open a subscription rather than streaming all txs.
Tests
Six tests cover the documented branch points well: empty filter → tx stream even with a subscription client configured; non-empty filter → request reaches arkd with empty subscription id and the correct expression; heartbeats, subscription-started events, and hex sweep txs are skipped without stalling; fallback Warn fires when no subscription client is configured; error from GetSubscription propagates; channel closes on EOF. The stubSubscriptions.stream assignment happens synchronously inside GetSubscription before the goroutine starts, so there's no data race in the test helpers.
The missing -race run is the honest gap called out in the PR description; nothing in the new code suggests a race, but CI is the right place to confirm it.
go.mod
Moving api-spec from indirect to direct is correct — the package now imports it directly in service.go and arkdsource.go. The module hash doesn't change; go.sum is unaffected.
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Arkana follow-up review — #15 (9df949d)
Status: rebase-only delta since bdf8eb4. Zero prior findings fixed; both minor nits remain open. No new defects introduced.
The 33-commit gap between bdf8eb4 and 9df949d7 is almost entirely base-branch evolution (market domain, banco CLI, wallet service, SQLite migrations, gRPC handler changes, etc. — 32 commits, 101 files that the PR does not touch). The PR itself is a clean rebase of the same four-file change I already reviewed. PR diff content is functionally identical to the previous head.
Prior-finding disposition
1. service.go nil guard always true — STILL OPEN
The guard in Run():
if s.arkdConn != nil {
src = src.WithSubscriptions(arkv1.NewIndexerServiceClient(s.arkdConn))
}is still there unchanged. grpc.NewClient either returns a non-nil *ClientConn or causes New() to return an error; a live Service always has arkdConn set. The guard reads as if there is a meaningful nil path — there isn't. Readability nit, no functional impact.
The parallel guard in Close() is consistent with how s.db and s.emulatorConn are guarded there; I read it as defensive style and do not object to it in Close().
2. No reconnect on subscription stream error — STILL OPEN (pre-existing, no regression)
A non-EOF, non-context error from subscribeFiltered's Recv() logs at Warn and exits, closing out. The executor sees a closed channel and terminates the plugin loop until process restart. Unchanged from the previous pass. Calling it out here only to confirm the rebase did not accidentally improve or worsen it.
Incremental observations (new base context)
go.mod — still correct after rebase. api-spec is promoted from indirect to direct at the same pinned pseudo-version v0.0.0-20260615144548-11cf2ba852c5. The go 1.26.5 directive was already on the base branch (60c8061 ci: bump golang version to 1.26.5), so the PR carries no version change there.
dialTarget reuse — clean. The base branch added dialTarget for the emulator connection. The PR reuses it for the arkd connection. The logic (strip http://, detect https://, default ports 80/443) is correct for both callers.
Base-branch struct growth — no interaction issues. The Service struct now carries marketRepo, tradeRepo, and indexer (all added in the base). The PR appends arkdConn cleanly alongside the other connection handles; no field shadows or zero-value surprises.
Rebase conflict warning is a false alarm. CodeRabbit's merge-conflict checkbox appears stale; the PR diff applies cleanly against the current base.
Scope containment confirmed. The 101 base-branch files are completely orthogonal to the four PR files (go.mod, service.go, arkdsource.go, arkdsource_test.go). No new cross-file entanglement was introduced.
No new blockers. The two open nits from the previous pass are the only outstanding feedback. Both are low-severity and do not block merge.
subscribeFiltered treated every Recv error as terminal: the goroutine returned, the channel closed, and the executor read that as end-of-stream. Once the last plugin's stream went, Run returned ErrAllStreamsClosed and the solver kept running while claiming nothing. A dropped connection, an arkd restart or a GOAWAY was enough to do it. That is a regression against the path it replaces. The unfiltered stream goes through client-lib, which wraps it in utils.StartReconnectingStream and heals itself; the filtered path hand-rolls its loop because that helper lives in client-lib/internal/utils and cannot be imported from here. So the loop re-subscribes instead of returning, carrying the same expression and an empty subscription id so arkd applies the filter again. Backoff is 1s doubling to 10s with 20% jitter, matching client-lib's own GrpcReconnectConfig — there is no reason for two streams against the same arkd to flap on different schedules. Only ctx ends the subscription: the first subscribe is synchronous, so an expression arkd rejects still surfaces to the caller, and after that a closed channel is not something the executor can act on. io.EOF reconnects with everything else, because it is how a server-streaming RPC reports that the server closed the stream and arkd has no "subscription finished" event — client-lib's own ShouldReconnect classifies it the same way. Reconnecting restores the stream, it does not replay what the stream missed. arkd creates a fresh subscription on each reconnect and GetSubscriptionRequest carries only a subscription id and a filter, with nothing to resume from, so txs that arrive while the subscription is down are not delivered afterwards. Both the break and the recovery are logged. A subscription that flaps silently is the same outage as one that dies silently.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 `@pkg/executor/arkdsource/arkdsource.go`:
- Around line 192-195: Update Subscribe around openSubscription to perform the
initial Recv and surface any error before creating out or starting the reconnect
goroutine; return the error with a nil channel, especially for
codes.InvalidArgument, rather than entering the retry loop. Add a test covering
an initial Recv that returns codes.InvalidArgument and asserting Subscribe
returns an error and nil channel.
🪄 Autofix
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: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: b02edf42-2c40-4a69-8cfc-d595cd0e6002
📒 Files selected for processing (4)
go.modinternal/core/application/service.gopkg/executor/arkdsource/arkdsource.gopkg/executor/arkdsource/arkdsource_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- internal/core/application/service.go
| stream, err := s.openSubscription(ctx, filter) | ||
| if err != nil { | ||
| return nil, err | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify that the pinned generated client does not call Recv while creating
# the GetSubscription client stream.
set -euo pipefail
version="$(awk '$1 == "github.com/arkade-os/arkd/api-spec" { print $2 }' go.mod)"
commit="${version##*-}"
tree="$(curl -fsSL "https://api.github.com/repos/arkade-os/arkd/git/trees/${commit}?recursive=1")"
path="$(printf '%s' "$tree" | jq -r \
'.tree[] | select(.path | test("indexer.*grpc.*pb\\.go$")) | .path' | head -n1)"
test -n "$path"
curl -fsSL "https://raw.githubusercontent.com/arkade-os/arkd/${commit}/${path}" |
grep -n -A35 -B5 'GetSubscription'Repository: arkade-os/solver
Length of output: 205
🏁 Script executed:
set -euo pipefail
printf '%s\n' '--- candidate files ---'
git ls-files | rg '(^|/)(arkdsource\.go|.*grpc.*\.pb\.go|go\.mod)$'
printf '%s\n' '--- target implementation ---'
cat -n pkg/executor/arkdsource/arkdsource.go | sed -n '150,240p'
printf '%s\n' '--- subscription symbols and call sites ---'
rg -n -C 4 'openSubscription|subscribeFiltered|readStream|GetSubscription|Subscribe\(' pkg
printf '%s\n' '--- module dependency ---'
rg -n 'github.com/arkade-os/arkd/api-spec|replace ' go.mod go.sumRepository: arkade-os/solver
Length of output: 30514
🏁 Script executed:
set -euo pipefail
printf '%s\n' '--- generated GetSubscription implementations in the repository ---'
rg -n -C 12 'func .*GetSubscription|GetSubscription\(' api-spec pkg --glob '*.go'
printf '%s\n' '--- stream response definitions and event handling ---'
rg -n -C 10 'type GetSubscriptionResponse|SubscriptionStarted|SubscriptionStartedEvent|GetSubscriptionRequest' api-spec pkg --glob '*.go'
printf '%s\n' '--- readStream and reconnect implementation ---'
cat -n pkg/executor/arkdsource/arkdsource.go | sed -n '227,305p'
printf '%s\n' '--- filtered subscription tests and stub Recv behavior ---'
cat -n pkg/executor/arkdsource/arkdsource_test.go | sed -n '40,230p'
cat -n pkg/executor/arkdsource/arkdsource_test.go | sed -n '395,480p'
printf '%s\n' '--- module/cache locations, without executing repository code ---'
go env GOPATH GOMODCACHE 2>/dev/null || true
find "${GOMODCACHE:-/nonexistent}" -path '*arkd*api-spec*' -type f -name '*.go' -print 2>/dev/null | head -20Repository: arkade-os/solver
Length of output: 39816
🏁 Script executed:
set -euo pipefail
moddir="$(go env GOMODCACHE)/github.com/arkade-os/arkd/api-spec@v0.0.0-20260615144548-11cf2ba852c5"
printf '%s\n' '--- generated GetSubscription client method ---'
rg -n -A45 -B8 'func \(c .*GetSubscription' "$moddir/protobuf/gen/ark/v1/indexer_grpc.pb.go"
printf '%s\n' '--- GetSubscription response/request protobuf definitions ---'
rg -n -A80 -B10 'type GetSubscription(Request|Response)|type SubscriptionStartedEvent' \
"$moddir/protobuf/gen/ark/v1/indexer.pb.go"
printf '%s\n' '--- source proto declaration, if present ---'
rg -n -A35 -B10 'GetSubscription|SubscriptionStarted' \
"$moddir" --glob '*.proto' --glob '*.go' | head -220Repository: arkade-os/solver
Length of output: 47640
🌐 Web query:
site:github.com/arkade-os/arkd GetSubscription SubscriptionStartedEvent CEL InvalidArgument
💡 Result:
In the context of the arkd indexer service, receiving an InvalidArgument error in relation to GetSubscription and CEL (Common Expression Language) typically indicates a failure in the validation or compilation of filter expressions provided by the client [1][2][3]. Specifically, the InvalidArgument status is returned by the GetSubscription RPC under several conditions involving CEL-based transaction filters [1][3]: 1. CEL Compilation Failures: If a provided CEL filter expression fails to compile (e.g., due to syntax errors or invalid expression structures), the server returns an InvalidArgument error [1][3]. A dedicated INVALID_TX_FILTER error (mapped to gRPC InvalidArgument) exists to identify these cases, often including the offending expression in the error metadata [3]. 2. Unsupported Expression Shapes: The system only supports specific CEL expression shapes (e.g., using predicates like has(tx.extension) or exact equality checks on hex-encoded bytes). Using unsupported operations—such as OR, NOT, inequality (!=), size, or.contains—will cause the filter to be rejected with an InvalidArgument error [4]. 3. Resource Limits: If the number of compiled transaction filters exceeds the maximum allowed limit (MaxTxFiltersPerListener), the server returns an InvalidArgument error [3]. This is specifically surfaced via a TX_FILTERS_LIMIT_EXCEEDED error [3]. 4. Script/Filter Validation: Any script parse errors or failures during the validation of the initial subscription filter will also trigger an InvalidArgument response [1]. When a client initiates a GetSubscription request—particularly in the single-connection flow where a SubscriptionStartedEvent is sent upon successful creation—the server performs these validations on the provided filter before the subscription is established [1][5][3]. If any validation fails, the RPC will return the error immediately [1].
Citations:
- 1: https://github.com/arkade-os/arkd/blob/master/api-spec/protobuf/ark/v1/indexer.proto
- 2: arkade-os/arkd@c4f1632
- 3: Add structured TX_FILTERS_LIMIT_EXCEEDED and INVALID_TX_FILTER errors arkd#1141
- 4: Add filters to indexer.GetVirtualTxs arkd#1090
- 5: single-connection GetSubscription flow arkd#951
🌐 Web query:
site:github.com/arkade-os/arkd "func (.*GetSubscription" OR "GetSubscription(ctx" indexer
💡 Result:
The code pattern you referenced, typically seen as func (i *Indexer) GetSubscription(ctx context.Context,...) or similar, is associated with the Arkade OS (arkd) project's indexer service [1][2]. In the arkd codebase, GetSubscription is a server-side streaming gRPC endpoint [1] used to receive real-time notifications about transactions related to specific vtxo scripts [1]. Key details about this method include: - Functionality: It allows clients to create or update subscriptions [1]. If a subscription ID is provided, it attaches to an existing subscription; if empty, the server automatically generates a new one [1][3]. - Evolution: The implementation has evolved from a multi-step process (SubscribeForScripts followed by GetSubscription) to a streamlined single-connection flow where subscription parameters can be provided inline [3]. - Lifecycle Management: Recent updates have focused on ensuring that GetSubscription streams are properly terminated upon reconnection to prevent "leaked" streams that compete for events [2]. These updates introduced an attachment/displacement mechanism through an internal broker, ensuring that reconnected streams correctly handle event delivery [2]. - Integration: The method is often used alongside filters (e.g., SubscriptionFilter) to scope the stream to specific transaction types or scripts [1][4]. This pattern is standard in the arkd repository's Go-based gRPC handlers, commonly found in files such as internal/interface/grpc/handlers/indexer.go [2].
Citations:
- 1: https://github.com/arkade-os/arkd/blob/master/api-spec/protobuf/ark/v1/indexer.proto
- 2: fix(indexer): terminate abandoned GetSubscription streams on reconnect arkd#1143
- 3: single-connection GetSubscription flow arkd#951
- 4: arkade-os/arkd@c4f1632
🏁 Script executed:
set -euo pipefail
url='https://raw.githubusercontent.com/arkade-os/arkd/master/internal/interface/grpc/handlers/indexer.go'
body="$(curl -fsSL "$url")"
printf '%s\n' '--- GetSubscription handler and nearby validation ---'
printf '%s\n' "$body" | rg -n -A80 -B20 'GetSubscription|SubscriptionStarted|InvalidArgument|filter'
printf '%s\n' '--- handler source size and method declarations ---'
printf '%s\n' "$body" | wc -l
printf '%s\n' "$body" | rg -n '^func 'Repository: arkade-os/solver
Length of output: 33627
Surface the initial GetSubscription error before returning.
The generated client does not call Recv. arkd validates the initial CEL filter before sending SubscriptionStartedEvent, so an invalid filter reaches this code as codes.InvalidArgument from the first Recv. The reconnect loop then retries the invalid filter indefinitely.
Consume the initial response before creating out, and return its error before starting the reconnect goroutine. Add a test that makes the first Recv return codes.InvalidArgument and asserts that Subscribe returns an error with a nil channel.
🤖 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 `@pkg/executor/arkdsource/arkdsource.go` around lines 192 - 195, Update
Subscribe around openSubscription to perform the initial Recv and surface any
error before creating out or starting the reconnect goroutine; return the error
with a nil channel, especially for codes.InvalidArgument, rather than entering
the retry loop. Add a test covering an initial Recv that returns
codes.InvalidArgument and asserting Subscribe returns an error and nil channel.
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Arkana follow-up review — #15 (8341d7f)
Status: one commit since 9df949d — the reconnect fix. Prior finding #2 is fixed; prior finding #1 remains open. One new finding on the reconnect logic itself.
Prior-finding disposition
1. service.go nil guard always true — STILL OPEN
The if s.arkdConn != nil guard in Run() (internal/core/application/service.go:171) is unchanged. grpc.NewClient either returns a non-nil *ClientConn or causes New() to return an error, so a live Service always has arkdConn set. The guard reads as if there is a meaningful nil path — there isn't. Readability nit, no functional impact. The parallel guard in Close() is fine as defensive style alongside s.db and s.emulatorConn.
2. No reconnect on subscription stream error — FIXED by 8341d7f
subscribeFiltered now calls readStream in a loop and re-establishes the subscription through reopenSubscription instead of returning on the first non-context error. Tested by TestSubscribe_ReconnectsAfterStreamFailure, TestSubscribe_ReconnectsOnServerEOF, TestSubscribe_CancelClosesChannelDuringReconnect, TestSubscribe_ReconnectBacksOff, and TestSubscribe_LogsReconnect.
New finding — non-retryable gRPC error codes enter the reconnect loop
pkg/executor/arkdsource/arkdsource.go — readStream and reopenSubscription
readStream returns any non-context Recv() error unconditionally. subscribeFiltered then calls reopenSubscription, which backs off and retries with the same filter expression until ctx is canceled. No error class is excluded.
This means a permanently-bad condition — an invalid CEL expression (codes.InvalidArgument), a revoked credential (codes.Unauthenticated, codes.PermissionDenied), or an unimplemented RPC (codes.Unimplemented) — enters the retry loop, logs "arkd subscription stream broke, reconnecting" at Warn roughly every 10 s, and keeps out open-but-empty for the lifetime of ctx. The executor never sees a closed source channel, so Run never returns ErrAllStreamsClosed, and the solver process keeps running while permanently claiming nothing.
This is worse than the pre-fix behaviour, where a terminal Recv() error closed the channel and at least let the executor surface the problem. The reconnect loop is the right fix for transient errors; it needs to break on permanent ones.
Concrete failure path:
- Operator misconfigures a CEL expression.
openSubscriptionsucceeds (gRPC streams are established before headers arrive).- First
readStreamcall:Recv()returnscodes.InvalidArgumentfrom arkd; returned as non-nil error. subscribeFilteredgoroutine: logs Warn, callsreopenSubscription.reopenSubscription: backs off 1 s → callsopenSubscription(succeeds) → returns new stream.- Goto 3. Indefinitely.
Suggested fix: classify errors before retrying. The standard partition for gRPC reconnect logic is:
// in readStream, after rerr != nil && ctx.Err() == nil:
if isPermError(rerr) {
return nil, rerr // or a sentinel that subscribeFiltered distinguishes
}
return rerr, nil
func isPermError(err error) bool {
switch status.Code(err) {
case codes.InvalidArgument, codes.Unauthenticated,
codes.PermissionDenied, codes.Unimplemented:
return true
}
return false
}On a permanent error subscribeFiltered should log at Error (not Warn), close out, and return. A test covering codes.InvalidArgument on first Recv() should assert that out closes and does not cycle.
Note: io.EOF correctly triggers reconnect and is tested — no change needed there.
Incremental correctness notes (no action needed)
Reconnect architecture is sound. reopenSubscription checks ctx.Done() before every time.After(delay) sleep, so a cancel that arrives mid-backoff is observed promptly (verified by TestSubscribe_CancelClosesChannelDuringReconnect). The jitter calculation at pkg/executor/arkdsource/arkdsource.go:jittered is bounded: range is [d×0.8, d×1.2), always positive for d > 0. math/rand/v2's global rand.Float64() is auto-seeded — no seed=1 footgun.
Filter expression is carried through every re-subscribe. openSubscription is called with the original filter on every attempt; empty SubscriptionId is preserved. Verified by TestSubscribe_ReconnectsAfterStreamFailure's three-request assertion. Correct.
Test helpers are race-safe. captureHook and stubSubscriptions are mutex-guarded. stubArkClient.calls is incremented in the calling goroutine before launching the background goroutine, and read by tests only after a channel rendezvous, so no data race. The -race CI run remains the right place to confirm this holds end-to-end.
No protocol-critical paths are touched. The nil-guard nit and the non-retryable-error loop are the outstanding items before merge.
The reconnect loop retried every non-context Recv error, which turned a permanent rejection into an endless one. A malformed CEL expression, a credential arkd refuses or an RPC it does not serve would re-subscribe every backoff interval forever, with the consumer holding a channel that never closes and never delivers. The executor never sees end-of-stream, so it never reports anything — quieter than the bug the loop was written to fix. I had reasoned that a bad expression could not reach the loop, because the first subscribe is synchronous. That was wrong about gRPC. Stream creation returns before the server's headers arrive, so GetSubscription hands back a stream with a nil error and arkd's InvalidArgument lands on the first Recv, inside the goroutine, after Subscribe has already returned. Only an unreachable arkd fails synchronously. So the loop classifies before retrying. InvalidArgument, Unauthenticated, PermissionDenied and Unimplemented close the stream and log at Error, leaving the executor to surface the fault. The set is deliberately limited to codes that are unambiguously permanent for this call rather than mirroring client-lib's table: judging a recoverable error permanent brings back the silent outage, so anything not named keeps reconnecting. io.EOF and Unavailable still reconnect, both covered.
New either sets arkdConn to a live conn or returns an error, so the nil check in Run never fires. It read as though there were a Service without a subscription client, which there is not. The guard in Close stays as defensive style on a teardown path.
There was a problem hiding this comment.
🧹 Nitpick comments (1)
pkg/executor/arkdsource/arkdsource_test.go (1)
212-243: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCover every permanent gRPC status code.
This test covers only
codes.InvalidArgument. Add cases forcodes.Unauthenticated,codes.PermissionDenied, andcodes.Unimplemented. The PR contract treats all four codes as terminal. A later change can otherwise retry one of these errors without test failure.🤖 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 `@pkg/executor/arkdsource/arkdsource_test.go` around lines 212 - 243, Expand TestSubscribe_PermanentErrorClosesChannel to cover codes.InvalidArgument, codes.Unauthenticated, codes.PermissionDenied, and codes.Unimplemented using table-driven subtests. For each status, preserve the assertions that the output channel closes, no resubscription occurs, and an Error-level log is emitted.
🤖 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 `@pkg/executor/arkdsource/arkdsource_test.go`:
- Around line 212-243: Expand TestSubscribe_PermanentErrorClosesChannel to cover
codes.InvalidArgument, codes.Unauthenticated, codes.PermissionDenied, and
codes.Unimplemented using table-driven subtests. For each status, preserve the
assertions that the output channel closes, no resubscription occurs, and an
Error-level log is emitted.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 4a2cdbab-ba9e-4d94-a264-42f57962b90a
📒 Files selected for processing (3)
internal/core/application/service.gopkg/executor/arkdsource/arkdsource.gopkg/executor/arkdsource/arkdsource_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
- internal/core/application/service.go
- pkg/executor/arkdsource/arkdsource.go
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Arkana follow-up review — #15 (888c38c)
Status: one commit since 8341d7f — the permanent-error fix + nil-guard cleanup. All three prior findings are now fixed. One nit remains on test coverage. No new defects.
Prior-finding disposition
1. service.go nil guard always true — FIXED by 888c38c
The guard is gone. Run() now reads:
// New either sets arkdConn or fails, so there is no unsubscribed Service to
// guard against here.
src := arkdsource.New(s.arkClient.Client(), s.log).
WithSubscriptions(arkv1.NewIndexerServiceClient(s.arkdConn))The comment confirms the intent. The defensive nil guard in Close() (alongside s.db and s.emulatorConn) is appropriate and untouched.
2. No reconnect on subscription stream error — FIXED in 8341d7f (confirmed, no regression)
Unchanged. The reconnect loop, reopenSubscription, and the five associated tests remain intact.
3. Non-retryable gRPC codes entering the reconnect loop — FIXED by 888c38c
permanentCodes (InvalidArgument, Unauthenticated, PermissionDenied, Unimplemented) and isPermanent() are added. The fix hits both legs of the problem:
subscribeFilteredgoroutine:isPermanent(rerr)on aRecv()error — logs at Error and returns, closingout. The executor can now surface the fault.reopenSubscription:isPermanent(err)on aopenSubscription()error — returns(nil, err)instead of spinning; caller logs at Error and returns.
The distinction between the two error arrival sites (Recv() for CEL rejection, openSubscription() for RPC-level faults like Unimplemented) is correctly modelled and documented.
TestSubscribe_PermanentErrorClosesChannel and TestSubscribe_UnavailableStillReconnects pin the permanent/transient boundary. The boundary is correctly drawn: io.EOF has no gRPC status, so status.Code(io.EOF) returns codes.Unknown, which is not in permanentCodes — reconnected, as intended.
Incremental observations
isPermanent in reopenSubscription is the right call for the right reason.
If openSubscription itself (i.e. the GetSubscription RPC) returns a permanent code — codes.Unimplemented is the realistic case if an arkd deploy drops the RPC — the loop now breaks with an error rather than cycling every backoff interval. Correct.
Test coverage nit — TestSubscribe_PermanentErrorClosesChannel exercises only codes.InvalidArgument.
pkg/executor/arkdsource/arkdsource_test.go:204–243
The other three permanent codes (Unauthenticated, PermissionDenied, Unimplemented) share the same permanentCodes lookup and will behave identically today. But a future change that removes one of those codes from the map would break a real invariant without a failing test. A table-driven expansion costs ten lines and pins the boundary against such drift. CodeRabbit raised this in the same pass; I'm echoing it for completeness. Not a blocker.
No protocol-critical paths are touched. No new defects introduced. The nil-guard nit from pass 1 is gone, the reconnect-termination bug from pass 4 is fixed. This is ready to merge once the table-driven test nit is addressed (or consciously deferred).
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Arkana follow-up review — #15 (1ffd0b4)
Status: one new commit (fixes) by @louisinger. 1/2 prior findings fixed; 1 still open. Two new findings, one of which needs attention before merge.
Prior-finding disposition
1. service.go nil guard always true — Fixed in 888c38c8 by @Kukks. The guard is gone; the comment explaining why was cleaned up in this commit. Closed.
2. No reconnect on subscription stream error — Still open (pre-existing, no regression)
This finding as stated in my first pass was imprecise — subscribeFiltered always had a reconnect loop; what lacked reconnect was subscribeAll. subscribeAll still exits the goroutine on a stream error with no retry, which is the expected client-lib behaviour. No change, no regression. Carrying it forward only for completeness.
Incremental review — 1ffd0b4b only
service.go — grpc.MaxCallRecvMsgSize(20 << 20)
pkg/executor/arkdsource/service.go:82–85 (new lines). Correct fix: subscription events carry full PSBTs that overflow the 4 MB gRPC default. Matching client-lib's own cap is the right ceiling.
The cleaned-up comment (removal of the now-redundant nil-guard note) is fine.
arkdsource.go — staleness watchdog
pkg/executor/arkdsource/arkdsource.go:289–296 (readStream):
The design is correct: cancel targets the child context (sctx), not ctx, so when the watchdog fires, Recv returns a context error, ctx.Err() is still nil, and readStream returns it as transient — triggering the reconnect loop rather than a clean exit. Defer ordering (Stop before cancel, LIFO) is also fine.
Finding A — streamStaleTimeout is a package-level constant; the watchdog cannot be exercised in tests.
reconnectDelay and reconnectMaxDelay are exported struct fields precisely so tests can shorten them (src.reconnectDelay = 5 * time.Millisecond). streamStaleTimeout is a hard-coded constant. There is no test that verifies a stale stream is cancelled and the subscription reopened, because 3 minutes is not a feasible test interval and there is no knob to shorten it.
The fix pattern is already in the codebase: add a streamStaleTimeout time.Duration field to Source, initialise it to 3 * time.Minute in New, and use s.streamStaleTimeout in readStream. That unblocks a test like the others in this file.
arkdsource.go — initial transient failure is now non-fatal
pkg/executor/arkdsource/arkdsource.go:232–245:
The rationale (executor drops a plugin on a Subscribe error, so a booting arkd would kill solverd) is sound and documented. cancel is nil on the transient-failure path — openSubscription calls cancel() before returning nil, nil, err — and the goroutine overwrites it from reopenSubscription before reaching readStream, so no nil-cancel is ever passed. Correct.
Finding B — the new code path has no test.
TestSubscribe_FilteredSubscribeErrorIsReturned was correctly updated to codes.Unauthenticated (permanent), matching the new semantics. But there is no test for the transient case: first call returns a transient error → goroutine retries in the background → stream eventually opens → packets flow. The stubSubscriptions helper already has reopenErr for a similar pattern (TestSubscribe_ReconnectBacksOff); the same approach works here with a firstCallErr / err that is cleared after the first call. Without this test the new code path — which is the behaviour change most likely to surprise an operator — has no coverage.
arkdsource_test.go
TestSubscribe_FilteredSubscribeErrorIsReturned: error type change to status.Error(codes.Unauthenticated, "boom") is the correct update. A plain errors.New is now transient under the new logic and would put the test into a retry loop rather than returning an error. The test now verifies the permanent-error surface path correctly.
Summary
| # | Finding | Severity | Required before merge? |
|---|---|---|---|
| A | streamStaleTimeout constant → watchdog untestable |
Medium | Yes — untested safety mechanism in the hot path |
| B | No test for initial transient failure + recovery | Medium | Yes — new behaviour change, zero coverage |
Both can be addressed with small additions to the existing test file and one field promotion. No protocol-critical paths are affected.
arkana-ai-bot
left a comment
arkana-ai-bot
left a comment
There was a problem hiding this comment.
Arkana follow-up review — #15 (1ffd0b4)
Status: one new commit ("fixes") by @louisinger. 1/2 prior findings fixed; 1 still open. Two new findings, one of which needs attention before merge.
Prior-finding disposition
1. service.go nil guard always true — Fixed in 888c38c8 by @Kukks. The guard is gone; the comment explaining why was cleaned up in this commit. Closed.
2. No reconnect on subscription stream error — Still open (pre-existing, no regression)
This finding as stated in my first pass was imprecise — subscribeFiltered always had a reconnect loop; what lacked reconnect was subscribeAll. subscribeAll still exits the goroutine on a stream error with no retry, which is the expected client-lib behaviour. No change, no regression. Carrying it forward only for completeness.
Incremental review — 1ffd0b4b only
service.go — grpc.MaxCallRecvMsgSize(20 << 20)
internal/core/application/service.go:82–85 (new lines). Correct fix: subscription events carry full PSBTs that overflow the 4 MB gRPC default. Matching client-lib's own cap is the right ceiling.
The cleaned-up comment (removal of the now-redundant nil-guard note) is fine.
arkdsource.go — staleness watchdog
pkg/executor/arkdsource/arkdsource.go in readStream:
defer cancel()
stale := time.AfterFunc(streamStaleTimeout, cancel)
defer stale.Stop()
for {
resp, rerr := stream.Recv()
…
stale.Reset(streamStaleTimeout)The design is correct: cancel targets the child context (sctx), not ctx, so when the watchdog fires, Recv returns a context error, ctx.Err() is still nil, and readStream returns it as transient — triggering the reconnect loop rather than a clean exit. Defer ordering (Stop before cancel, LIFO) is also fine.
Finding A — streamStaleTimeout is a package-level constant; the watchdog cannot be exercised in tests.
reconnectDelay and reconnectMaxDelay are exported struct fields precisely so tests can shorten them (src.reconnectDelay = 5 * time.Millisecond). streamStaleTimeout is a hard-coded constant. There is no test that verifies a stale stream is cancelled and the subscription reopened, because 3 minutes is not a feasible test interval and there is no knob to shorten it.
The fix pattern is already in the codebase: add a staleTimeout time.Duration field to Source, initialise it to 3 * time.Minute in New, and reference s.staleTimeout in readStream. That unblocks a test that stubs a stream delivering no messages, sets a short timeout, and asserts the source reopens the subscription.
arkdsource.go — initial transient failure is now non-fatal
pkg/executor/arkdsource/arkdsource.go in subscribeFiltered:
stream, cancel, err := s.openSubscription(ctx, filter)
if err != nil {
if isPermanent(err) {
return nil, err
}
s.log.WithError(err).Warn("arkd unreachable on first subscribe, retrying in background")
stream = nil
}The rationale (executor drops a plugin on a Subscribe error, so a booting arkd would kill solverd) is sound and documented. cancel is nil on the transient-failure path — openSubscription calls cancel() before returning nil, nil, err — and the goroutine overwrites it from reopenSubscription before reaching readStream, so no nil-cancel is ever passed. Correct.
Finding B — the new code path has no test.
TestSubscribe_FilteredSubscribeErrorIsReturned was correctly updated to codes.Unauthenticated (permanent), matching the new semantics. But there is no test for the transient case: first call returns a transient error → goroutine retries in the background → stream eventually opens → packets flow. The stubSubscriptions helper already has a reopenErr field for a similar pattern (TestSubscribe_ReconnectBacksOff); a symmetric firstCallErr that is cleared after the first call would cover this path. Without it, the behaviour change most likely to surprise an operator has no test.
arkdsource_test.go
TestSubscribe_FilteredSubscribeErrorIsReturned: error type change to status.Error(codes.Unauthenticated, "boom") is the correct update. A plain errors.New is now transient under the new logic and would put the test into a retry loop rather than returning an error. The test now verifies the permanent-error surface path correctly.
Summary
| # | Finding | Severity | Required before merge? |
|---|---|---|---|
| A | streamStaleTimeout constant → watchdog untestable |
Medium | Yes — untested safety mechanism in the connection hot path |
| B | No test for initial transient failure + recovery | Medium | Yes — new behaviour change with zero coverage |
Both are addressed by small additions to the existing test file and one field promotion in Source. No protocol-critical paths are affected.
3 participants
executor.Runhas passedp.Filter()tosource.Subscribesince it was written, andarkdsourcehas been throwing it away just as long — the doc comment says so outright. SoFilter()is dead code and every plugin reads the full tx stream. That's fine for the swap plugin, which sets no filter, but covclaimd is about to want one, and right now it would look wired without being wired.It turns out it can't be honoured on the RPC
arkdsourceuses.GetTransactionsStreamis on the Ark service and takes no filter at all; the CEL expressions live on the indexer'sGetSubscription, inSubscriptionFilter.expressions. And client-lib can't carry one either —NewSubscriptiontakes a script list and nothing else, and it always opens the stream with a subscription id already assigned, which is precisely the case where the proto says the filter is ignored. So the filtered path here talks to the generatedarkv1.IndexerServiceClientdirectly, with an empty subscription id so arkd creates the subscription and applies the expression itself.That's the part I'd most like your read on. Dropping to the generated stub means
arkdsourceneeds a gRPC conn it can't get fromclient.Client, soSourcegrows an optionalWithSubscriptions(shaped afterExecutor.WithLogger) and solverd dials arkd a second time through the existingdialTarget— the go-sdk keeps its own conn private.grpc.NewClientis lazy, so the extra conn costs nothing when no plugin sets a filter. The alternative is adding expressions to client-lib'sIndexerupstream in arkd and waiting for it, which is cleaner but not something I can do from here. Happy to switch if you'd rather have it that way.Dropping to the stub also meant hand-rolling the receive loop, and my first version of it got that wrong in a way worth flagging: every
Recverror was terminal. A dropped connection, an arkd restart or a GOAWAY closed the channel for good,executor.Runread that as end-of-stream, and once the last plugin's stream went it returnedErrAllStreamsClosed— solver still running, just not claiming anything. The unfiltered path doesn't have this problem, because client-lib wraps it inutils.StartReconnectingStream; that's the helper I can't reuse, since it lives inclient-lib/internal/utils. So the loop now re-subscribes instead of returning, carrying the same expression and an empty subscription id so arkd applies the filter again, backing off 1s doubling to 10s with 20% jitter to match client-lib's ownGrpcReconnectConfig— no reason for two streams against the same arkd to flap on different schedules.io.EOFreconnects along with the rest: it's how a server-streaming RPC reports that the server closed the stream, arkd has no "subscription finished" event, and client-lib's ownShouldReconnectclassifies it the same way.Retrying everything turned out to be its own bug, and since I'd argued against classifying on a mistaken premise it's worth spelling out. I'd assumed a malformed CEL expression couldn't reach the loop, because the first subscribe is synchronous. gRPC doesn't work that way: stream creation returns before the server's headers arrive, so
GetSubscriptionhands back a stream with a nil error and arkd'sInvalidArgumentlands on the firstRecv, inside the goroutine, afterSubscribehas already returned nil. I checked that against a real gRPC server rather than reasoning about it — a handler that fails immediately withInvalidArgumentgivesGetSubscription() error = <nil>and surfaces the code onRecv. An unreachable server does fail synchronously, which is what misled me. Left alone, a bad expression would have re-subscribed every 10s forever while the consumer held a channel that never closed and never delivered: noErrAllStreamsClosed, no signal at all, which is worse than what I set out to fix.So the loop classifies before retrying.
InvalidArgument,Unauthenticated,PermissionDeniedandUnimplementedclose the stream and log at Error, leaving the executor to surface them. I kept that list to codes unambiguously permanent for this call rather than mirroring client-lib'sShouldReconnecttable, because judging a recoverable error permanent puts the silent outage straight back — anything not named keeps reconnecting.Unavailable, which is what a restarting arkd looks like, has its own test to pin it on the retry side.Worth being plain about what that buys and what it doesn't: reconnecting restores the stream, it doesn't replay what the stream missed. arkd creates a fresh subscription on each reconnect and
GetSubscriptionRequestcarries only a subscription id and a filter, with nothing to resume from, so txs arriving while the subscription is down aren't delivered afterwards. Both the break and the recovery are logged, since a subscription that flaps silently is the same outage as one that dies silently.Two other behaviours worth calling out. An empty filter still means the full stream and deliberately does not become an empty subscription — one with no expressions and no scripts matches nothing, so the swap plugin would have gone silent rather than unfiltered, which is the kind of change that looks like it works until nothing trades. And when a plugin does set a filter but no subscription client is configured,
Subscribelogs at Warn and falls back to the full stream; the plugin still filters inMatch, so nothing breaks, but the silent version of that is what let this sit unnoticed.No dependency bump:
arkd/api-specwas already pinned as an indirect dependency at a build that hasSubscriptionFilter, so the onlygo.modchange is the line moving to the direct block, andgo.sumis untouched.There's also a small readability fix in
service.go: theif s.arkdConn != nilguard inRunnever fired, sinceNeweither sets the field or returns an error. The parallel guard inClosestays.On evidence: twelve tests in
pkg/executor/arkdsource. Five cover the filter wiring — that an empty filter keeps using the tx stream and never opens a subscription, that a non-empty one arrives at arkd asexpressionswith an empty subscription id, that heartbeats, the subscription-started event and a hex sweep tx are all skipped without stalling the stream, that the fallback warns, and that a failed subscribe is returned to the caller. Seven cover the stream lifecycle: that a packet still arrives after aRecvfailure and again after a second one, with the expression intact on all three subscribe requests; thatio.EOFreconnects; thatUnavailablereconnects; thatInvalidArgumentinstead closes the channel, logs at Error and does not re-subscribe; that canceling ctx closes the channel promptly even with an hour of backoff pending; that backoff actually backs off instead of spinning; and that both the break and the recovery reach the log.I checked those bite rather than assuming. Putting the terminal
returnback fails four — three on "source did not open a subscription", one on "did not retry at all". Removing the permanent-error classification fails the new one on "channel stayed open on a permanent error: the consumer is stranded", which is the stranded-consumer symptom itself.One test changed meaning rather than being added: the old one asserted the channel closes when the stream ends, which is the behaviour this fixes, so it's now the EOF-reconnect test.
go build ./...,go vet ./...andgolangci-lint run ./...(v2.9.0, the version CI pins) are clean. Every non-e2e package passes, unchanged from before this commit;test/e2efails on this box either way for want of the arkd docker stack.I can't run
-racelocally — no cgo toolchain here — somake teston CI is what exercises it, and it matters more now there's a reconnect goroutine. It's green there, with the stubs mutex-guarded for it. I haven't run any of this against a live arkd, though: the subscription path is covered by stubs and by reading arkd's handler (GetSubscriptioncreates the subscription inline when the id is empty,matchesTxevaluates the expressions, and the event carriesTxas the same base64 PSBT the Ark stream uses), but a real end-to-end subscription, and a real reconnect against a restarting arkd, are unverified.The covclaimd side that uses this is arkade-os/covclaimd#5.
Summary by CodeRabbit
New Features
Reliability Improvements
Bug Fixes