Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: remove block_id from starknet_subscribeTransactionStatus #2501

Merged
merged 3 commits into from
Jan 22, 2025
Merged

fix: remove block_id from starknet_subscribeTransactionStatus #2501

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1eb1176
fix: removed block_id from starknet_subscribeTransactionStatus
vbar Jan 20, 2025
9821042
starknet_subscribeTransactionStatus doc update
vbar Jan 21, 2025
a977074
using RpcVersion::v08 to serialize subscription ID
vbar Jan 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 57 additions & 74 deletions crates/rpc/src/method/subscribe_transaction_status.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ use std::time::Duration;

use axum::async_trait;
use pathfinder_common::receipt::ExecutionStatus;
use pathfinder_common::{BlockId, BlockNumber, TransactionHash};
use pathfinder_common::{BlockNumber, TransactionHash};
use reply::transaction_status as status;
use starknet_gateway_client::GatewayApi;
use starknet_gateway_types::reply;
Expand All @@ -12,7 +12,6 @@ use tokio::time::MissedTickBehavior;

use super::REORG_SUBSCRIPTION_NAME;
use crate::context::RpcContext;
use crate::error::ApplicationError;
use crate::jsonrpc::{RpcError, RpcSubscriptionFlow, SubscriptionMessage};
use crate::Reorg;

Expand All @@ -21,15 +20,13 @@ pub struct SubscribeTransactionStatus;
#[derive(Debug, Clone, Default)]
pub struct Params {
transaction_hash: TransactionHash,
block_id: Option<BlockId>,
}

