From 393aa1ce271c8645146049b6de1f5d26792408d1 Mon Sep 17 00:00:00 2001 From: Ivan Shumkov Date: Tue, 10 Nov 2026 10:01:20 +0700 Subject: [PATCH 1/2] fix(dapi): stream every instance of a re-signed asset unlock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rs-dapi dedupes the raw transactions it streams to a subscription by txid. From protocol version 15 Platform emits asset unlocks with payload version 2, whose txid leaves out the request height, quorum hash and signature, so every instance Platform re-signs after an expiry shares one txid. Core relays each instance, but a subscriber that received an earlier one never received the refreshed one, nor the one that was mined, and kept a stale signature (review of #5634, finding 5). A transaction is now marked delivered by txid, still used by instant locks and merkle blocks, and, when its instance hash (rust-dashcore's Transaction::instance_hash, the hash of the full serialization) differs from its txid, also by that instance hash. Raw transactions are deduped by instance on the live, mempool and history paths. The instance set fills only for stable-txid asset unlocks, so per-stream memory for other transactions is unchanged. Tests: - should_stream_every_instance_of_a_re_signed_asset_unlock: two instances of one version 2 asset unlock, then the first again, through flush_transaction_pending. Before the fix only the first instance reached the subscriber; after it both do, once each, and the txid is still found as delivered. - should_not_stream_again_the_asset_unlock_instances_a_replay_delivered: instances a history replay delivered are skipped live; a new instance is not. Test would have caught this in CI: ✖ before fix, ✔ after. Co-Authored-By: Claude Opus 5.5 --- .../streaming_service/transaction_stream.rs | 223 +++++++++++++++--- 1 file changed, 193 insertions(+), 30 deletions(-) diff --git a/packages/rs-dapi/src/services/streaming_service/transaction_stream.rs b/packages/rs-dapi/src/services/streaming_service/transaction_stream.rs index 0b05ab5dbf2..51fa2027364 100644 --- a/packages/rs-dapi/src/services/streaming_service/transaction_stream.rs +++ b/packages/rs-dapi/src/services/streaming_service/transaction_stream.rs @@ -11,6 +11,7 @@ use dapi_grpc::core::v0::{ }; use dapi_grpc::tonic::{Request, Response, Status}; use dashcore_rpc::dashcore::consensus::Decodable as _; +use dashcore_rpc::dashcore::consensus::encode::deserialize; use dashcore_rpc::dashcore::{Block, InstantLock, Transaction}; use futures::TryFutureExt; use tokio::sync::{Mutex as AsyncMutex, OwnedSemaphorePermit, mpsc, watch}; @@ -41,6 +42,7 @@ type TxResponseSender = mpsc::Sender; type TxResponseStream = ReceiverStream; type TxResponse = Response; type DeliveredTxSet = Arc>; +type DeliveredTxInstanceSet = Arc>; type DeliveredBlockSet = Arc>; type DeliveredInstantLockSet = Arc>; type GateSender = watch::Sender; @@ -87,6 +89,23 @@ impl BoundedDeliveredIds { } } +/// Marks `transaction` delivered: its txid in `delivered_txs` and, when the txid does not tell +/// its instances apart, the hash of its full serialization in `delivered_tx_instances`. Returns +/// false if this instance was already delivered. +fn mark_transaction_instance_delivered( + delivered_txs: &mut BoundedDeliveredIds, + delivered_tx_instances: &mut BoundedDeliveredIds, + transaction: &Transaction, +) -> bool { + let txid = transaction.txid(); + let txid_is_new = delivered_txs.mark_delivered(txid.to_byte_array().to_vec()); + let instance_hash = transaction.instance_hash(); + if instance_hash == txid { + return txid_is_new; + } + delivered_tx_instances.mark_delivered(instance_hash.to_byte_array().to_vec()) +} + fn validate_transaction_from_block_shape( from_block: &dapi_grpc::core::v0::transactions_with_proofs_request::FromBlock, ) -> Result<(), Status> { @@ -106,6 +125,7 @@ fn validate_transaction_from_block_shape( #[derive(Clone)] struct TransactionStreamState { delivered_txs: DeliveredTxSet, + delivered_tx_instances: DeliveredTxInstanceSet, delivered_blocks: DeliveredBlockSet, delivered_instant_locks: DeliveredInstantLockSet, gate_sender: GateSender, @@ -122,6 +142,9 @@ impl TransactionStreamState { delivered_txs: Arc::new(AsyncMutex::new(BoundedDeliveredIds::new( MAX_DELIVERED_IDS_PER_STREAM, ))), + delivered_tx_instances: Arc::new(AsyncMutex::new(BoundedDeliveredIds::new( + MAX_DELIVERED_IDS_PER_STREAM, + ))), delivered_blocks: Arc::new(AsyncMutex::new(BoundedDeliveredIds::new( MAX_DELIVERED_IDS_PER_STREAM, ))), @@ -195,27 +218,44 @@ impl TransactionStreamState { } } - /// Marks a transaction as delivered. Returns false if it was already delivered. - async fn mark_transaction_delivered(&self, txid: &[u8]) -> bool { + /// Marks an instance of a transaction as delivered. Returns false if this instance was + /// already delivered. + /// + /// The txid tells instances apart for every transaction except an asset unlock with payload + /// version 2 or higher, whose txid leaves out the quorum signing info: Platform re-signs such + /// an unlock when it expires, every instance up to the mined one shares the txid, and Core + /// relays each. Those instances are told apart by the hash of their full serialization. The + /// txid is kept either way, for the instant locks and merkle blocks that name a transaction + /// by it. + async fn mark_transaction_delivered(&self, transaction: &Transaction) -> bool { tracing::trace!( - txid = txid_to_hex(txid), + txid = %transaction.txid(), "transaction_stream=mark_transaction_delivered" ); - let mut guard = self.delivered_txs.lock().await; - guard.mark_delivered(txid.to_vec()) + let mut delivered_txs = self.delivered_txs.lock().await; + let mut delivered_tx_instances = self.delivered_tx_instances.lock().await; + mark_transaction_instance_delivered( + &mut delivered_txs, + &mut delivered_tx_instances, + transaction, + ) } - async fn mark_transactions_delivered(&self, txids: I) - where - I: IntoIterator>, - { - let mut guard = self.delivered_txs.lock().await; - for txid in txids { + /// Marks instances of transactions as delivered, as [`Self::mark_transaction_delivered`] + /// does. + async fn mark_transactions_delivered(&self, transactions: &[Transaction]) { + let mut delivered_txs = self.delivered_txs.lock().await; + let mut delivered_tx_instances = self.delivered_tx_instances.lock().await; + for transaction in transactions { tracing::trace!( - txid = txid_to_hex(&txid), + txid = %transaction.txid(), "transaction_stream=mark_transaction_delivered" ); - guard.mark_delivered(txid); + mark_transaction_instance_delivered( + &mut delivered_txs, + &mut delivered_tx_instances, + transaction, + ); } } @@ -517,15 +557,13 @@ impl StreamingServiceImpl { ) -> bool { let maybe_response = match event { StreamingEvent::CoreRawTransaction { data } => { - let (Some(txid_bytes), Some(txid_hex)) = ( - super::txid_bytes_from_bytes(&data), - super::txid_hex_from_bytes(&data), - ) else { + let Ok(transaction) = deserialize::(&data) else { tracing::debug!("transactions_with_proofs=transaction_no_txid"); return true; }; + let txid_hex = transaction.txid().to_string(); - let already_delivered = !state.mark_transaction_delivered(&txid_bytes).await; + let already_delivered = !state.mark_transaction_delivered(&transaction).await; if already_delivered { trace!( subscriber_id, @@ -1116,9 +1154,8 @@ impl StreamingServiceImpl { .ok_or_else(|| { Status::resource_exhausted("mempool snapshot exceeds the per-stream byte limit") })?; - let txid_bytes = transaction.txid().to_byte_array(); - if !state.mark_transaction_delivered(&txid_bytes).await { + if !state.mark_transaction_delivered(&transaction).await { trace!( txid = %transaction.txid(), "transactions_with_proofs=skip_duplicate_mempool_transaction" @@ -1219,7 +1256,7 @@ impl StreamingServiceImpl { >::as_ref(&hash).to_vec(); let mut matching: Vec> = Vec::new(); - let mut matching_hashes: Vec> = Vec::new(); + let mut matching_transactions: Vec = Vec::new(); let mut match_flags: Vec = Vec::with_capacity(txs_bytes.len()); for tx_bytes in txs_bytes.iter() { @@ -1257,8 +1294,8 @@ impl StreamingServiceImpl { match_flags.push(matches_for_merkle); // Only send raw transactions when they matched the bloom filter if filter_matched { - if let Some(hash_bytes) = super::txid_bytes_from_bytes(tx_bytes) { - matching_hashes.push(hash_bytes); + if let Ok(transaction) = deserialize::(tx_bytes.as_slice()) { + matching_transactions.push(transaction); } matching.push(tx_bytes.clone()); } @@ -1266,7 +1303,9 @@ impl StreamingServiceImpl { if !matching.is_empty() { if let Some(state) = state.as_ref() { - state.mark_transactions_delivered(matching_hashes).await; + state + .mark_transactions_delivered(&matching_transactions) + .await; } let raw_transactions = RawTransactions { @@ -1408,11 +1447,19 @@ fn txid_to_hex(txid: &[u8]) -> String { mod tests { use super::*; use dashcore_rpc::dashcore::{ - Block, BlockHash, CompactTarget, OutPoint, ScriptBuf, Transaction as CoreTx, TxIn, - TxMerkleNode, TxOut, Txid, Witness, + Block, BlockHash, CompactTarget, OutPoint, PubkeyHash, QuorumHash, ScriptBuf, + Transaction as CoreTx, TxIn, TxMerkleNode, TxOut, Txid, Witness, block::{Header as BlockHeader, Version}, + bls_sig_utils::BLSSignature, consensus::encode::{deserialize, serialize}, merkle_tree::MerkleBlock, + transaction::special_transaction::{ + TransactionPayload, + asset_unlock::{ + qualified_asset_unlock::AssetUnlockPayload, request_info::AssetUnlockRequestInfo, + unqualified_asset_unlock::AssetUnlockBasePayload, + }, + }, }; use std::time::Duration; use tokio::time::{sleep, timeout}; @@ -1495,10 +1542,14 @@ mod tests { #[tokio::test] async fn should_dedupe_transactions_blocks_and_instant_locks() { let state = TransactionStreamState::new(); - let txid = vec![1u8; 32]; - assert!(state.mark_transaction_delivered(&txid).await); - assert!(!state.mark_transaction_delivered(&txid).await); - assert!(state.has_transaction_been_delivered(&txid).await); + let transaction = sample_tx(1); + assert!(state.mark_transaction_delivered(&transaction).await); + assert!(!state.mark_transaction_delivered(&transaction).await); + assert!( + state + .has_transaction_been_delivered(&transaction.txid().to_byte_array()) + .await + ); let block_hash = vec![2u8; 32]; assert!(state.mark_block_delivered(&block_hash).await); @@ -1813,4 +1864,116 @@ mod tests { ); assert_eq!(indexes, vec![0, 1]); } + + /// The instance of withdrawal 101's version 2 asset unlock signed at `request_height` by a + /// quorum named after it. Its txid leaves out the request height, quorum hash and signature. + fn version_2_asset_unlock(request_height: u32) -> CoreTx { + let quorum = request_height.to_le_bytes()[0]; + CoreTx { + version: 3, + lock_time: 0, + input: vec![], + output: vec![TxOut { + value: 100_000_000, + script_pubkey: ScriptBuf::new_p2pkh(&PubkeyHash::from_byte_array([0x11; 20])), + }], + special_transaction_payload: Some(TransactionPayload::AssetUnlockPayloadType( + AssetUnlockPayload { + base: AssetUnlockBasePayload { + version: 2, + index: 101, + fee: 70_000, + }, + request_info: AssetUnlockRequestInfo { + request_height, + quorum_hash: QuorumHash::from_byte_array([quorum; 32]), + }, + quorum_sig: BLSSignature::from([quorum; 96]), + }, + )), + } + } + + /// Platform re-signs a version 2 asset unlock when it expires, and every instance it signs + /// shares one txid. Core relays each instance, so a subscriber that got an earlier one must + /// still get each later one, up to the instance Core mines: the earlier signature can never + /// be mined once its height is past. The same instance again is still a duplicate. + #[tokio::test] + async fn should_stream_every_instance_of_a_re_signed_asset_unlock() { + let first = version_2_asset_unlock(500); + let re_signed = version_2_asset_unlock(549); + assert_eq!(first.txid(), re_signed.txid(), "the instances share a txid"); + assert_ne!(first.instance_hash(), re_signed.instance_hash()); + + let state = TransactionStreamState::new(); + let (tx_sender, mut rx) = mpsc::channel(8); + let mut pending: Vec<(StreamingEvent, String)> = [&first, &re_signed, &first] + .into_iter() + .map(|instance| { + ( + StreamingEvent::CoreRawTransaction { + data: serialize(instance), + }, + "tx_handle".to_string(), + ) + }) + .collect(); + + assert!( + StreamingServiceImpl::flush_transaction_pending( + &FilterType::CoreAllTxs, + "subscriber", + &tx_sender, + &state, + &mut pending, + ) + .await + ); + drop(tx_sender); + + let mut streamed = Vec::new(); + while let Some(response) = rx.recv().await { + match response.expect("status ok").responses { + Some(Responses::RawTransactions(raw)) => streamed.extend(raw.transactions), + other => panic!("unexpected response: {:?}", other), + } + } + let streamed_instances: Vec = streamed + .iter() + .map(|bytes| { + deserialize::(bytes) + .expect("a transaction") + .instance_hash() + }) + .collect(); + assert_eq!( + streamed_instances, + vec![first.instance_hash(), re_signed.instance_hash()], + "each instance is streamed once, in order" + ); + assert!( + state + .has_transaction_been_delivered(&first.txid().to_byte_array()) + .await, + "an instant lock or a merkle block still finds the asset unlock by its txid" + ); + } + + /// The instances a history replay delivered are not streamed again when Core reports them + /// live, while an instance the replay did not deliver still is. + #[tokio::test] + async fn should_not_stream_again_the_asset_unlock_instances_a_replay_delivered() { + let state = TransactionStreamState::new(); + let replayed = [version_2_asset_unlock(500), version_2_asset_unlock(549)]; + state.mark_transactions_delivered(&replayed).await; + + for instance in &replayed { + assert!(!state.mark_transaction_delivered(instance).await); + } + assert!( + state + .mark_transaction_delivered(&version_2_asset_unlock(598)) + .await + ); + } } From 45e4c9d1c43e84f2b32546b08a06881683b481af Mon Sep 17 00:00:00 2001 From: Ivan Shumkov Date: Tue, 10 Nov 2026 11:30:13 +0700 Subject: [PATCH 2/2] refactor(dapi): keep only delivery ids of replayed transactions The history replay kept every matching decoded transaction until the block was sent, only to derive the ids it marks delivered. It now decodes each transaction once and keeps its txid and, for a stable-txid asset unlock, its instance hash. The unused txid_bytes_from_bytes helper is removed, and the live path's deserialization failure is logged as transaction_deserialize_failed. Addresses the PastaClaw review of #5832. No behaviour change; the existing tests, including both asset unlock instance tests, pass. Co-Authored-By: Claude Opus 5.5 --- .../src/services/streaming_service/mod.rs | 9 -- .../streaming_service/transaction_stream.rs | 106 +++++++++--------- 2 files changed, 56 insertions(+), 59 deletions(-) diff --git a/packages/rs-dapi/src/services/streaming_service/mod.rs b/packages/rs-dapi/src/services/streaming_service/mod.rs index 3585a2f1a3d..0429855ad72 100644 --- a/packages/rs-dapi/src/services/streaming_service/mod.rs +++ b/packages/rs-dapi/src/services/streaming_service/mod.rs @@ -375,15 +375,6 @@ pub(crate) fn txid_hex_from_bytes(bytes: &[u8]) -> Option { .map(|tx| tx.txid().to_string()) } -/// Decode transaction bytes and return the txid in raw byte form. -pub(crate) fn txid_bytes_from_bytes(bytes: &[u8]) -> Option> { - use dashcore_rpc::dashcore::Transaction as CoreTx; - use dashcore_rpc::dashcore::consensus::encode::deserialize; - - deserialize::(bytes) - .ok() - .map(|tx| tx.txid().to_byte_array().to_vec()) -} /// Decode block bytes and return the block hash in hex and as printable string. pub(crate) fn block_hash_from_block_bytes(bytes: &[u8]) -> Option<([u8; 32], String)> { deserialize::(bytes) diff --git a/packages/rs-dapi/src/services/streaming_service/transaction_stream.rs b/packages/rs-dapi/src/services/streaming_service/transaction_stream.rs index 51fa2027364..da14a4735cd 100644 --- a/packages/rs-dapi/src/services/streaming_service/transaction_stream.rs +++ b/packages/rs-dapi/src/services/streaming_service/transaction_stream.rs @@ -89,21 +89,40 @@ impl BoundedDeliveredIds { } } -/// Marks `transaction` delivered: its txid in `delivered_txs` and, when the txid does not tell -/// its instances apart, the hash of its full serialization in `delivered_tx_instances`. Returns -/// false if this instance was already delivered. +/// What a stream records about a transaction it delivered: the txid and, when the txid does not +/// tell the transaction's instances apart, the hash of its full serialization. +struct TransactionDeliveryIds { + txid: [u8; 32], + instance_hash: Option<[u8; 32]>, +} + +impl TransactionDeliveryIds { + fn of(transaction: &Transaction) -> Self { + let txid = transaction.txid(); + let instance_hash = transaction.instance_hash(); + Self { + txid: txid.to_byte_array(), + instance_hash: (instance_hash != txid).then(|| instance_hash.to_byte_array()), + } + } +} + +/// Marks a transaction delivered: its txid in `delivered_txs` and its instance hash, if it has +/// one, in `delivered_tx_instances`. Returns false if this instance was already delivered. fn mark_transaction_instance_delivered( delivered_txs: &mut BoundedDeliveredIds, delivered_tx_instances: &mut BoundedDeliveredIds, - transaction: &Transaction, + ids: &TransactionDeliveryIds, ) -> bool { - let txid = transaction.txid(); - let txid_is_new = delivered_txs.mark_delivered(txid.to_byte_array().to_vec()); - let instance_hash = transaction.instance_hash(); - if instance_hash == txid { - return txid_is_new; + tracing::trace!( + txid = txid_to_hex(&ids.txid), + "transaction_stream=mark_transaction_delivered" + ); + let txid_is_new = delivered_txs.mark_delivered(ids.txid.to_vec()); + match ids.instance_hash { + Some(instance_hash) => delivered_tx_instances.mark_delivered(instance_hash.to_vec()), + None => txid_is_new, } - delivered_tx_instances.mark_delivered(instance_hash.to_byte_array().to_vec()) } fn validate_transaction_from_block_shape( @@ -228,33 +247,25 @@ impl TransactionStreamState { /// txid is kept either way, for the instant locks and merkle blocks that name a transaction /// by it. async fn mark_transaction_delivered(&self, transaction: &Transaction) -> bool { - tracing::trace!( - txid = %transaction.txid(), - "transaction_stream=mark_transaction_delivered" - ); let mut delivered_txs = self.delivered_txs.lock().await; let mut delivered_tx_instances = self.delivered_tx_instances.lock().await; mark_transaction_instance_delivered( &mut delivered_txs, &mut delivered_tx_instances, - transaction, + &TransactionDeliveryIds::of(transaction), ) } /// Marks instances of transactions as delivered, as [`Self::mark_transaction_delivered`] /// does. - async fn mark_transactions_delivered(&self, transactions: &[Transaction]) { + async fn mark_transactions_delivered(&self, transactions: &[TransactionDeliveryIds]) { let mut delivered_txs = self.delivered_txs.lock().await; let mut delivered_tx_instances = self.delivered_tx_instances.lock().await; - for transaction in transactions { - tracing::trace!( - txid = %transaction.txid(), - "transaction_stream=mark_transaction_delivered" - ); + for ids in transactions { mark_transaction_instance_delivered( &mut delivered_txs, &mut delivered_tx_instances, - transaction, + ids, ); } } @@ -558,7 +569,7 @@ impl StreamingServiceImpl { let maybe_response = match event { StreamingEvent::CoreRawTransaction { data } => { let Ok(transaction) = deserialize::(&data) else { - tracing::debug!("transactions_with_proofs=transaction_no_txid"); + tracing::debug!("transactions_with_proofs=transaction_deserialize_failed"); return true; }; let txid_hex = transaction.txid().to_string(); @@ -1256,37 +1267,34 @@ impl StreamingServiceImpl { >::as_ref(&hash).to_vec(); let mut matching: Vec> = Vec::new(); - let mut matching_transactions: Vec = Vec::new(); + let mut matching_ids: Vec = Vec::new(); let mut match_flags: Vec = Vec::with_capacity(txs_bytes.len()); for tx_bytes in txs_bytes.iter() { + let transaction = deserialize::(tx_bytes.as_slice()); let filter_matched = match &filter { FilterType::CoreAllTxs => true, - FilterType::CoreBloomFilter(bloom, flags) => { - match deserialize::(tx_bytes.as_slice()) { - Ok(tx) => { - let matches = super::bloom::matches_transaction( - Arc::clone(bloom), - &tx, - *flags, - ); - trace!(height,matches, txid = %tx.txid(), "transactions_with_proofs=bloom_match"); - matches - } - Err(e) => { - debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed, checking raw-bytes contains()"); - let guard = bloom.read().unwrap(); - guard.contains(tx_bytes) - } + FilterType::CoreBloomFilter(bloom, flags) => match &transaction { + Ok(tx) => { + let matches = + super::bloom::matches_transaction(Arc::clone(bloom), tx, *flags); + trace!(height,matches, txid = %tx.txid(), "transactions_with_proofs=bloom_match"); + matches } - } + Err(e) => { + debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed, checking raw-bytes contains()"); + let guard = bloom.read().unwrap(); + guard.contains(tx_bytes) + } + }, _ => false, }; + let ids = transaction.as_ref().ok().map(TransactionDeliveryIds::of); // Include previously delivered transactions in PMT regardless of bloom match let mut matches_for_merkle = filter_matched; if let Some(state) = state.as_ref() - && let Some(hash_bytes) = super::txid_bytes_from_bytes(tx_bytes) - && state.has_transaction_been_delivered(&hash_bytes).await + && let Some(ids) = &ids + && state.has_transaction_been_delivered(&ids.txid).await { matches_for_merkle = true; } @@ -1294,18 +1302,14 @@ impl StreamingServiceImpl { match_flags.push(matches_for_merkle); // Only send raw transactions when they matched the bloom filter if filter_matched { - if let Ok(transaction) = deserialize::(tx_bytes.as_slice()) { - matching_transactions.push(transaction); - } + matching_ids.extend(ids); matching.push(tx_bytes.clone()); } } if !matching.is_empty() { if let Some(state) = state.as_ref() { - state - .mark_transactions_delivered(&matching_transactions) - .await; + state.mark_transactions_delivered(&matching_ids).await; } let raw_transactions = RawTransactions { @@ -1965,7 +1969,9 @@ mod tests { async fn should_not_stream_again_the_asset_unlock_instances_a_replay_delivered() { let state = TransactionStreamState::new(); let replayed = [version_2_asset_unlock(500), version_2_asset_unlock(549)]; - state.mark_transactions_delivered(&replayed).await; + state + .mark_transactions_delivered(&replayed.each_ref().map(TransactionDeliveryIds::of)) + .await; for instance in &replayed { assert!(!state.mark_transaction_delivered(instance).await);