Conversation
📝 WalkthroughWalkthroughThe PR enhances the DirectSubmitter with concurrent submission controls, per-signer sequence management, and recovery logic for pending transactions. It introduces in-flight limits, sequence reconciliation after restarts, and comprehensive test coverage for edge cases including sequence mismatches and recovery scenarios. Changes
Sequence DiagramsequenceDiagram
participant Client as Submit Client
participant Submitter as DirectSubmitter
participant Blockchain as Blockchain (BroadcastTx)
participant Confirmer as Confirmation Handler
Client->>Submitter: Submit(tx)
activate Submitter
Submitter->>Submitter: startSubmission (increment inFlight)
Submitter->>Submitter: Fetch next sequence
rect rgba(100, 150, 200, 0.5)
Note over Submitter,Blockchain: Broadcast Path
Submitter->>Blockchain: BroadcastTx(tx with nonce)
Blockchain-->>Submitter: TxHash or Error
end
alt Sequence Mismatch Error
Submitter->>Submitter: Extract expected sequence from error
Submitter->>Submitter: reconcilePendingLocked()
Submitter->>Submitter: invalidateSequenceLocked()
Submitter->>Submitter: rememberPendingLocked(txHash)
else Success
Submitter->>Confirmer: Wait for confirmation
Confirmer-->>Submitter: Confirmation received
Submitter->>Submitter: clearPending()
end
Submitter->>Submitter: finishSubmission (decrement inFlight)
Submitter-->>Client: Result
deactivate Submitter
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
📝 Coding Plan
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 |
|
@codex review |
![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
🧹 Nitpick comments (1)
pkg/submit/direct.go (1)
49-55: Consider documenting themaxInFlightdefault behavior.When
maxInFlightis 0 (the default, since it's not initialized inNewDirectSubmitter), the in-flight limit is effectively disabled due to the checks.maxInFlight > 0 && s.inFlight >= s.maxInFlight. This is reasonable for backward compatibility, but a comment would clarify the intent.📝 Suggested documentation
mu sync.Mutex inFlight int accountNumber uint64 nextSequence uint64 sequenceReady bool pendingSequences map[string]uint64 + // maxInFlight limits concurrent submissions; 0 means unlimited. maxInFlight int🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/submit/direct.go` around lines 49 - 55, Document the default behavior of the maxInFlight field: add a brief comment on the maxInFlight field (or in NewDirectSubmitter) stating that a value of 0 disables the in-flight limit for backward compatibility and that the code enforces the limit only when s.maxInFlight > 0 (see the check s.maxInFlight > 0 && s.inFlight >= s.maxInFlight). This clarifies intent for future maintainers and callers of DirectSubmitter.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@pkg/submit/direct.go`:
- Around line 243-266: The reconcilePendingLocked loop can leave entries in the
pendingSequences map forever when GetTx returns NotFound; update the data model
and logic: change pendingSequences to store a struct with the original sequence
and a timestamp (e.g., pendingEntry{seq int64, added time.Time}), and in
reconcilePendingLocked (and wherever entries are added) remove entries whose
added timestamp is older than a configurable threshold (or enforce a max map
size by evicting oldest entries) instead of keeping them indefinitely; reference
the reconcilePendingLocked method, pendingSequences map, GetTx call and
isTxNotFound branch to implement the time-based cleanup or size eviction and
update s.nextSequence logic accordingly.
---
Nitpick comments:
In `@pkg/submit/direct.go`:
- Around line 49-55: Document the default behavior of the maxInFlight field: add
a brief comment on the maxInFlight field (or in NewDirectSubmitter) stating that
a value of 0 disables the in-flight limit for backward compatibility and that
the code enforces the limit only when s.maxInFlight > 0 (see the check
s.maxInFlight > 0 && s.inFlight >= s.maxInFlight). This clarifies intent for
future maintainers and callers of DirectSubmitter.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: c7ecbbc3-76eb-4e0a-9469-cd70408cc776
📒 Files selected for processing (2)
pkg/submit/direct.gopkg/submit/direct_test.go
| func (s *DirectSubmitter) reconcilePendingLocked(ctx context.Context) error { | ||
| if len(s.pendingSequences) == 0 { | ||
| return nil | ||
| } | ||
|
|
||
| nextSequence := s.nextSequence | ||
| for hash, sequence := range s.pendingSequences { | ||
| _, err := s.app.GetTx(ctx, hash) | ||
| if err == nil { | ||
| delete(s.pendingSequences, hash) | ||
| continue | ||
| } | ||
| return nil, fmt.Errorf("broadcast blob tx: %w", err) | ||
| if isTxNotFound(err) { | ||
| if sequence >= nextSequence { | ||
| nextSequence = sequence + 1 | ||
| } | ||
| continue | ||
| } | ||
| return fmt.Errorf("reconcile pending blob tx %s: %w", hash, err) | ||
| } | ||
| if err := checkTxStatus("broadcast", broadcast); err != nil { | ||
| return nil, err | ||
|
|
||
| s.nextSequence = nextSequence | ||
| return nil | ||
| } |
There was a problem hiding this comment.
Potential unbounded growth of pendingSequences map.
When GetTx returns NotFound, the entry is not deleted from pendingSequences. This is correct for in-flight transactions that haven't been indexed yet. However, if a transaction is dropped from the mempool or evicted, the entry will remain indefinitely, leading to slow memory growth over time.
Consider adding a cleanup mechanism, such as removing entries older than a certain age or limiting the map size.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/submit/direct.go` around lines 243 - 266, The reconcilePendingLocked loop
can leave entries in the pendingSequences map forever when GetTx returns
NotFound; update the data model and logic: change pendingSequences to store a
struct with the original sequence and a timestamp (e.g., pendingEntry{seq int64,
added time.Time}), and in reconcilePendingLocked (and wherever entries are
added) remove entries whose added timestamp is older than a configurable
threshold (or enforce a max map size by evicting oldest entries) instead of
keeping them indefinitely; reference the reconcilePendingLocked method,
pendingSequences map, GetTx call and isTxNotFound branch to implement the
time-based cleanup or size eviction and update s.nextSequence logic accordingly.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5693353a33
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| s.sequenceReady = true | ||
| if err := s.reconcilePendingLocked(ctx); err != nil { | ||
| return nil, err |
There was a problem hiding this comment.
nextAccountLocked sets sequenceReady to true before running reconcilePendingLocked, so a transient GetTx error during reconciliation returns from Submit while leaving the cache marked ready. After that, later submissions skip both account refresh and reconciliation, continuing with potentially stale nextSequence/pending state instead of retrying reconciliation on the next request.
Useful? React with 👍 / 👎.
| for { | ||
| tx, err := s.app.GetTx(ctx, hash) | ||
| if err == nil { | ||
| s.clearPending(hash) |
There was a problem hiding this comment.
Calling s.clearPending(hash) here can block a confirmed submission behind another goroutine that is still inside broadcastTx, because clearPending and broadcastTx share s.mu and broadcastTx holds it across RPC calls. In that situation, this request has already observed confirmation but cannot return until the unrelated broadcast finishes, which can add unbounded latency and bypass the intended confirmation-timeout behavior.
Useful? React with 👍 / 👎.
1 participant
Overview
Summary by CodeRabbit
Release Notes
New Features
Bug Fixes
Tests