impl crate::dto::DeserializeForVersion for Params {
fn deserialize(value: crate::dto::Value) -> Result<Self, serde_json::Error> {
value.deserialize_map(|value| {
Ok(Self {
transaction_hash: value.deserialize("transaction_hash").map(TransactionHash)?,
block_id: value.deserialize_optional_serde("block_id")?,
})
})
}
Expand Down Expand Up @@ -143,49 +140,51 @@ impl RpcSubscriptionFlow for SubscribeTransactionStatus {
let mut l2_blocks = state.notifications.l2_blocks.subscribe();
let mut reorgs = state.notifications.reorgs.subscribe();
let storage = state.storage.clone();
if let Some(first_block) = params.block_id {
// Check if we have the transaction in our database, and if so, send the
// relevant transaction status updates.
let (first_block, l1_state, tx_with_receipt) =
util::task::spawn_blocking(move |_| -> Result<_, RpcError> {
let mut conn = storage.connection().map_err(RpcError::InternalError)?;
let db = conn.transaction().map_err(RpcError::InternalError)?;
let first_block = db
.block_number(first_block.try_into().map_err(|_| {
RpcError::ApplicationError(ApplicationError::CallOnPending)
})?)
.map_err(RpcError::InternalError)?;
let l1_block_number =
db.latest_l1_state().map_err(RpcError::InternalError)?;
let tx_with_receipt = db
.transaction_with_receipt(tx_hash)
.map_err(RpcError::InternalError)?;
Ok((first_block, l1_block_number, tx_with_receipt))
})
.await
.map_err(|e| RpcError::InternalError(e.into()))??;
let first_block = first_block
.ok_or_else(|| RpcError::ApplicationError(ApplicationError::BlockNotFound))?;
if let Some((_, receipt, _, block_number)) = tx_with_receipt {
// We already have the transaction in the database.
if let Some(parent) = block_number.parent() {
// This transaction was pending in the parent block.
if first_block <= parent {
if sender
.send(parent, FinalityStatus::Received, None)
.await
.is_err()
{
// Subscription closing.
break;
}
}
// Check if we have the transaction in our database, and if so, send the
// relevant transaction status updates.
let (l1_state, tx_with_receipt) =
util::task::spawn_blocking(move |_| -> Result<_, RpcError> {
let mut conn = storage.connection().map_err(RpcError::InternalError)?;
let db = conn.transaction().map_err(RpcError::InternalError)?;
let l1_block_number = db.latest_l1_state().map_err(RpcError::InternalError)?;
let tx_with_receipt = db
.transaction_with_receipt(tx_hash)
.map_err(RpcError::InternalError)?;
Ok((l1_block_number, tx_with_receipt))
})
.await
.map_err(|e| RpcError::InternalError(e.into()))??;
if let Some((_, receipt, _, block_number)) = tx_with_receipt {
// We already have the transaction in the database.
if let Some(parent) = block_number.parent() {
// This transaction was pending in the parent block.
if sender
.send(parent, FinalityStatus::Received, None)
.await
.is_err()
{
// Subscription closing.
break;
}
if first_block <= block_number {
}
if sender
.send(
block_number,
FinalityStatus::AcceptedOnL2,
Some(receipt.execution_status.clone()),
)
.await
.is_err()
{
// Subscription closing.
break;
}
if let Some(l1_state) = l1_state {
if l1_state.block_number >= block_number {
if sender
.send(
block_number,
FinalityStatus::AcceptedOnL2,
l1_state.block_number,
FinalityStatus::AcceptedOnL1,
Some(receipt.execution_status.clone()),
)
.await
Expand All @@ -195,22 +194,6 @@ impl RpcSubscriptionFlow for SubscribeTransactionStatus {
break;
}
}
if let Some(l1_state) = l1_state {
if l1_state.block_number >= block_number {
if sender
.send(
l1_state.block_number,
FinalityStatus::AcceptedOnL1,
Some(receipt.execution_status.clone()),
)
.await
.is_err()
{
// Subscription closing.
break;
}
}
}
}
}
let pending = pending_data.borrow_and_update().clone();
Expand Down Expand Up @@ -478,7 +461,7 @@ mod tests {
"finality_status": "RECEIVED",
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
serde_json::json!({
Expand All @@ -492,7 +475,7 @@ mod tests {
"execution_status": "SUCCEEDED",
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
})
]
Expand Down Expand Up @@ -579,7 +562,7 @@ mod tests {
"execution_status": "SUCCEEDED",
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
})
);
Expand Down Expand Up @@ -608,7 +591,7 @@ mod tests {
"finality_status": "RECEIVED",
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
serde_json::json!({
Expand All @@ -623,7 +606,7 @@ mod tests {
"failure_reason": "tx revert"
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
]
Expand Down Expand Up @@ -653,7 +636,7 @@ mod tests {
"finality_status": "RECEIVED",
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
serde_json::json!({
Expand All @@ -667,7 +650,7 @@ mod tests {
"execution_status": "SUCCEEDED"
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
serde_json::json!({
Expand All @@ -681,7 +664,7 @@ mod tests {
"execution_status": "SUCCEEDED"
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
]
Expand Down Expand Up @@ -713,7 +696,7 @@ mod tests {
"finality_status": "RECEIVED",
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
serde_json::json!({
Expand All @@ -728,7 +711,7 @@ mod tests {
"failure_reason": "tx revert"
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
serde_json::json!({
Expand All @@ -743,7 +726,7 @@ mod tests {
"failure_reason": "tx revert"
}
},
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V07)).unwrap(),
"subscription_id": subscription_id.serialize(Serializer::new(RpcVersion::V08)).unwrap(),
}
}),
]
Expand Down Expand Up @@ -943,7 +926,7 @@ mod tests {
let (receiver_tx, receiver_rx) = mpsc::channel(1024);
handle_json_rpc_socket(router.clone(), sender_tx, receiver_rx);
let params = serde_json::json!(
{"block_id": {"block_number": 0}, "transaction_hash": tx_hash}
{"transaction_hash": tx_hash}
);
receiver_tx
.send(Ok(Message::Text(
Expand Down Expand Up @@ -1097,7 +1080,7 @@ mod tests {
let (receiver_tx, receiver_rx) = mpsc::channel(1024);
handle_json_rpc_socket(router.clone(), sender_tx, receiver_rx);
let params = serde_json::json!(
{"block_id": {"block_number": 0}, "transaction_hash": tx_hash}
{"transaction_hash": tx_hash}
);
receiver_tx
.send(Ok(Message::Text(
Expand All @@ -1122,7 +1105,7 @@ mod tests {
_ => panic!("Expected text message"),
};

let subscription_id = crate::dto::Value::new(subscription_id, RpcVersion::V07);
let subscription_id = crate::dto::Value::new(subscription_id, RpcVersion::V08);
let subscription_id: SubscriptionId = subscription_id.deserialize().unwrap();
for msg in expected(subscription_id) {
let status = sender_rx.recv().await.unwrap().unwrap();
Expand Down
2 changes: 1 addition & 1 deletion doc/rpc/v08/starknet_ws_api.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -360,4 +360,4 @@
}
}
}
}
}
Loading