refactor: bubblegum backfill orders off of slot

rpcpool · Sep 17, 2024 · 4b56567 · 4b56567
1 parent febc55c
commit 4b56567
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 11 deletions.
diff --git a/backfill/src/worker/program_transformer.rs b/backfill/src/worker/program_transformer.rs
@@ -19,9 +19,9 @@ impl ProgramTransformerWorkerArgs {
         &self,
         context: BubblegumBackfillContext,
         forwarder: UnboundedSender<DownloadMetadataInfo>,
-    ) -> Result<(JoinHandle<()>, Sender<(Option<i64>, TransactionInfo)>)> {
+    ) -> Result<(JoinHandle<()>, Sender<TransactionInfo>)> {
         let (sender, mut receiver) =
-            channel::<(Option<i64>, TransactionInfo)>(self.program_transformer_channel_size);
+            channel::<TransactionInfo>(self.program_transformer_channel_size);
 
         let handle = tokio::spawn(async move {
             let mut transactions = Vec::new();
@@ -35,9 +35,9 @@ impl ProgramTransformerWorkerArgs {
                 transactions.push(gap);
             }
 
-            transactions.sort_by(|(a, _), (b, _)| a.cmp(b));
+            transactions.sort_by(|a, b| a.slot.cmp(&b.slot));
 
-            for (_, transaction) in transactions {
+            for transaction in transactions {
                 if let Err(e) = program_transformer.handle_transaction(&transaction).await {
                     error!("handle transaction: {:?}", e)
                 };

diff --git a/backfill/src/worker/transaction.rs b/backfill/src/worker/transaction.rs
@@ -147,7 +147,7 @@ impl SignatureWorkerArgs {
     pub fn start(
         &self,
         context: crate::BubblegumBackfillContext,
-        forwarder: Sender<(Option<i64>, TransactionInfo)>,
+        forwarder: Sender<TransactionInfo>,
     ) -> Result<(JoinHandle<()>, Sender<Signature>)> {
         let (sig_sender, mut sig_receiver) = channel::<Signature>(self.signature_channel_size);
         let worker_count = self.signature_worker_count;
@@ -177,16 +177,13 @@ impl SignatureWorkerArgs {
 
 async fn queue_transaction<'a>(
     client: Rpc,
-    sender: Sender<(Option<i64>, TransactionInfo)>,
+    sender: Sender<TransactionInfo>,
     signature: Signature,
 ) -> Result<(), ErrorKind> {
     let transaction = client.get_transaction(&signature).await?;
 
     sender
-        .send((
-            transaction.block_time,
-            FetchedEncodedTransactionWithStatusMeta(transaction).try_into()?,
-        ))
+        .send(FetchedEncodedTransactionWithStatusMeta(transaction).try_into()?)
         .await
         .map_err(|e| ErrorKind::Generic(e.to_string()))?;
 
@@ -195,7 +192,7 @@ async fn queue_transaction<'a>(
 
 fn spawn_transaction_worker(
     client: Rpc,
-    sender: Sender<(Option<i64>, TransactionInfo)>,
+    sender: Sender<TransactionInfo>,
     signature: Signature,
 ) -> JoinHandle<()> {
     tokio::spawn(async move {