fix(platform-wallet): commit wallet events off the async runtime - #4370
fix(platform-wallet): commit wallet events off the async runtime#4370romchornyi wants to merge 1 commit into
Conversation
`run_wallet_event_adapter` called `commit_batch` — and through it `persister.store()` — inline on the tokio worker driving it. `store()` is synchronous, and for the SQLite backend commits a real transaction per call; its own trait docs say so, and warn that a slow write blocks every other wallet accessor for its duration. What they do not say, because until now it was not true, is that it also blocks the runtime those accessors run on. Field evidence from a testnet restore of a 6663-transaction wallet: - drains coalesced into ever larger, ever rarer batches — folded 1 → 47 → 164 → 512, with gaps of 42s, 144s and finally 1109s between them; - the metrics tick covering the 512-event drain reported `busy_ratio=1106 mean_poll_us=1397886` — a 1.4s mean poll on a runtime that read 24µs one second later; - `Blocks: last_activity: 549s` at the same moment, so the SPV managers sharing that runtime were starved, not idle; - the durable watermark topped out at height 2179999 against a chain tip of 2520064 and never caught up, so the home timeline — which only advances when a batch lands — showed roughly a third of the history ten minutes after core sync reported 100%. The commit now runs on `spawn_blocking`. The handle is awaited rather than raced against `cancel`: a store that has started must finish, and dropping the handle would not stop the thread in any case — shutdown is observed at the next `recv`. `AdapterFaultState` and the freeze latch move behind an `Arc<Mutex<..>>` and an `Arc<AtomicBool>` rather than being moved into the closure by value. That is deliberate: if the commit thread ever panicked, moving them would lose a wallet's frozen watermark, which would un-freeze a wallet whose verification had failed — the one outcome the fail-closed guard exists to prevent. The lock is uncontended by construction (one drain commits at a time, and this task is the only writer). A panicking commit thread is now reported and the drain skipped, rather than taking the adapter down with it. cargo test -p platform-wallet --lib # 662 passed cargo clippy --all-targets + fmt # clean
|
⛔ Blockers found — Opus deferred (commit e2b806b) |
📝 WalkthroughWalkthroughThe adapter now persists folded batches through ChangesBatch persistence execution
Estimated code review effort: 3 (Moderate) | ~20 minutes Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![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 `@packages/rs-platform-wallet/src/changeset/core_bridge.rs`:
- Around line 409-420: Update the commit-task panic handling around the
committed match to capture all batch wallet IDs before moving batch into the
closure, then in the Err(join_error) branch lock fault and call fault_wallet()
for each captured ID before continuing. Preserve the existing error log and
ensure the normal Ok(diag) path remains unchanged.
🪄 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: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: dccdef63-12af-43c6-98ec-0003f1aca707
📒 Files selected for processing (1)
packages/rs-platform-wallet/src/changeset/core_bridge.rs
| let diag = match committed { | ||
| Ok(diag) => diag, | ||
| // The commit thread panicked. The fault state survives (it lives | ||
| // behind the handle above), but this batch's outcome is unknown, | ||
| // so it is reported rather than silently folded into the next one. | ||
| Err(join_error) => { | ||
| tracing::error!( | ||
| error = %join_error, | ||
| folded, | ||
| "wallet-event commit thread failed; batch outcome unknown" | ||
| ); | ||
| continue; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Freeze wallets when the commit task panics.
A panic bypasses commit_batch and its fault.fault_wallet() calls. The JoinError branch only logs and continues. The next drain can persist a later sync height although rows from this batch have an unknown outcome.
Capture the batch wallet IDs before moving batch into the closure. In the Err(join_error) branch, lock fault and call fault_wallet() for every captured wallet ID before continuing. This preserves the fail-closed watermark rule for panics.
Proposed fix
+ let batch_wallet_ids: Vec<WalletId> = batch.keys().copied().collect();
let committed = tokio::task::spawn_blocking(move || {
// ...
})
.await;
let diag = match committed {
Ok(diag) => diag,
Err(join_error) => {
+ let mut fault = fault
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner());
+ for wallet_id in batch_wallet_ids {
+ fault.fault_wallet(wallet_id, &sync_fault);
+ }
tracing::error!(
error = %join_error,
folded,
"wallet-event commit thread failed; batch outcome unknown"
);
continue;
}
};🤖 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 `@packages/rs-platform-wallet/src/changeset/core_bridge.rs` around lines 409 -
420, Update the commit-task panic handling around the committed match to capture
all batch wallet IDs before moving batch into the closure, then in the
Err(join_error) branch lock fault and call fault_wallet() for each captured ID
before continuing. Preserve the existing error log and ensure the normal
Ok(diag) path remains unchanged.
thepastaclaw
left a comment
thepastaclaw
left a comment
There was a problem hiding this comment.
Preliminary review — Codex only
Moving synchronous persistence into spawn_blocking correctly prevents wallet commits from parking Tokio workers, but the new panic branch continues after consuming a batch whose persistence outcome is unknown, allowing a later watermark to advance past missing rows. The adapter must fail closed after a commit panic, and the off-runtime boundary should have deterministic regression coverage.
Source: reviewer backend gpt-5.6-sol (Codex general and Rust-quality lanes); final verifier backend gpt-5.6-sol (Codex). openclaw-agent/cliproxy/gpt-5.6-sol is orchestration-only and not reviewer evidence.
Validated blockers were found in the Codex precheck. Opus is deferred until a fresh Codex revalidation clears the blocker gate.
Review provenance
- Codex reviewers:
gpt-5.6-sol— general (completed),gpt-5.6-sol— rust-quality (completed) - Verifier:
gpt-5.6-sol— verifier - Sonnet: not run (deferred by blocker gate)
🔴 1 blocking | 🟡 1 suggestion(s)
🤖 Prompt for all review comments with AI agents
These findings are from an automated code review. Verify each finding against the current code and only fix it if needed.
In `packages/rs-platform-wallet/src/changeset/core_bridge.rs`:
- [BLOCKING] packages/rs-platform-wallet/src/changeset/core_bridge.rs:409-420: Continuing after a commit panic can advance the watermark past lost rows
`commit_batch` consumes the folded batch and calls `store()` for each wallet. If `store()` panics before its outcome is known, unwinding bypasses the `Err` arm that calls `fault_wallet`, drops the remainder of the consumed batch, and returns a `JoinError`. This branch then continues with an unaffected fault state, so a later event for the same wallet can successfully persist a higher `synced_height` even though rows from the panicked batch may be absent. That violates the adapter's fail-closed invariant; before this PR, the panic terminated the adapter and prevented later watermark advancement. Stop the adapter and latch `sync_fault`, or capture every batch wallet ID before moving the batch and fault all of them before continuing.
- [SUGGESTION] packages/rs-platform-wallet/src/changeset/core_bridge.rs:388-407: The off-runtime persistence boundary has no regression test
The existing `ProbePersister` returns immediately and checks only persistence outcomes. Those tests still pass if `commit_batch` is moved back inline onto the Tokio worker, so they do not protect the primary behavior introduced by this PR. Add a controllably blocking persister and run the adapter on a current-thread or single-worker runtime, then verify that another future makes progress while `store()` remains blocked. The fixture should also cover a panicking store and assert that no later store for the affected wallet carries `synced_height`.
| let diag = match committed { | ||
| Ok(diag) => diag, | ||
| // The commit thread panicked. The fault state survives (it lives | ||
| // behind the handle above), but this batch's outcome is unknown, | ||
| // so it is reported rather than silently folded into the next one. | ||
| Err(join_error) => { | ||
| tracing::error!( | ||
| error = %join_error, | ||
| folded, | ||
| "wallet-event commit thread failed; batch outcome unknown" | ||
| ); | ||
| continue; |
There was a problem hiding this comment.
🔴 Blocking: Continuing after a commit panic can advance the watermark past lost rows
commit_batch consumes the folded batch and calls store() for each wallet. If store() panics before its outcome is known, unwinding bypasses the Err arm that calls fault_wallet, drops the remainder of the consumed batch, and returns a JoinError. This branch then continues with an unaffected fault state, so a later event for the same wallet can successfully persist a higher synced_height even though rows from the panicked batch may be absent. That violates the adapter's fail-closed invariant; before this PR, the panic terminated the adapter and prevented later watermark advancement. Stop the adapter and latch sync_fault, or capture every batch wallet ID before moving the batch and fault all of them before continuing.
| let diag = match committed { | |
| Ok(diag) => diag, | |
| // The commit thread panicked. The fault state survives (it lives | |
| // behind the handle above), but this batch's outcome is unknown, | |
| // so it is reported rather than silently folded into the next one. | |
| Err(join_error) => { | |
| tracing::error!( | |
| error = %join_error, | |
| folded, | |
| "wallet-event commit thread failed; batch outcome unknown" | |
| ); | |
| continue; | |
| Err(join_error) => { | |
| tracing::error!( | |
| error = %join_error, | |
| folded, | |
| "wallet-event commit thread failed; stopping adapter because batch outcome is unknown" | |
| ); | |
| sync_fault.store(true, Ordering::Relaxed); | |
| break; | |
| } |
source: ['codex']
| let committed = tokio::task::spawn_blocking(move || { | ||
| // The lock is uncontended by construction — this task is the only | ||
| // writer, and one drain commits at a time — so it never blocks; | ||
| // it exists to carry the state, not to arbitrate. | ||
| let mut fault = fault_for_commit | ||
| .lock() | ||
| .unwrap_or_else(|poisoned| poisoned.into_inner()); | ||
| let mut freeze_logged = freeze_for_commit.load(Ordering::Relaxed); | ||
| let diag = commit_batch( | ||
| &*persister_for_commit, | ||
| batch, | ||
| folded, | ||
| &mut fault, | ||
| &sync_fault_for_commit, | ||
| &mut freeze_logged, | ||
| ); | ||
| freeze_for_commit.store(freeze_logged, Ordering::Relaxed); | ||
| diag | ||
| }) | ||
| .await; |
There was a problem hiding this comment.
🟡 Suggestion: The off-runtime persistence boundary has no regression test
The existing ProbePersister returns immediately and checks only persistence outcomes. Those tests still pass if commit_batch is moved back inline onto the Tokio worker, so they do not protect the primary behavior introduced by this PR. Add a controllably blocking persister and run the adapter on a current-thread or single-worker runtime, then verify that another future makes progress while store() remains blocked. The fixture should also cover a panicking store and assert that no later store for the affected wallet carries synced_height.
source: ['codex']
3 participants
Issue being fixed or feature implemented
run_wallet_event_adaptercalledcommit_batch— and through itpersister.store()— inline on the tokio worker driving it.store()is synchronous and, for the SQLite backend, commits a real transaction per call. Its own trait docs (traits.rs:200-206, 263-267) already warn that a slow write blocks every other wallet accessor for its duration; what they do not say, because until now it was not true, is that it also blocks the runtime those accessors run on.Found while investigating a user report that a restored wallet's transaction history appears only in large, minutes-apart jumps long after Core sync reports 100%.
Evidence from a testnet restore of a 6663-transaction wallet (52k-line session log):
folded1 → 47 → 164 → 512 (theADAPTER_STORE_BATCH_LIMITceiling), with gaps of 42s, 144s and finally 1109s between them.Blocks: … last_activity: 549sat the same moment — the SPV managers sharing that runtime were starved, not idle.The escalating
foldedcounts are the symptom, not the cause: events pile up in the channel because the previous drain's synchronousstore()is still holding a worker.What was done?
commit_batchnow runs ontokio::task::spawn_blocking.The handle is awaited rather than raced against
cancel. A store that has started must be allowed to finish, and dropping aspawn_blockinghandle does not stop the thread in any case. Shutdown is observed at the nextrecv, which is where the loop already handles it.AdapterFaultStateand the freeze latch move behindArc<Mutex<..>>/Arc<AtomicBool>instead of being moved into the closure by value. This is the part worth reviewing: moving them would mean a panicking commit thread loses a wallet's frozen watermark — un-freezing a wallet whose verification had failed, which is the single outcome the fail-closed guard exists to prevent. The lock is uncontended by construction (this task is the only writer, and one drain commits at a time), so it carries state rather than arbitrating access.A panicking commit thread is now reported and its drain skipped, rather than taking the adapter down with it.
Not done here
Nothing about batch sizing, the channel, or
ADAPTER_STORE_BATCH_LIMIT. With the blocking call off the runtime the coalescing behaves as designed; tuning it before re-measuring would be guessing.How Has This Been Tested?
Not covered: no test reproduces the stall. Doing so needs a persister whose
store()blocks for a controllable duration plus assertions on runtime poll latency — worth adding, but it would not have caught this class of bug by construction, only this instance of it. The change is behaviour-preserving for the commit itself: the samecommit_batch, the same inputs, the same diagnostics line.A before/after restore on device is the measurement that matters, and I have the "before" trace above to compare against.
Breaking Changes
None. No public API changes;
PlatformWalletPersistenceimplementors are unaffected (the trait already requiresSend + Sync).Checklist:
For repository code-owners and collaborators only
Summary by CodeRabbit
Bug Fixes
Performance