diff --git a/magicblock-committor-service/src/committor_processor.rs b/magicblock-committor-service/src/committor_processor.rs index 5061eb4b0..8e2bea95f 100644 --- a/magicblock-committor-service/src/committor_processor.rs +++ b/magicblock-committor-service/src/committor_processor.rs @@ -174,19 +174,21 @@ impl CommittorProcessor { Ok(signatures) } - pub async fn pending_intent_bundles( + pub async fn load_recovery_intent_bundles( &self, ) -> CommittorServiceResult> { let recovery_time = unix_timestamp(); - let rows = self.persister.get_pending_commit_statuses( - recovery_time.saturating_sub(RECOVERY_MAX_AGE_SECS), - )?; + let min_created_at = + recovery_time.saturating_sub(RECOVERY_MAX_AGE_SECS); + let mut rows = + self.persister.get_pending_commit_statuses(min_created_at)?; + rows.extend(self.persister.get_failed_commit_statuses(min_created_at)?); if rows.is_empty() { return Ok(Vec::new()); } let recovery_base_slot = self.magicblock_rpc_client.get_slot().await?; - let bundles = pending_rows_to_scheduled_intent_bundles( + let bundles = rows_to_scheduled_intent_bundles( rows, self.auth_pubkey(), recovery_base_slot, @@ -200,7 +202,7 @@ impl CommittorProcessor { info!( intent_count = bundles.len(), accounts_count, - "Loaded pending commit intents from persistence for recovery" + "Loaded commit intents from persistence for recovery" ); } @@ -263,7 +265,7 @@ impl CommittorProcessor { } } -fn pending_rows_to_scheduled_intent_bundles( +fn rows_to_scheduled_intent_bundles( rows: Vec, payer: Pubkey, recovery_base_slot: u64, @@ -277,13 +279,13 @@ fn pending_rows_to_scheduled_intent_bundles( grouped_rows .into_iter() .filter_map(|(message_id, rows)| { - if rows.iter().any(|row| { - !pending_row_is_in_recovery_window(row, recovery_time) - }) + if rows + .iter() + .any(|row| !row_is_in_recovery_window(row, recovery_time)) { warn!( intent_id = message_id, - "Skipping pending commit intent outside recovery window" + "Skipping commit intent outside recovery window" ); return None; } @@ -296,7 +298,7 @@ fn pending_rows_to_scheduled_intent_bundles( }) { warn!( intent_id = message_id, - "Skipping pending commit intent: rows disagree on slot or ephemeral_blockhash" + "Skipping commit intent: rows disagree on slot or ephemeral_blockhash" ); return None; } @@ -305,7 +307,7 @@ fn pending_rows_to_scheduled_intent_bundles( for row in rows { let Some((account, undelegate)) = - committed_account_from_pending_row(row, recovery_base_slot) + committed_account_from_row(row, recovery_base_slot) else { continue; }; @@ -346,7 +348,7 @@ fn pending_rows_to_scheduled_intent_bundles( .collect() } -fn pending_row_is_in_recovery_window( +fn row_is_in_recovery_window( row: &CommitStatusRow, recovery_time: u64, ) -> bool { @@ -360,7 +362,7 @@ fn unix_timestamp() -> u64 { .as_secs() } -fn committed_account_from_pending_row( +fn committed_account_from_row( row: CommitStatusRow, recovery_base_slot: u64, ) -> Option<(CommittedAccount, bool)> { @@ -456,7 +458,7 @@ mod tests { let commit_pubkey = Pubkey::new_unique(); let undelegate_pubkey = Pubkey::new_unique(); - let bundles = pending_rows_to_scheduled_intent_bundles( + let bundles = rows_to_scheduled_intent_bundles( vec![ pending_row( 9, @@ -501,7 +503,7 @@ mod tests { let payer = Pubkey::new_unique(); let recovery_time = RECOVERY_MAX_AGE_SECS + 1; - let recoverable = pending_rows_to_scheduled_intent_bundles( + let recoverable = rows_to_scheduled_intent_bundles( vec![pending_row_with_timestamps( 1, recovery_time - RECOVERY_MAX_AGE_SECS, @@ -513,7 +515,7 @@ mod tests { ); assert_eq!(recoverable.len(), 1); - let recent = pending_rows_to_scheduled_intent_bundles( + let recent = rows_to_scheduled_intent_bundles( vec![pending_row_with_timestamps(2, recovery_time, recovery_time)], payer, 7, @@ -521,7 +523,7 @@ mod tests { ); assert_eq!(recent.len(), 1); - let too_old = pending_rows_to_scheduled_intent_bundles( + let too_old = rows_to_scheduled_intent_bundles( vec![pending_row_with_timestamps( 3, recovery_time - RECOVERY_MAX_AGE_SECS - 1, diff --git a/magicblock-committor-service/src/persist/commit_persister.rs b/magicblock-committor-service/src/persist/commit_persister.rs index 4d17be308..2de11a607 100644 --- a/magicblock-committor-service/src/persist/commit_persister.rs +++ b/magicblock-committor-service/src/persist/commit_persister.rs @@ -61,6 +61,10 @@ pub trait IntentPersister: Send + Sync + Clone + 'static { &self, min_created_at: u64, ) -> CommitPersistResult>; + fn get_failed_commit_statuses( + &self, + min_created_at: u64, + ) -> CommitPersistResult>; fn get_commit_status_by_message( &self, message_id: u64, @@ -257,6 +261,16 @@ impl IntentPersister for IntentPersisterImpl { .get_pending_commit_statuses(min_created_at) } + fn get_failed_commit_statuses( + &self, + min_created_at: u64, + ) -> CommitPersistResult> { + self.commits_db + .lock() + .expect(POISONED_MUTEX_MSG) + .get_failed_commit_statuses(min_created_at) + } + fn get_commit_status_by_message( &self, message_id: u64, @@ -415,6 +429,18 @@ impl IntentPersister for Option { } } + fn get_failed_commit_statuses( + &self, + min_created_at: u64, + ) -> CommitPersistResult> { + match self { + Some(persister) => { + persister.get_failed_commit_statuses(min_created_at) + } + None => Ok(Vec::new()), + } + } + fn get_commit_status_by_message( &self, message_id: u64, diff --git a/magicblock-committor-service/src/persist/db.rs b/magicblock-committor-service/src/persist/db.rs index 314c627ff..56e93e891 100644 --- a/magicblock-committor-service/src/persist/db.rs +++ b/magicblock-committor-service/src/persist/db.rs @@ -585,6 +585,35 @@ impl CommittsDb { extract_committor_rows(&mut rows) } + pub(crate) fn get_failed_commit_statuses( + &self, + min_created_at: u64, + ) -> CommitPersistResult> { + let query = format!( + "WITH eligible_messages AS ( + SELECT message_id + FROM commit_status + WHERE commit_status IN (?1, ?2) + GROUP BY message_id + HAVING MIN(created_at) >= ?3 + ) + {SELECT_ALL_COMMIT_STATUS_COLUMNS} + WHERE commit_status IN (?1, ?2) + AND message_id IN ( + SELECT message_id FROM eligible_messages + ) + ORDER BY message_id, created_at, pubkey" + ); + let stmt = &mut self.conn.prepare(&query)?; + let mut rows = stmt.query(params![ + "FailedFinalize", + "FailedProcess", + u64_into_i64(min_created_at) + ])?; + + extract_committor_rows(&mut rows) + } + pub(crate) fn get_commit_status( &self, message_id: u64, @@ -919,6 +948,105 @@ mod tests { ); } + #[test] + fn test_get_failed_commit_statuses() { + let (mut db, _file) = setup_test_db(); + let pending = create_test_row(1, 0); + let mut failed_finalize = create_test_row(2, 0); + failed_finalize.commit_status = + CommitStatus::FailedFinalize(CommitStatusSignatures { + commit_stage_signature: Signature::new_unique(), + finalize_stage_signature: None, + }); + let mut failed_process = create_test_row(3, 0); + failed_process.commit_status = CommitStatus::FailedProcess(None); + let mut succeeded = create_test_row(4, 0); + succeeded.commit_status = + CommitStatus::Succeeded(CommitStatusSignatures { + commit_stage_signature: Signature::new_unique(), + finalize_stage_signature: Some(Signature::new_unique()), + }); + + db.insert_commit_status_rows(&[ + pending, + failed_finalize.clone(), + failed_process.clone(), + succeeded, + ]) + .unwrap(); + + let rows = db.get_failed_commit_statuses(0).unwrap(); + assert_eq!(rows, vec![failed_finalize, failed_process]); + } + + #[test] + fn test_get_failed_commit_statuses_filters_recovery_window() { + let (mut db, _file) = setup_test_db(); + let mut failed = create_test_row(1, 0); + failed.created_at = 10; + failed.last_retried_at = 19; + failed.commit_status = + CommitStatus::FailedFinalize(CommitStatusSignatures { + commit_stage_signature: Signature::new_unique(), + finalize_stage_signature: None, + }); + let mut failed_same_message = create_test_row(1, 0); + failed_same_message.created_at = 11; + failed_same_message.last_retried_at = 18; + failed_same_message.commit_status = CommitStatus::FailedProcess(None); + let mut too_old = create_test_row(2, 0); + too_old.created_at = 9; + too_old.last_retried_at = 9; + too_old.commit_status = + CommitStatus::FailedFinalize(CommitStatusSignatures { + commit_stage_signature: Signature::new_unique(), + finalize_stage_signature: None, + }); + let mut too_recent = create_test_row(3, 0); + too_recent.created_at = 20; + too_recent.last_retried_at = 20; + too_recent.commit_status = + CommitStatus::FailedFinalize(CommitStatusSignatures { + commit_stage_signature: Signature::new_unique(), + finalize_stage_signature: None, + }); + let mut partially_eligible = create_test_row(4, 0); + partially_eligible.created_at = 10; + partially_eligible.last_retried_at = 19; + partially_eligible.commit_status = + CommitStatus::FailedFinalize(CommitStatusSignatures { + commit_stage_signature: Signature::new_unique(), + finalize_stage_signature: None, + }); + let mut partially_failed_process = create_test_row(4, 0); + partially_failed_process.created_at = 11; + partially_failed_process.last_retried_at = 20; + partially_failed_process.commit_status = + CommitStatus::FailedProcess(None); + + db.insert_commit_status_rows(&[ + failed.clone(), + failed_same_message.clone(), + too_old, + too_recent.clone(), + partially_eligible.clone(), + partially_failed_process.clone(), + ]) + .unwrap(); + + let rows = db.get_failed_commit_statuses(10).unwrap(); + assert_eq!( + rows, + vec![ + failed, + failed_same_message, + too_recent, + partially_eligible, + partially_failed_process, + ] + ); + } + #[test] fn test_insert_and_retrieve_bundle_signature() { let (db, _file) = setup_test_db(); diff --git a/magicblock-committor-service/src/service.rs b/magicblock-committor-service/src/service.rs index b1415b8bc..fb492f8d3 100644 --- a/magicblock-committor-service/src/service.rs +++ b/magicblock-committor-service/src/service.rs @@ -209,7 +209,7 @@ impl CommittorActor { } GetPendingIntentBundles { respond_to } => { let pending_intents = - self.processor.pending_intent_bundles().await; + self.processor.load_recovery_intent_bundles().await; if let Err(e) = respond_to.send(pending_intents) { error!(message_type = "GetPendingIntentBundles", error = ?e, "Failed to send response"); }