From 416a0d69dc005abfbfdff2f73aec2f1acd0a8a07 Mon Sep 17 00:00:00 2001 From: Ammar Arif Date: Mon, 25 May 2026 23:33:05 -0500 Subject: [PATCH 1/3] feat(messaging): expose pending messages via service handle Track messages gathered from the settlement chain but not yet accepted by the tx pool in a volatile in-memory buffer, queryable through MessagingServiceHandle::pending_messages(). Co-Authored-By: Claude Opus 4.7 --- crates/messaging/src/service.rs | 177 +++++++++++++++++++++++++++++++- 1 file changed, 175 insertions(+), 2 deletions(-) diff --git a/crates/messaging/src/service.rs b/crates/messaging/src/service.rs index 3e3381311..ce25e2132 100644 --- a/crates/messaging/src/service.rs +++ b/crates/messaging/src/service.rs @@ -1,9 +1,12 @@ +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + use anyhow::Context; use futures::StreamExt; use katana_pool::api::TransactionPool; use katana_pool::TxPool; use katana_primitives::chain::ChainId; -use katana_primitives::transaction::{ExecutableTxWithHash, TxHash}; +use katana_primitives::transaction::{ExecutableTxWithHash, L1HandlerTx, TxHash}; use katana_provider::api::messaging::{ MessagingCheckpoint, MessagingCheckpointProvider, MessagingL1ToL2IndexWriter, }; @@ -14,6 +17,7 @@ use tracing::{info, warn}; use crate::stream::collector::ethereum::EthereumCollector; use crate::stream::collector::starknet::StarknetCollector; +use crate::stream::collector::OrderedMessage; use crate::stream::trigger::IntervalTrigger; use crate::stream::MessageStream; use crate::{MessagingOutcome, Messenger, SettlementChainConfig, LOG_TARGET}; @@ -139,6 +143,8 @@ where let pool = self.pool.clone(); let provider = self.provider.clone(); + let pending = PendingMessages::default(); + let pending_for_task = pending.clone(); let (shutdown_tx, shutdown_rx) = oneshot::channel(); let task_handle = tokio::spawn(async move { @@ -154,6 +160,12 @@ where let total_messages = messages.len(); let mut inserted: usize = 0; + // Mark the whole batch as pending the moment it's gathered. + // Each entry is removed below once its pool insert succeeds; + // any left over after an error-break stays pending and is + // reconciled (overwritten by key) on the next gather. + pending_for_task.insert_batch(&messages); + for msg in messages { let hash = msg.tx.calculate_hash(); info!(target: LOG_TARGET, tx_hash = %format!("{:#x}", hash), "L1Handler transaction added to the pool."); @@ -165,6 +177,9 @@ where Ok(_) => { inserted += 1; + // Accepted by the pool: no longer pending. + pending_for_task.remove(msg.block, msg.tx_index); + // Atomically persist the L1->L2 index entry and the // checkpoint in a single DB transaction. If either // write or the commit fails, NEITHER is persisted — @@ -232,7 +247,7 @@ where info!(target: LOG_TARGET, "Messaging service started."); - Ok(MessagingServiceHandle { shutdown_tx: Some(shutdown_tx), task_handle }) + Ok(MessagingServiceHandle { shutdown_tx: Some(shutdown_tx), task_handle, pending }) } } @@ -296,10 +311,82 @@ where Ok(()) } +/// A message that has been gathered from the settlement chain but not yet +/// accepted by the transaction pool. +#[derive(Debug, Clone)] +pub struct PendingMessage { + /// The settlement block the message was emitted in. + pub block: u64, + /// The transaction index within `block`. + pub tx_index: u64, + /// The settlement-chain transaction hash that emitted the originating event/log. + pub l1_tx_hash: [u8; 32], + /// The hash of the L2 `L1Handler` transaction this message will become. + pub l2_tx_hash: TxHash, + /// The `L1Handler` transaction converted from the settlement-chain event. + pub tx: L1HandlerTx, +} + +/// Volatile, in-memory registry of messages that have been gathered from the +/// settlement chain but not yet accepted by the transaction pool. +/// +/// A message is inserted the moment its batch is gathered ("picked up") and +/// removed the instant the pool accepts it. Entries are keyed by their +/// `(block, tx_index)` position, so a message re-gathered after a failed pool +/// insert overwrites its prior entry rather than duplicating it. +/// +/// This state is intentionally volatile: it is created fresh on each +/// [`MessagingService::start`] and discarded on shutdown. It reflects only the +/// in-flight window of the currently running drain loop, never history. +#[derive(Debug, Clone, Default)] +pub struct PendingMessages { + inner: Arc>>, +} + +impl PendingMessages { + /// Mark every message in a freshly gathered batch as pending. + fn insert_batch(&self, messages: &[OrderedMessage]) { + let mut guard = self.inner.lock().expect("pending messages lock poisoned"); + for msg in messages { + guard.insert( + (msg.block, msg.tx_index), + PendingMessage { + block: msg.block, + tx_index: msg.tx_index, + l1_tx_hash: msg.l1_tx_hash, + l2_tx_hash: msg.tx.calculate_hash(), + tx: msg.tx.clone(), + }, + ); + } + } + + /// Drop the entry for a message the pool has accepted. + fn remove(&self, block: u64, tx_index: u64) { + self.inner.lock().expect("pending messages lock poisoned").remove(&(block, tx_index)); + } + + /// Snapshot the currently pending messages, ordered by `(block, tx_index)`. + pub fn snapshot(&self) -> Vec { + self.inner.lock().expect("pending messages lock poisoned").values().cloned().collect() + } + + /// The number of messages currently pending. + pub fn len(&self) -> usize { + self.inner.lock().expect("pending messages lock poisoned").len() + } + + /// Whether there are no pending messages. + pub fn is_empty(&self) -> bool { + self.inner.lock().expect("pending messages lock poisoned").is_empty() + } +} + /// Handle to a running messaging server, providing lifecycle control. pub struct MessagingServiceHandle { shutdown_tx: Option>, task_handle: JoinHandle<()>, + pending: PendingMessages, } impl std::fmt::Debug for MessagingServiceHandle { @@ -320,6 +407,15 @@ impl MessagingServiceHandle { pub async fn stopped(self) { let _ = self.task_handle.await; } + + /// Snapshot the messages gathered from the settlement chain but not yet + /// accepted by the transaction pool, ordered by `(block, tx_index)`. + /// + /// This is a point-in-time view of volatile in-memory state; an empty result + /// means nothing is currently in flight, not that no messages were ever seen. + pub fn pending_messages(&self) -> Vec { + self.pending.snapshot() + } } #[cfg(test)] @@ -330,6 +426,23 @@ mod tests { use super::*; + /// Builds a stub `OrderedMessage` whose tx internals don't matter — the buffer + /// only keys on `(block, tx_index)`. `nonce` is varied so distinct positions + /// produce distinct computed L2 hashes, letting tests tell entries apart. + fn msg(block: u64, tx_index: u64) -> OrderedMessage { + let tx = L1HandlerTx { + calldata: vec![], + chain_id: ChainId::default(), + message_hash: Default::default(), + paid_fee_on_l1: 0, + nonce: Felt::from(block * 1000 + tx_index), + entry_point_selector: Felt::ZERO, + version: Felt::ZERO, + contract_address: Default::default(), + }; + OrderedMessage { block, tx_index, l1_tx_hash: [0u8; 32], tx } + } + #[test] fn resume_cursor_falls_back_to_default_from_block_when_no_checkpoint_persisted() { let provider = DbProviderFactory::new_in_memory(); @@ -446,4 +559,64 @@ mod tests { assert_eq!(cp.block, 10, "checkpoint should reflect the latest committed message"); assert_eq!(cp.tx_index, 2); } + + /// A gathered batch becomes pending in `(block, tx_index)` order, and each entry + /// disappears once its position is removed (the pool-accepted path). + #[test] + fn pending_messages_track_batch_then_clear_on_remove() { + let pending = PendingMessages::default(); + assert!(pending.is_empty()); + + // Intentionally out of order — the snapshot must come back sorted. + pending.insert_batch(&[msg(7, 1), msg(5, 0), msg(7, 0)]); + + let snapshot = pending.snapshot(); + let positions: Vec<_> = snapshot.iter().map(|m| (m.block, m.tx_index)).collect(); + assert_eq!( + positions, + vec![(5, 0), (7, 0), (7, 1)], + "snapshot should be ordered by (block, tx_index)" + ); + assert_eq!(pending.len(), 3); + + // The pool accepts (5, 0): it leaves the pending set. + pending.remove(5, 0); + + let positions: Vec<_> = pending.snapshot().iter().map(|m| (m.block, m.tx_index)).collect(); + assert_eq!(positions, vec![(7, 0), (7, 1)], "removed position should be gone"); + } + + /// Re-gathering the same message after a failed pool insert must not duplicate it: + /// `(block, tx_index)` is the key, so a re-insert overwrites the prior entry. + #[test] + fn insert_batch_is_idempotent_by_position() { + let pending = PendingMessages::default(); + + pending.insert_batch(&[msg(3, 0), msg(3, 1)]); + // Next tick re-gathers the unprocessed tail alongside fresh messages. + pending.insert_batch(&[msg(3, 1), msg(4, 0)]); + + let positions: Vec<_> = pending.snapshot().iter().map(|m| (m.block, m.tx_index)).collect(); + assert_eq!( + positions, + vec![(3, 0), (3, 1), (4, 0)], + "re-gathered (3, 1) should not duplicate" + ); + } + + /// The computed L2 hash and the L1 origin hash are surfaced on each entry so a + /// consumer can correlate a pending message with the pool/block once it lands. + #[test] + fn pending_message_exposes_l1_and_l2_hashes() { + let pending = PendingMessages::default(); + let mut m = msg(1, 0); + m.l1_tx_hash = [0xab; 32]; + let expected_l2 = m.tx.calculate_hash(); + + pending.insert_batch(&[m]); + + let entry = pending.snapshot().pop().expect("one pending entry"); + assert_eq!(entry.l1_tx_hash, [0xab; 32]); + assert_eq!(entry.l2_tx_hash, expected_l2); + } } From 5bccdead29ea6cc7c940a636d58af5cad7d3319f Mon Sep 17 00:00:00 2001 From: Ammar Arif Date: Mon, 25 May 2026 23:42:21 -0500 Subject: [PATCH 2/3] chore(messaging): fix clippy lints in test code Drop a redundant clone on a Copy `Address`, factor a complex test-helper return type into a `StreamHarness` alias, and inline `format!` args. These targets aren't linted by the `-p katana` CI clippy, so they had accumulated unnoticed. Co-Authored-By: Claude Opus 4.7 --- .../messaging/src/stream/collector/ethereum.rs | 2 +- crates/messaging/src/stream/mod.rs | 17 ++++++++--------- crates/messaging/tests/e2e.rs | 2 +- crates/messaging/tests/eth_settlement.rs | 4 ++-- 4 files changed, 12 insertions(+), 13 deletions(-) diff --git a/crates/messaging/src/stream/collector/ethereum.rs b/crates/messaging/src/stream/collector/ethereum.rs index 6099a6f6f..f4345160f 100644 --- a/crates/messaging/src/stream/collector/ethereum.rs +++ b/crates/messaging/src/stream/collector/ethereum.rs @@ -268,7 +268,7 @@ mod tests { let chain_id = ChainId::Named(NamedChainId::Goerli); let message_hash = compute_l1_to_l2_message_hash( - from_address.clone(), + from_address, ContractAddress(to_address), selector, &payload, diff --git a/crates/messaging/src/stream/mod.rs b/crates/messaging/src/stream/mod.rs index a6a77777d..74afb740f 100644 --- a/crates/messaging/src/stream/mod.rs +++ b/crates/messaging/src/stream/mod.rs @@ -411,18 +411,17 @@ mod tests { OrderedMessage { block, tx_index, l1_tx_hash: [0u8; 32], tx: stub_tx() } } - /// Build a stream wired to a fresh `MockCollector` + `ManualTrigger`. - /// Returns the boxed stream alongside handles for queueing mock responses - /// and firing trigger ticks. - fn build( - from_block: u64, - from_tx_index: u64, - confirmation_depth: u64, - ) -> ( + /// The boxed stream plus the handles needed to drive it in tests. + type StreamHarness = ( Pin, ManualTrigger>>>, Arc, ManualTriggerHandle, - ) { + ); + + /// Build a stream wired to a fresh `MockCollector` + `ManualTrigger`. + /// Returns the boxed stream alongside handles for queueing mock responses + /// and firing trigger ticks. + fn build(from_block: u64, from_tx_index: u64, confirmation_depth: u64) -> StreamHarness { let collector = Arc::new(MockCollector::new()); let (trigger, handle) = ManualTrigger::new(); let stream = Box::pin(MessageStream::with_cursor( diff --git a/crates/messaging/tests/e2e.rs b/crates/messaging/tests/e2e.rs index 1b2eb222b..530e3c1be 100644 --- a/crates/messaging/tests/e2e.rs +++ b/crates/messaging/tests/e2e.rs @@ -65,7 +65,7 @@ async fn test_messaging() { let messaging_config = MessagingConfig { settlement: katana_messaging::SettlementChainConfig::Ethereum { - rpc_url: Url::parse(&format!("http://localhost:{}", port)).unwrap(), + rpc_url: Url::parse(&format!("http://localhost:{port}")).unwrap(), contract_address: *core_contract.address(), }, interval: 2, diff --git a/crates/messaging/tests/eth_settlement.rs b/crates/messaging/tests/eth_settlement.rs index 976db861e..cab06e569 100644 --- a/crates/messaging/tests/eth_settlement.rs +++ b/crates/messaging/tests/eth_settlement.rs @@ -50,7 +50,7 @@ async fn collects_single_message_from_anvil() { let l1_test_contract = Contract1::deploy(&l1_provider, *core_contract.address()).await.unwrap(); let settlement = SettlementChainConfig::Ethereum { - rpc_url: Url::parse(&format!("http://localhost:{}", port)).unwrap(), + rpc_url: Url::parse(&format!("http://localhost:{port}")).unwrap(), contract_address: *core_contract.address(), }; @@ -138,7 +138,7 @@ async fn collects_multiple_messages_in_same_block() { let l1_test_contract = Contract1::deploy(&l1_provider, *core_contract.address()).await.unwrap(); let settlement = SettlementChainConfig::Ethereum { - rpc_url: Url::parse(&format!("http://localhost:{}", port)).unwrap(), + rpc_url: Url::parse(&format!("http://localhost:{port}")).unwrap(), contract_address: *core_contract.address(), }; From e78f7efb86f965a5b1610abc842830c3f7a3385e Mon Sep 17 00:00:00 2001 From: Ammar Arif Date: Tue, 26 May 2026 11:17:20 -0500 Subject: [PATCH 3/3] test(messaging): cover pending-message drain-loop wiring Extract the per-batch gather/insert/commit logic from the spawned task in `start()` into a `process_batch` async fn (behavior-preserving), so the buffer wiring is unit-testable without a real settlement chain. Add tests for the three paths the buffer cares about: pool accepts all (pending clears), pool rejects (messages stay pending), and partial failure (only the un-pooled tail stays pending). Co-Authored-By: Claude Opus 4.7 --- crates/messaging/src/service.rs | 272 ++++++++++++++++++++++++-------- 1 file changed, 205 insertions(+), 67 deletions(-) diff --git a/crates/messaging/src/service.rs b/crates/messaging/src/service.rs index ce25e2132..a85817b04 100644 --- a/crates/messaging/src/service.rs +++ b/crates/messaging/src/service.rs @@ -158,72 +158,9 @@ where Some(MessagingOutcome { settlement_block, messages }) => { let total_messages = messages.len(); - let mut inserted: usize = 0; - - // Mark the whole batch as pending the moment it's gathered. - // Each entry is removed below once its pool insert succeeds; - // any left over after an error-break stays pending and is - // reconciled (overwritten by key) on the next gather. - pending_for_task.insert_batch(&messages); - - for msg in messages { - let hash = msg.tx.calculate_hash(); - info!(target: LOG_TARGET, tx_hash = %format!("{:#x}", hash), "L1Handler transaction added to the pool."); - - let pool_tx = ExecutableTxWithHash { hash, transaction: msg.tx.into() }; - let insert_result = pool.add_transaction(pool_tx).await; - - match insert_result { - Ok(_) => { - inserted += 1; - - // Accepted by the pool: no longer pending. - pending_for_task.remove(msg.block, msg.tx_index); - - // Atomically persist the L1->L2 index entry and the - // checkpoint in a single DB transaction. If either - // write or the commit fails, NEITHER is persisted — - // on restart we'll re-gather and re-attempt. Splitting - // these previously meant a failed index write paired - // with a successful checkpoint write would silently - // drop the L1->L2 mapping forever. - if let Err(error) = commit_message( - &provider, - &msg.l1_tx_hash, - hash, - msg.block, - msg.tx_index, - ) { - warn!( - target: LOG_TARGET, - %error, - block = msg.block, - tx_index = msg.tx_index, - tx_hash = %format!("{hash:#x}"), - "Failed to commit messaging state; aborting batch, will retry on next gather.", - ); - break; - } - } - Err(e) => { - warn!( - target: LOG_TARGET, - error = %e, - block = msg.block, - tx_index = msg.tx_index, - tx_hash = %format!("{hash:#x}"), - "Failed to add L1Handler transaction to pool; will retry on next gather.", - ); - - // Stop processing this batch. The stream's cursor - // was already advanced past the current gather range; - // the retry for this message will rely on the pool's - // hash-level deduplication of successful inserts and - // re-gather on the next tick. - break; - } - } - } + let inserted = + process_batch(messages, &pool, &provider, &pending_for_task) + .await; if inserted > 0 { info!( @@ -290,6 +227,88 @@ impl Clone for MessagingService { } } +/// Process one gathered batch: mark every message pending, then insert each into +/// the pool in order, removing it from the pending set and atomically persisting +/// its L1->L2 index entry + checkpoint the moment the pool accepts it. +/// +/// Returns the number of messages inserted. On the first pool insert or commit +/// failure the batch is abandoned: the offending message and the untouched tail +/// stay pending and are re-gathered (and overwritten by position) on the next +/// tick. The stream cursor only advances for committed messages. +async fn process_batch( + messages: Vec, + pool: &Pl, + provider: &P, + pending: &PendingMessages, +) -> usize +where + P: ProviderFactory, +

::ProviderMut: + MessagingCheckpointProvider + MessagingL1ToL2IndexWriter + MutableProvider, + Pl: TransactionPool, +{ + let mut inserted: usize = 0; + + // Mark the whole batch as pending the moment it's gathered. Each entry is + // removed below once its pool insert succeeds; any left over after an + // error-break stays pending and is reconciled (overwritten by key) on the + // next gather. + pending.insert_batch(&messages); + + for msg in messages { + let hash = msg.tx.calculate_hash(); + info!(target: LOG_TARGET, tx_hash = %format!("{hash:#x}"), "L1Handler transaction added to the pool."); + + let pool_tx = ExecutableTxWithHash { hash, transaction: msg.tx.into() }; + + match pool.add_transaction(pool_tx).await { + Ok(_) => { + inserted += 1; + + // Accepted by the pool: no longer pending. + pending.remove(msg.block, msg.tx_index); + + // Atomically persist the L1->L2 index entry and the checkpoint in a + // single DB transaction. If either write or the commit fails, NEITHER + // is persisted — on restart we'll re-gather and re-attempt. Splitting + // these previously meant a failed index write paired with a successful + // checkpoint write would silently drop the L1->L2 mapping forever. + if let Err(error) = + commit_message(provider, &msg.l1_tx_hash, hash, msg.block, msg.tx_index) + { + warn!( + target: LOG_TARGET, + %error, + block = msg.block, + tx_index = msg.tx_index, + tx_hash = %format!("{hash:#x}"), + "Failed to commit messaging state; aborting batch, will retry on next gather.", + ); + break; + } + } + Err(e) => { + warn!( + target: LOG_TARGET, + error = %e, + block = msg.block, + tx_index = msg.tx_index, + tx_hash = %format!("{hash:#x}"), + "Failed to add L1Handler transaction to pool; will retry on next gather.", + ); + + // Stop processing this batch. The stream's cursor was already advanced + // past the current gather range; the retry for this message will rely + // on the pool's hash-level deduplication of successful inserts and + // re-gather on the next tick. + break; + } + } + } + + inserted +} + /// Atomically record the L1->L2 mapping and advance the checkpoint inside a single /// DB transaction. Returns an error if any of the staged writes or the commit fail. fn commit_message

( @@ -420,12 +439,62 @@ impl MessagingServiceHandle { #[cfg(test)] mod tests { - use katana_primitives::Felt; + use std::sync::atomic::{AtomicUsize, Ordering}; + + use katana_pool::api::validation::{ + InvalidTransactionError, ValidationOutcome, ValidationResult, Validator, + }; + use katana_pool::ordering::FiFo; + use katana_pool::pool::Pool; + use katana_pool::validation::NoopValidator; + use katana_primitives::{ContractAddress, Felt}; use katana_provider::api::messaging::MessagingL1ToL2IndexProvider; use katana_provider::DbProviderFactory; use super::*; + /// A validator that rejects every transaction, so `pool.add_transaction` + /// always returns `Err` — used to drive `process_batch`'s pool-failure path. + #[derive(Debug)] + struct RejectingValidator; + + impl Validator for RejectingValidator { + type Transaction = ExecutableTxWithHash; + + async fn validate(&self, tx: Self::Transaction) -> ValidationResult { + Ok(ValidationOutcome::Invalid { + tx, + error: InvalidTransactionError::NonAccount { address: ContractAddress::default() }, + }) + } + } + + /// A validator that accepts the first `accept` transactions and rejects the + /// rest — used to drive the partial-failure path where some of a batch lands + /// in the pool and the remainder stays pending. + #[derive(Debug)] + struct AcceptThenReject { + accept: usize, + seen: AtomicUsize, + } + + impl Validator for AcceptThenReject { + type Transaction = ExecutableTxWithHash; + + async fn validate(&self, tx: Self::Transaction) -> ValidationResult { + if self.seen.fetch_add(1, Ordering::SeqCst) < self.accept { + Ok(ValidationOutcome::Valid(tx)) + } else { + Ok(ValidationOutcome::Invalid { + tx, + error: InvalidTransactionError::NonAccount { + address: ContractAddress::default(), + }, + }) + } + } + } + /// Builds a stub `OrderedMessage` whose tx internals don't matter — the buffer /// only keys on `(block, tx_index)`. `nonce` is varied so distinct positions /// produce distinct computed L2 hashes, letting tests tell entries apart. @@ -619,4 +688,73 @@ mod tests { assert_eq!(entry.l1_tx_hash, [0xab; 32]); assert_eq!(entry.l2_tx_hash, expected_l2); } + + /// The happy path: every message in the batch is accepted by the pool, so the + /// pending set ends empty, both txs land in the pool, and the checkpoint + /// advances to the last committed message. + #[tokio::test] + async fn process_batch_inserts_all_and_clears_pending_when_pool_accepts() { + let pool = Pool::new(NoopValidator::new(), FiFo::new()); + let provider = DbProviderFactory::new_in_memory(); + let pending = PendingMessages::default(); + + let inserted = process_batch(vec![msg(5, 0), msg(5, 1)], &pool, &provider, &pending).await; + + assert_eq!(inserted, 2, "both messages should be accepted by the pool"); + assert!(pending.is_empty(), "accepted messages must leave the pending set"); + assert_eq!(pool.size(), 2, "both L1Handler txs should be in the pool"); + + let db_tx = provider.provider_mut(); + let cp = db_tx.messaging_checkpoint(CHECKPOINT_ID).unwrap().expect("checkpoint"); + db_tx.commit().unwrap(); + assert_eq!((cp.block, cp.tx_index), (5, 1), "checkpoint advances to the last commit"); + } + + /// The feature's reason for existing: when the pool rejects, the message was + /// picked up from settlement but never pooled, so it must remain queryable as + /// pending. The first reject aborts the batch, leaving the untouched tail + /// pending too, and nothing is committed. + #[tokio::test] + async fn process_batch_keeps_messages_pending_when_pool_rejects() { + let pool = Pool::new(RejectingValidator, FiFo::new()); + let provider = DbProviderFactory::new_in_memory(); + let pending = PendingMessages::default(); + + let inserted = process_batch(vec![msg(9, 0), msg(9, 1)], &pool, &provider, &pending).await; + + assert_eq!(inserted, 0, "a rejecting pool inserts nothing"); + assert_eq!(pool.size(), 0); + + let positions: Vec<_> = pending.snapshot().iter().map(|m| (m.block, m.tx_index)).collect(); + assert_eq!(positions, vec![(9, 0), (9, 1)], "rejected messages stay pending"); + + let db_tx = provider.provider_mut(); + let cp = db_tx.messaging_checkpoint(CHECKPOINT_ID).unwrap(); + db_tx.commit().unwrap(); + assert!(cp.is_none(), "nothing is committed when nothing is pooled"); + } + + /// Partial failure: the pool accepts the first message and rejects the second. + /// The accepted one leaves the pending set and advances the checkpoint; the + /// rejected one stays pending, so a query surfaces exactly the un-pooled tail. + #[tokio::test] + async fn process_batch_keeps_only_the_unpooled_tail_pending_on_partial_failure() { + let pool = + Pool::new(AcceptThenReject { accept: 1, seen: AtomicUsize::new(0) }, FiFo::new()); + let provider = DbProviderFactory::new_in_memory(); + let pending = PendingMessages::default(); + + let inserted = process_batch(vec![msg(3, 0), msg(3, 1)], &pool, &provider, &pending).await; + + assert_eq!(inserted, 1, "only the first message is accepted"); + assert_eq!(pool.size(), 1); + + let positions: Vec<_> = pending.snapshot().iter().map(|m| (m.block, m.tx_index)).collect(); + assert_eq!(positions, vec![(3, 1)], "only the un-pooled message stays pending"); + + let db_tx = provider.provider_mut(); + let cp = db_tx.messaging_checkpoint(CHECKPOINT_ID).unwrap().expect("checkpoint"); + db_tx.commit().unwrap(); + assert_eq!((cp.block, cp.tx_index), (3, 0), "checkpoint stops at the one commit"); + } }