feat(mempool): kupo support for resolving inputs in mempool input plugin#622
Conversation
Signed-off-by: cryptodj413 <shinjirohara2@gmail.com>
📝 WalkthroughWalkthroughAdds Kupo-based input resolution to the mempool input plugin, including new Mempool fields (kupoUrl, kupoClient, kupoDisabled, kupoInvalidPatternLogged), getKupoClient and resolveTransactionInputs implementations, CLI/config wiring via WithKupoUrl and a new Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 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 |
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (6)
input/mempool/options.go (1)
79-84: Go naming convention: exported acronyms should be all-caps.Per Go conventions, acronyms in exported identifiers should be fully capitalized —
WithKupoURLrather thanWithKupoUrl. The same applies to thekupo-urlCLI option's wiring inplugin.go. The unexported fieldkupoUrlis fine as-is.Proposed rename
-// WithKupoUrl specifies the Kupo API URL for resolving transaction inputs (e.g. http://localhost:1442). -func WithKupoUrl(kupoUrl string) MempoolOptionFunc { +// WithKupoURL specifies the Kupo API URL for resolving transaction inputs (e.g. http://localhost:1442). +func WithKupoURL(kupoUrl string) MempoolOptionFunc { return func(m *Mempool) { m.kupoUrl = kupoUrl } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@input/mempool/options.go` around lines 79 - 84, Rename the exported option function WithKupoUrl to WithKupoURL and update all call sites and plugin wiring that reference it (including the CLI wiring in plugin.go that currently uses the `kupo-url` flag) so they call WithKupoURL instead; leave the unexported field kupoUrl unchanged, run go vet/gofmt to ensure exports are consistent, and update any tests or imports that referenced WithKupoUrl to the new WithKupoURL identifier.input/mempool/mempool.go (5)
427-462: Kupo queries are not cancellable on shutdown — potential shutdown delay.Each
resolveTransactionInputscall creates contexts fromcontext.Background(). IfStop()is called mid-resolution, the in-flight HTTP calls to Kupo will block for up todefaultKupoTimeout(30s) per input before the poll goroutine can exit. For transactions with many inputs, this could significantly delay graceful shutdown.Consider deriving the context from a parent that is cancelled when
m.doneChanis closed.Sketch: pass a cancellable context
One approach is to store a
context.CancelFuncalongsidedoneChanso that all in-flight work is cancelled onStop():-func (m *Mempool) resolveTransactionInputs(tx ledger.Transaction) ([]ledger.TransactionOutput, error) { +func (m *Mempool) resolveTransactionInputs(ctx context.Context, tx ledger.Transaction) ([]ledger.TransactionOutput, error) { var resolvedInputs []ledger.TransactionOutput k, err := m.getKupoClient() if err != nil { return nil, err } for _, input := range tx.Inputs() { txID := input.Id().String() txIndex := int(input.Index()) pattern := fmt.Sprintf("%d@%s", txIndex, txID) - ctx, cancel := context.WithTimeout(context.Background(), defaultKupoTimeout) + queryCtx, cancel := context.WithTimeout(ctx, defaultKupoTimeout) - matches, err := k.Matches(ctx, kugo.Pattern(pattern)) + matches, err := k.Matches(queryCtx, kugo.Pattern(pattern)) cancel()Then in
pollOnce, derivectxfromm.doneChan(e.g., via a helper or a stored cancel context).🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@input/mempool/mempool.go` around lines 427 - 462, The resolveTransactionInputs code uses context.Background() for each Kupo query which prevents shutdown from cancelling in-flight requests; change it to derive per-request contexts from a cancellable parent context stored on the mempool struct (e.g., m.ctx/m.cancel) that Stop() cancels: add a context.Context and context.CancelFunc fields to the mempool, initialize them when starting/polling, call m.cancel() in Stop(), and replace context.WithTimeout(context.Background(), defaultKupoTimeout) with context.WithTimeout(m.ctx, defaultKupoTimeout) so k.Matches(ctx, ...) is cancelled on shutdown; keep existing timeout/error handling and behavior for m.kupoDisabled and logging intact.
373-376: Cached Kupo client is never invalidated on connection loss.
getKupoClientcaches the client after the first successful health check. If Kupo becomes unreachable later, match queries will fail repeatedly but the health check won't re-run (the client is already cached). This is acceptable becauseresolveTransactionInputshandles errors gracefully, but be aware that the health check only validates initial reachability.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@input/mempool/mempool.go` around lines 373 - 376, getKupoClient currently returns a cached m.kupoClient without re-validating connectivity; update getKupoClient to perform a lightweight health/ping check on the cached client and if that check fails clear m.kupoClient (set to nil) and return an error so a fresh client will be created on the next call. Specifically, in getKupoClient check the existing m.kupoClient via its Health/Ping method (or perform a trivial request), and on any failure assign m.kupoClient = nil and return the error; keep the existing flow of creating and caching a client when no cached client exists. Ensure you reference m.kupoClient and getKupoClient when making the change so future connection losses cause revalidation and recreation.
391-418: Redundant timeout: bothhttp.Client.Timeoutand request context impose the same deadline.Line 392 sets a context timeout of
kupoHealthTimeoutand line 398 setshttp.Client{Timeout: kupoHealthTimeout}. These are redundant — the shorter one wins, but since they're identical, one can be dropped for clarity. Typically the context-based approach is preferred as it's more composable.Simplify by removing the client-level timeout
- httpClient := &http.Client{Timeout: kupoHealthTimeout} + httpClient := &http.Client{}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@input/mempool/mempool.go` around lines 391 - 418, The health check sets a deadline twice (context.WithTimeout using kupoHealthTimeout and http.Client{Timeout: kupoHealthTimeout}); remove the redundant client-level timeout and rely on the request context instead. Specifically, delete or change the instantiation of httpClient that sets Timeout (the &http.Client{Timeout: kupoHealthTimeout} creation used with httpClient.Do(req)) so the client does not impose its own timeout (use the default client or a plain &http.Client{}), keeping the context-based timeout via context.WithTimeout and preserving the existing error handling around http.NewRequestWithContext, httpClient.Do, resp handling, and assignment to m.kupoClient / return k.
436-446: String-based error detection is fragile but pragmatic.Matching on
"Invalid pattern!"and JSON unmarshal error text is brittle — upstream library changes could silently break this detection. Consider documenting this coupling (e.g., a comment noting the kugo library version this was tested against) so future maintainers know to re-verify after kugo upgrades.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@input/mempool/mempool.go` around lines 436 - 446, The string-based error detection block that checks err.Error() for "Invalid pattern!" and the JSON unmarshal text is fragile; update the code around m.kupoInvalidPatternLogged / m.kupoDisabled and the logger.Debug call by adding a clear comment that documents the exact kugo library version (and commit/tag) this behavior was validated against, why we match these substrings, and a TODO to re-verify or replace with a sentinel error when upgrading kugo; optionally extract the literal substrings into named constants to make the coupling explicit for future maintainers.
104-111: Dual configuration path — CLI option vs global config fallback.The Kupo URL is sourced from two places: the mempool-specific
WithKupoUrloption (via CLI/env) and the globalconfig.GetConfig().KupoUrl. The chainsync plugin does the same (seeinput/chainsync/chainsync.golines 137). While this provides convenience, it could surprise users if the global config value unexpectedly overrides an intentionally empty CLI option — the current code cannot distinguish "user didn't set it" from "user explicitly left it empty."This is consistent with the existing chainsync approach, so probably fine for now, but worth noting.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@input/mempool/mempool.go` around lines 104 - 111, The code currently falls back to config.GetConfig().KupoUrl whenever m.kupoUrl is empty, which makes it impossible to tell if the user explicitly provided an empty value or never set the option; change the behavior so fallback only happens when the user did NOT set the mempool option. Add a presence flag (e.g., m.kupoUrlSet) that is set by the WithKupoUrl option (or detect presence via the option builder), then update the logic around m.kupoUrl to: if !m.kupoUrlSet { m.kupoUrl = config.GetConfig().KupoUrl } and keep the current logging using m.kupoUrl; mirror the same pattern used by the chainsync plugin (see input/chainsync/chainsync.go) to avoid surprising overrides.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@input/mempool/mempool.go`:
- Around line 427-462: The resolveTransactionInputs code uses
context.Background() for each Kupo query which prevents shutdown from cancelling
in-flight requests; change it to derive per-request contexts from a cancellable
parent context stored on the mempool struct (e.g., m.ctx/m.cancel) that Stop()
cancels: add a context.Context and context.CancelFunc fields to the mempool,
initialize them when starting/polling, call m.cancel() in Stop(), and replace
context.WithTimeout(context.Background(), defaultKupoTimeout) with
context.WithTimeout(m.ctx, defaultKupoTimeout) so k.Matches(ctx, ...) is
cancelled on shutdown; keep existing timeout/error handling and behavior for
m.kupoDisabled and logging intact.
- Around line 373-376: getKupoClient currently returns a cached m.kupoClient
without re-validating connectivity; update getKupoClient to perform a
lightweight health/ping check on the cached client and if that check fails clear
m.kupoClient (set to nil) and return an error so a fresh client will be created
on the next call. Specifically, in getKupoClient check the existing m.kupoClient
via its Health/Ping method (or perform a trivial request), and on any failure
assign m.kupoClient = nil and return the error; keep the existing flow of
creating and caching a client when no cached client exists. Ensure you reference
m.kupoClient and getKupoClient when making the change so future connection
losses cause revalidation and recreation.
- Around line 391-418: The health check sets a deadline twice
(context.WithTimeout using kupoHealthTimeout and http.Client{Timeout:
kupoHealthTimeout}); remove the redundant client-level timeout and rely on the
request context instead. Specifically, delete or change the instantiation of
httpClient that sets Timeout (the &http.Client{Timeout: kupoHealthTimeout}
creation used with httpClient.Do(req)) so the client does not impose its own
timeout (use the default client or a plain &http.Client{}), keeping the
context-based timeout via context.WithTimeout and preserving the existing error
handling around http.NewRequestWithContext, httpClient.Do, resp handling, and
assignment to m.kupoClient / return k.
- Around line 436-446: The string-based error detection block that checks
err.Error() for "Invalid pattern!" and the JSON unmarshal text is fragile;
update the code around m.kupoInvalidPatternLogged / m.kupoDisabled and the
logger.Debug call by adding a clear comment that documents the exact kugo
library version (and commit/tag) this behavior was validated against, why we
match these substrings, and a TODO to re-verify or replace with a sentinel error
when upgrading kugo; optionally extract the literal substrings into named
constants to make the coupling explicit for future maintainers.
- Around line 104-111: The code currently falls back to
config.GetConfig().KupoUrl whenever m.kupoUrl is empty, which makes it
impossible to tell if the user explicitly provided an empty value or never set
the option; change the behavior so fallback only happens when the user did NOT
set the mempool option. Add a presence flag (e.g., m.kupoUrlSet) that is set by
the WithKupoUrl option (or detect presence via the option builder), then update
the logic around m.kupoUrl to: if !m.kupoUrlSet { m.kupoUrl =
config.GetConfig().KupoUrl } and keep the current logging using m.kupoUrl;
mirror the same pattern used by the chainsync plugin (see
input/chainsync/chainsync.go) to avoid surprising overrides.
In `@input/mempool/options.go`:
- Around line 79-84: Rename the exported option function WithKupoUrl to
WithKupoURL and update all call sites and plugin wiring that reference it
(including the CLI wiring in plugin.go that currently uses the `kupo-url` flag)
so they call WithKupoURL instead; leave the unexported field kupoUrl unchanged,
run go vet/gofmt to ensure exports are consistent, and update any tests or
imports that referenced WithKupoUrl to the new WithKupoURL identifier.
Signed-off-by: cryptodj413 <shinjirohara2@gmail.com>
There was a problem hiding this comment.
🧹 Nitpick comments (2)
input/mempool/mempool.go (2)
373-419: No negative caching — failed health checks are retried on everyresolveTransactionInputscall.When Kupo is unreachable,
getKupoClientis called for every new transaction in every poll cycle (every 5 s by default). Each call performs a full HTTP health check that will time out after 3 s, adding significant latency and generating repetitive warn logs.Consider caching the failure with a short backoff (e.g., skip retries for 30–60 s after a health-check failure) so the plugin doesn't repeatedly block on a dead endpoint:
♻️ Sketch: add a cooldown after health-check failure
type Mempool struct { // ...existing fields... kupoClient *kugo.Client kupoDisabled bool kupoInvalidPatternLogged bool + kupoLastHealthFail time.Time } func (m *Mempool) getKupoClient() (*kugo.Client, error) { if m.kupoClient != nil { return m.kupoClient, nil } + // Back off for 60s after a failed health check + if !m.kupoLastHealthFail.IsZero() && time.Since(m.kupoLastHealthFail) < 60*time.Second { + return nil, errors.New("kupo health check cooling down") + } // ...existing URL validation and health check code... resp, err := httpClient.Do(req) if err != nil { + m.kupoLastHealthFail = time.Now() // ...existing error handling... }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@input/mempool/mempool.go` around lines 373 - 419, getKupoClient currently retries a full 3s health check on every call when Kupo is down, causing repeated latency and logs; add short negative-caching: introduce a small cooldown field (e.g., m.kupoFailUntil time.Time) and check at the top of getKupoClient (if time.Now().Before(m.kupoFailUntil) return nil, fmt.Errorf("kupo unavailable until %v", m.kupoFailUntil)); on any health-check failure set m.kupoFailUntil = time.Now().Add(30*time.Second) (or 60s) and return the error immediately; protect reads/writes with the existing mutex (or add one, e.g., m.mu) around accesses to m.kupoClient and m.kupoFailUntil to avoid races.
436-446: Fragile string-based error detection for disabling Kupo.The
"Invalid pattern!"and JSON unmarshal error substrings depend on implementation details of Kupo's HTTP API response format and Go'sencoding/json. If Kupo changes its error response body, this detection silently breaks — instead of disabling gracefully, each input resolution would fail repeatedly and log warnings every poll cycle.Consider at minimum adding a comment documenting why these specific strings are matched and which Kupo versions they apply to. The JSON unmarshal error (
"cannot unmarshal object into Go value of type []kugo.Match") is stable (from Go's standard library), but the"Invalid pattern!"string is not. Ideally,kugocould expose a typed error (e.g.,kugo.ErrInvalidPattern) upstream, butkugov1.3.0 currently does not provide this.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@input/mempool/mempool.go` around lines 436 - 446, Replace the fragile substring checks for disabling Kupo with a more robust approach: in the block that currently inspects err.Error() (around m.kupoInvalidPatternLogged, m.kupoDisabled, and logger.Debug), first try to detect a typed sentinel from the kupo client (e.g., kugo.ErrInvalidPattern or any exported error on the kupo package) and use that to disable resolution; if no typed error exists, fall back to a documented regex match or HTTP-status-based check (and keep the existing JSON unmarshal substring only as a secondary fallback), update the logger.Debug call to include the full error, and add a clear comment referencing kupo v1.3.0 and why the specific strings/regex are used so future maintainers know the rationale and can remove the fallback when kupo provides a typed error.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@input/mempool/mempool.go`:
- Around line 373-419: getKupoClient currently retries a full 3s health check on
every call when Kupo is down, causing repeated latency and logs; add short
negative-caching: introduce a small cooldown field (e.g., m.kupoFailUntil
time.Time) and check at the top of getKupoClient (if
time.Now().Before(m.kupoFailUntil) return nil, fmt.Errorf("kupo unavailable
until %v", m.kupoFailUntil)); on any health-check failure set m.kupoFailUntil =
time.Now().Add(30*time.Second) (or 60s) and return the error immediately;
protect reads/writes with the existing mutex (or add one, e.g., m.mu) around
accesses to m.kupoClient and m.kupoFailUntil to avoid races.
- Around line 436-446: Replace the fragile substring checks for disabling Kupo
with a more robust approach: in the block that currently inspects err.Error()
(around m.kupoInvalidPatternLogged, m.kupoDisabled, and logger.Debug), first try
to detect a typed sentinel from the kupo client (e.g., kugo.ErrInvalidPattern or
any exported error on the kupo package) and use that to disable resolution; if
no typed error exists, fall back to a documented regex match or
HTTP-status-based check (and keep the existing JSON unmarshal substring only as
a secondary fallback), update the logger.Debug call to include the full error,
and add a clear comment referencing kupo v1.3.0 and why the specific
strings/regex are used so future maintainers know the rationale and can remove
the fallback when kupo provides a typed error.
2 participants
Closes #497
Summary by cubic
Adds Kupo support to resolve mempool transaction inputs and include them in transaction events. Adds a --kupo-url (KUPO_URL) option with health checks, timeouts, and graceful fallback when Kupo is unavailable or patterns aren’t supported.
New Features
Bug Fixes
Written for commit 6137c31. Summary will update on new commits.
Summary by CodeRabbit
New Features
Configuration
kupo-urlcommand-line option andWithKupoUrl()configuration method to enable Kupo integration (optional, disabled by default).Improvements