feat(pipeline): log transaction lifecycle timestamps in the order pool - #114
Open
MavenRain wants to merge 1 commit into
Open
feat(pipeline): log transaction lifecycle timestamps in the order pool#114MavenRain wants to merge 1 commit into
MavenRain wants to merge 1 commit into
Conversation
Closes flashbots#9. Adds `TxLifecycleLog` to the order pool. For every transaction the pool sees it records when the transaction arrived (reth pool acceptance, stamped with `ValidPoolTransaction::timestamp`, or `eth_sendBundle`), when a payload job first considered it, when it was first included in a payload, and when it was mined in a canonical block, and emits one `info!` line per mined transaction on the `rblib::pool::lifecycle` target with the three deltas. Reverted blocks clear the mined stage until the transaction is mined again. `OrderPool::remove` becomes the single eviction path: members are marked dropped only when no other live order holds them, and never when they were received from the host node mempool. The host maintenance loop drains queued arrivals before processing a canonical event, so "received" stays ahead of "mined". The log is a bounded window: entries are pruned by retention (default 2 min) on committed blocks and on demand at capacity (default 250 000, ~2k tx/s sustained); both are builder options on `TxLifecycleLog`, which also exposes `get`, `snapshot` and refusal counters. 30 unit tests with explicit instants and one LocalNode end-to-end test (Ethereum + Optimism). Signed-off-by: Onyeka Obi <softwareengineerasaservant@isurvivable.cv>
Contributor
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Adds a transaction lifecycle log to the order pool to track when transactions are received, considered, included, mined, dropped, and reverted, including late pipeline events and reorg handling.
Changes:
- Introduces
TxLifecycleLogwith retention/capacity pruning and structured stage recording. - Wires lifecycle recording into pool insertion/removal, pipeline event listeners, mempool listener, and canonical chain commit/revert handling.
- Adds unit + integration tests covering lifecycle transitions and listener edge cases.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/pipeline/src/pool/report.rs | Records inclusion/commit/revert lifecycle stages and prunes lifecycle log on committed blocks. |
| crates/pipeline/src/pool/mod.rs | Adds lifecycle log to OrderPool and updates insert/remove paths to record lifecycle transitions. |
| crates/pipeline/src/pool/maintain.rs | Improves pipeline event listener behavior (biased drop arm + buffered drain) and routes inclusion events into lifecycle recording. |
| crates/pipeline/src/pool/lifecycle.rs | New lifecycle log implementation with pruning, capacity guarding, and extensive tests. |
| crates/pipeline/src/pool/host.rs | Subscribes to mempool transaction events and records arrival timestamps; drains queued tx events before canonical processing. |
| crates/pipeline/src/pipelines/tests/mod.rs | Registers new lifecycle test module. |
| crates/pipeline/src/pipelines/tests/lifecycle.rs | Adds end-to-end tests validating lifecycle tracking through pipeline + node + mining. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| /// see the sizing and memory notes on the type. | ||
| pub const DEFAULT_CAPACITY: usize = 250_000; | ||
| /// Entries whose last activity is older than this are pruned by default. | ||
| pub const DEFAULT_RETENTION: Duration = Duration::from_mins(2); |
|
|
||
| #[test] | ||
| fn prune_removes_only_entries_older_than_retention() { | ||
| let retention = Duration::from_mins(1); |
| assert_eq!(TxLifecycleLog::default().capacity(), 250_000); | ||
| assert_eq!( | ||
| TxLifecycleLog::default().retention(), | ||
| Duration::from_mins(2) |
| assert_eq!(log.refused_orders(), 1); | ||
|
|
||
| // once the records are stale a prune makes room | ||
| let later = base + 2 * Duration::from_mins(1); |
Comment on lines
+851
to
+861
| pub(super) fn prune_if_due(&self, now: Instant) -> Option<usize> { | ||
| let interval = self.retention / 10; | ||
| let due = { | ||
| let mut last = self.shared.last_prune.lock(); | ||
| let due = | ||
| last.is_none_or(|last| now.saturating_duration_since(last) >= interval); | ||
| if due { | ||
| *last = Some(now); | ||
| } | ||
| due | ||
| }; |
Comment on lines
+1524
to
+1526
| let size = size_of::<TxLifecycle>(); | ||
| assert!(size <= 256, "TxLifecycle is {size} bytes"); | ||
| let record = size_of::<OrderRecord>(); |
Comment on lines
+121
to
+122
| let wall = SystemTime::now() | ||
| .checked_sub(at.elapsed()) |
Comment on lines
141
to
+143
| self.inner.host.remove_transaction(*order_hash); | ||
|
|
||
| if let Some((_, order)) = removed { |
Comment on lines
+100
to
113
| // record the mined stage of all transactions in the block that this | ||
| // pool has seen, before they get evicted from the pool below. | ||
| block.body().transactions().iter().for_each(|tx| { | ||
| self.inner.lifecycle.record_mined( | ||
| *tx.tx_hash(), | ||
| block_number, | ||
| block_timestamp, | ||
| now, | ||
| ); | ||
| }); | ||
|
|
||
| // remove all orders that had any of their transactions included in the | ||
| // payload | ||
| for tx in block.body().transactions() { |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Closes #9.
Adds a transaction lifecycle log to the order pool: for every transaction the pool sees, it records when the transaction arrived (mempool acceptance or
eth_sendBundle), when it was first considered by a payload job, when it was first included in a payload, and when it was mined in a canonical block, and logs one line per mined transaction with the three deltas. Reorgs clear the mined stage until the transaction is mined again.What is added
pool/lifecycle.rs:TxLifecycleLog(cheap-to-clone handle,with_retention/with_capacitybuilders,get/snapshot/len/refusedaccessors) holding oneTxLifecycleper transaction hash:source()(TxSource::{Mempool, Bundle(hash), Transaction}),received_at(),first_considered()/first_included()(TxPayloadStage: instant + payload id),mined()(TxMined: instant, block number, block timestamp),dropped(),reorg_count(), and the deltastime_to_first_consideration()/time_to_inclusion()/time_to_mined(). The state machine is documented on the type (arrival, considered, included, mined or dropped, with revert and revive edges).rblib::pool::lifecycle:debug!per stage,info!on mined withreceived_at_unix_ms,to_first_consideration_ms,to_inclusion_ms,to_mined_ms,block,source(andreorgswhen non-zero), soRUST_LOG=rblib::pool::lifecycle=infogives one line per mined transaction the pool saw.OrderPool:insertrecords arrival for bundle members and API-submitted transactions; the existingreport_inclusion_attemptstub now records "considered" for every member of the order, and the newreport_inclusion_successrecords "included";report_committed_blockrecords "mined" for every transaction of the block; the newreport_reverted_blockclears "mined" on reorgs;removeis now the single eviction path and marks members "dropped" only when no other live order holds them (never for mempool-sourced transactions, which still live in the reth pool).OrderPool::lifecycle()exposes the log,OrderPool::with_lifecycle(log)installs a configured one.HostNode: subscribes to the reth pool'snew_transactions_listener_for(TransactionListenerKind::All)at attach time and stamps mempool arrivals with reth's ownValidPoolTransaction::timestamp(the moment the pool validated the transaction), so arrival is not "first time we considered it". On every canonical-chain event the maintenance loop first drains the arrivals already queued on the listener (boundedtry_recvloop; a transaction's arrival is always queued before any block that can contain it), then processes reverted blocks, then committed ones. The select stays fair, so mempool traffic cannot starve the canonical arm.Bounds and behaviour under load
ASSUMED_ARRIVAL_RATE_PER_SEC). When still full, new transactions are not tracked andrefused()(refused_orders()for bundle records) is bumped. Both knobs are builder options; raise them together for a longer window.snapshot().Testing
pool/lifecycle.rs, 3 inpool/host.rsand 1 listener test inpipelines/tests/lifecycle.rs(events buffered before the pipeline drop are still recorded), all with explicit instants (happy path deltas, first-sighting-wins, unknown hashes are no-ops, prune boundary, capacity + prune-on-demand + refused counters, bounded bundle records, clones share the log, bundle members through the order-hash resolution, late events for an evicted bundle still reach its members, considered-before-received provisional entry, reorg clears and re-stamps mined, dropped then mined or re-received revives, eviction drops only orphaned members and never mempool transactions, invalidated bundles drop their members, committed and reverted blocks reach bundle members, listener drain bound and disconnect, entry size envelope, defaults cover the assumed arrival rate).pipelines::tests::lifecycle::transaction_lifecycle_is_recorded_end_to_end, Ethereum + Optimism) on aLocalNode: send a transfer over RPC, build a block, drive it canonical, assert source Mempool, considered, included, mined with the block number/timestamp, and ordered deltas.cargo +nightly fmt --check, stable clippy (all-features and no-default-features,-D warnings) on the changed crate,cargo docwithRUSTDOCFLAGS=-D warnings,typos.Note: with the current stable toolchain, clippy on
mainalready fails on two pre-existing lints inorderpool2(fixed in a separate small PR), and a freshCargo.lockresolvesalloy-*to 1.8.x which conflicts withop-alloy-network 0.23.1; I built locally with the alloy family pinned to 1.7.3. Neither is touched here.