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

[PAUSED] ✨ [MempoolStatusEvent] Add DataProposalPoda mempool status event #799

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[PAUSED] ✨ [MempoolStatusEvent] Add DataProposalPoda mempool status event #799

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e00c546
add dataproposalpoda mempool status event
hhalex Feb 18, 2025
0a423b8
Fix transaction status not updated + indexer test
hhalex Feb 18, 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
1 change: 1 addition & 0 deletions crates/hyle-model/src/api.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ pub enum TransactionTypeDb {
pub enum TransactionStatusDb {
WaitingDissemination,
DataProposalCreated,
DataProposalPoda,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Je clarifierais bien que c'est un "local Poda" ou quelque chose - tant qu'on est pas passé dans une lane c'est pas "réel". Pas trop idée du wording... D'autant que cet event pourrait être envoyé plusieurs fois.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pour moi une fois qu'on a un poda, ce n'est plus local ? Mais effectivement ça peut être intéressant de garder qqpart tous les events d'updates de podas

Success,
Failure,
Sequenced,
Expand Down
6 changes: 6 additions & 0 deletions crates/hyle-model/src/node/mempool.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,12 @@ pub struct TxId(pub DataProposalHash, pub TxHash);
#[cfg_attr(feature = "full", derive(utoipa::ToSchema))]
pub struct DataProposalHash(pub String);

impl DataProposalHash {
pub fn genesis(v_pubkey: ValidatorPublicKey) -> DataProposalHash {
DataProposalHash(hex::encode(v_pubkey.0))
}
}

impl Hashed<DataProposalHash> for DataProposal {
fn hashed(&self) -> DataProposalHash {
if let Some(hash) = self.hash_cache.read().unwrap().as_ref() {
Expand Down
80 changes: 80 additions & 0 deletions src/indexer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -343,6 +343,45 @@ impl Indexer {
.await
.context("Upserting data at status data_proposal_created")?;
}

MempoolStatusEvent::DataProposalPoda {
data_proposal_hash: _,
txs_metadatas,
signatures: _,
} => {
let mut query_builder = QueryBuilder::new("INSERT INTO transactions (tx_hash, parent_dp_hash, version, transaction_type, transaction_status)");

query_builder.push_values(txs_metadatas, |mut b, value| {
let tx_type: TransactionTypeDb = value.transaction_kind.into();
let version = i32::try_from(value.version)
.map_err(|_| anyhow::anyhow!("Tx version is too large to fit into an i32"))
.log_error("Converting version number into i32")
.unwrap_or(0);

let tx_hash: TxHashDb = value.id.1.into();
let parent_data_proposal_hash_db: DataProposalHashDb = value.id.0.into();

b.push_bind(tx_hash)
.push_bind(parent_data_proposal_hash_db)
.push_bind(version)
.push_bind(tx_type)
.push_bind(TransactionStatusDb::DataProposalPoda);
});

// If the TX is already present, we try to update its status, only if the status is lower ('waiting_dissemination', 'data_proposal_created').
query_builder.push(" ON CONFLICT(tx_hash, parent_dp_hash) DO UPDATE SET ");

query_builder.push("transaction_status=");
query_builder.push_bind(TransactionStatusDb::DataProposalPoda);
query_builder
.push(" WHERE transactions.transaction_status IN ('waiting_dissemination', 'data_proposal_created')");

query_builder
.build()
.execute(transaction.deref_mut())
.await
.context("Upserting data at status data_proposal_poda")?;
}
}

transaction.commit().await?;
Expand Down Expand Up @@ -1196,6 +1235,47 @@ mod test {
)
.await;

let data_proposal_poda_event = MempoolStatusEvent::DataProposalPoda {
signatures: vec![],
data_proposal_hash: data_proposal.hashed(),
txs_metadatas: vec![
register_tx_1_wd.metadata(parent_data_proposal_hash.clone()),
register_tx_2_wd.metadata(parent_data_proposal_hash.clone()),
blob_transaction_wd.metadata(parent_data_proposal_hash.clone()),
proof_tx_1_wd.metadata(parent_data_proposal_hash.clone()),
],
};

indexer
.handle_mempool_status_event(data_proposal_poda_event.clone())
.await
.expect("MempoolStatusEvent");

assert_tx_status(
&server,
register_tx_1_wd.hashed(),
TransactionStatusDb::DataProposalPoda,
)
.await;
assert_tx_status(
&server,
register_tx_2_wd.hashed(),
TransactionStatusDb::DataProposalPoda,
)
.await;
assert_tx_status(
&server,
blob_transaction_wd.hashed(),
TransactionStatusDb::DataProposalPoda,
)
.await;
assert_tx_status(
&server,
proof_tx_1_wd.hashed(),
TransactionStatusDb::DataProposalPoda,
)
.await;

let mut signed_block = SignedBlock::default();
signed_block.consensus_proposal.timestamp = 1234;
signed_block.consensus_proposal.slot = 2;
Expand Down
2 changes: 1 addition & 1 deletion src/indexer/migrations/20241009154948_create_tables.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ CREATE TABLE blocks (
);

CREATE TYPE transaction_type AS ENUM ('blob_transaction', 'proof_transaction', 'stake');
CREATE TYPE transaction_status AS ENUM ('data_proposal_created','waiting_dissemination','success', 'failure', 'sequenced', 'timed_out');
CREATE TYPE transaction_status AS ENUM ('data_proposal_poda', 'data_proposal_created','waiting_dissemination','success', 'failure', 'sequenced', 'timed_out');

CREATE TABLE transactions (
parent_dp_hash TEXT NOT NULL, -- Data Proposal hash
Expand Down
37 changes: 37 additions & 0 deletions src/mempool.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,11 @@ pub enum MempoolStatusEvent {
data_proposal_hash: DataProposalHash,
txs_metadatas: Vec<TransactionMetadata>,
},
DataProposalPoda {
data_proposal_hash: DataProposalHash,
txs_metadatas: Vec<TransactionMetadata>,
signatures: Vec<ValidatorPublicKey>,
},
}
impl BusMessage for MempoolStatusEvent {}

Expand Down Expand Up @@ -827,6 +832,14 @@ impl Mempool {
|| old_voting_power < 2 * f && new_voting_power >= 2 * f
|| new_voting_power == 3 * f + 1
{
let txs_metadatas = self.lanes.tx_metadatas(&data_proposal_hash)?;

self.bus.send(MempoolStatusEvent::DataProposalPoda {
data_proposal_hash: data_proposal_hash.clone(),
txs_metadatas,
signatures: validators.clone(),
})?;

self.broadcast_net_message(MempoolNetMessage::PoDAUpdate(
data_proposal_hash,
signatures,
Expand Down Expand Up @@ -1486,6 +1499,30 @@ pub mod test {
let pub_key = self.validator_pubkey().clone();
self.pop_validator_data_proposal(&pub_key)
}
pub fn peek_data_proposal(&mut self) -> (DataProposal, DataProposalHash, LaneBytesSize) {
let pub_key = self.validator_pubkey().clone();
self.peek_validator_data_proposal(&pub_key)
}

pub fn peek_validator_data_proposal(
&mut self,
validator: &ValidatorPublicKey,
) -> (DataProposal, DataProposalHash, LaneBytesSize) {
// Get the latest lane entry
let latest_data_proposal_hash = self.current_hash(validator).unwrap();
let latest_lane_entry = self
.mempool
.lanes
.get_by_hash(validator, &latest_data_proposal_hash)
.unwrap()
.unwrap();

(
latest_lane_entry.data_proposal.clone(),
latest_data_proposal_hash,
latest_lane_entry.cumul_size,
)
}

pub fn pop_validator_data_proposal(
&mut self,
Expand Down
27 changes: 26 additions & 1 deletion src/mempool/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,31 @@ pub trait Storage {
Ok(cut)
}

fn tx_metadatas(
&mut self,
data_proposal_hash: &DataProposalHash,
) -> Result<Vec<TransactionMetadata>> {
let id = self.id().clone();
match self.get_by_hash(&id, data_proposal_hash)? {
Some(lane_entry) => {
let parent_data_proposal_hash = lane_entry
.data_proposal
.parent_data_proposal_hash
.clone()
.unwrap_or(DataProposalHash::genesis(id.clone()));
let txs_metadatas: Vec<TransactionMetadata> = lane_entry
.data_proposal
.txs
.iter()
.map(|tx| tx.metadata(parent_data_proposal_hash.clone()))
.collect();
Ok(txs_metadatas)
}
None => {
bail!("Could not find lane entry for {id} ({data_proposal_hash})");
}
}
}
// Called by the initial proposal validator to aggregate votes
fn on_data_vote(
&mut self,
Expand Down Expand Up @@ -520,7 +545,7 @@ pub trait Storage {
let parent_data_proposal = data_proposal
.parent_data_proposal_hash
.clone()
.unwrap_or_default();
.unwrap_or_else(|| DataProposalHash(hex::encode(validator_key.0.clone())));

let tx_metadatas = data_proposal
.txs
Expand Down
67 changes: 67 additions & 0 deletions src/tests/autobahn_testing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -339,11 +339,42 @@ async fn autobahn_basic_flow() {
node1.mempool_ctx.submit_tx(&register_tx);
node1.mempool_ctx.submit_tx(&register_tx_2);

assert_chanmsg_matches!(
node1.mempool_ctx.mempool_status_event_receiver,
MempoolStatusEvent::WaitingDissemination { parent_data_proposal_hash, tx } => {
assert_eq!(parent_data_proposal_hash, DataProposalHash(node1.mempool_ctx.validator_pubkey().to_string()));
assert_eq!(tx, register_tx);
}
);

assert_chanmsg_matches!(
node1.mempool_ctx.mempool_status_event_receiver,
MempoolStatusEvent::WaitingDissemination { parent_data_proposal_hash, tx } => {
assert_eq!(parent_data_proposal_hash, DataProposalHash(node1.mempool_ctx.validator_pubkey().to_string()));
assert_eq!(tx, register_tx_2);
}
);

node1
.mempool_ctx
.make_data_proposal_with_pending_txs()
.expect("Should create data proposal");

assert_chanmsg_matches!(
node1.mempool_ctx.mempool_status_event_receiver,
MempoolStatusEvent::DataProposalCreated { data_proposal_hash, txs_metadatas } => {

// First txs refer to a genesis data proposal that has a hash corresponding to validator pubkey
let genesis_dp_hash = hex::encode(node1.mempool_ctx.validator_pubkey().clone().0);
for t in txs_metadatas.iter() {
assert_eq!(genesis_dp_hash, t.id.0.0);
}

assert_eq!(data_proposal_hash, node1.mempool_ctx.peek_data_proposal().0.hashed());
assert_eq!(txs_metadatas.len(), node1.mempool_ctx.peek_data_proposal().0.txs.len());
}
);

broadcast! {
description: "Disseminate Tx",
from: node1.mempool_ctx, to: [node2.mempool_ctx, node3.mempool_ctx, node4.mempool_ctx],
Expand All @@ -369,6 +400,42 @@ async fn autobahn_basic_flow() {
message_matches: MempoolNetMessage::DataVote(_, _)
};

assert_chanmsg_matches!(
node1.mempool_ctx.mempool_status_event_receiver,
MempoolStatusEvent::DataProposalPoda { data_proposal_hash, txs_metadatas, signatures } => {

// First txs refer to a genesis data proposal that has a hash corresponding to validator pubkey
let genesis_dp_hash = hex::encode(node1.mempool_ctx.validator_pubkey().clone().0);
for t in txs_metadatas.iter() {
assert_eq!(genesis_dp_hash, t.id.0.0);
}

assert!(signatures.contains(node1.mempool_ctx.validator_pubkey()));
assert!(signatures.contains(node2.mempool_ctx.validator_pubkey()));

assert_eq!(data_proposal_hash, node1.mempool_ctx.peek_data_proposal().0.hashed());
assert_eq!(txs_metadatas.len(), node1.mempool_ctx.peek_data_proposal().0.txs.len());
}
);
assert_chanmsg_matches!(
node1.mempool_ctx.mempool_status_event_receiver,
MempoolStatusEvent::DataProposalPoda { data_proposal_hash, txs_metadatas, signatures } => {

// First txs refer to a genesis data proposal that has a hash corresponding to validator pubkey
let genesis_dp_hash = hex::encode(node1.mempool_ctx.validator_pubkey().clone().0);
for t in txs_metadatas.iter() {
assert_eq!(genesis_dp_hash, t.id.0.0);
}

assert!(signatures.contains(node1.mempool_ctx.validator_pubkey()));
assert!(signatures.contains(node2.mempool_ctx.validator_pubkey()));
assert!(signatures.contains(node3.mempool_ctx.validator_pubkey()));

assert_eq!(data_proposal_hash, node1.mempool_ctx.peek_data_proposal().0.hashed());
assert_eq!(txs_metadatas.len(), node1.mempool_ctx.peek_data_proposal().0.txs.len());
}
);

let data_proposal_hash_node1 = node1
.mempool_ctx
.current_hash(node1.mempool_ctx.validator_pubkey())
Expand Down
Loading