From 30deabd6835692bb90a78c9f198e76d191bf28a3 Mon Sep 17 00:00:00 2001 From: taco-paco Date: Tue, 30 Jun 2026 13:56:55 +0700 Subject: [PATCH 1/6] hotfix: load failed intents on restart & reschedule --- .../src/committor_processor.rs | 8 +- .../src/persist/commit_persister.rs | 12 +- .../src/persist/db.rs | 128 ++++++++++++++++++ magicblock-committor-service/src/service.rs | 9 +- 4 files changed, 144 insertions(+), 13 deletions(-) diff --git a/magicblock-committor-service/src/committor_processor.rs b/magicblock-committor-service/src/committor_processor.rs index b610480f7..5012b277b 100644 --- a/magicblock-committor-service/src/committor_processor.rs +++ b/magicblock-committor-service/src/committor_processor.rs @@ -154,11 +154,11 @@ impl CommittorProcessor { Ok(signatures) } - /// Fetches pending bundles from DB - pub async fn pending_intent_bundles( + /// Fetches pending and failed bundles from DB for recovery + pub async fn load_recovery_intent_bundles( &self, ) -> CommittorServiceResult> { - // Extract pending bundles satisfying predicate + // Extract pending and failed bundles satisfying predicate let now = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() @@ -180,7 +180,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" ); } diff --git a/magicblock-committor-service/src/persist/commit_persister.rs b/magicblock-committor-service/src/persist/commit_persister.rs index 4c641863f..efc816115 100644 --- a/magicblock-committor-service/src/persist/commit_persister.rs +++ b/magicblock-committor-service/src/persist/commit_persister.rs @@ -254,7 +254,7 @@ impl IntentPersister for IntentPersisterImpl { .get_commit_statuses_by_id(message_id) } - /// Returns pending bundles created at or after `min_created_at`. + /// Returns pending and failed bundles created at or after `min_created_at`. /// NOTE: this constructs `ScheduleIntentBundle` only from existing information. /// As persister doesn't save `ScheduleIntentBundle::payer` info, Pubkey::default is used. /// `CommittedAccount` information like slot may also be outdated. @@ -263,11 +263,11 @@ impl IntentPersister for IntentPersisterImpl { &self, min_created_at: u64, ) -> CommitPersistResult> { - let rows = self - .commits_db - .lock() - .expect(POISONED_MUTEX_MSG) - .get_pending_commit_statuses(min_created_at)?; + let commits_db = self.commits_db.lock().expect(POISONED_MUTEX_MSG); + let mut rows = + commits_db.get_pending_commit_statuses(min_created_at)?; + rows.extend(commits_db.get_failed_commit_statuses(min_created_at)?); + drop(commits_db); Ok(pending_rows_to_scheduled_intent_bundles( rows, 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 572ada184..0c9b6a07e 100644 --- a/magicblock-committor-service/src/service.rs +++ b/magicblock-committor-service/src/service.rs @@ -180,9 +180,12 @@ where } async fn reschedule_pending_bundles(&self) -> CommittorServiceResult<()> { - // Fetch pending bundles from DB - let mut bundles = - self.processor.pending_intent_bundles().await.inspect_err(|err| { + // Fetch pending and failed bundles from DB + let mut bundles = self + .processor + .load_recovery_intent_bundles() + .await + .inspect_err(|err| { error!(error = ?err, "Failed to load pending intent bundles for recovery"); })?; if bundles.is_empty() { From 1206ca77801b7f807386309c871b756e13c91ba9 Mon Sep 17 00:00:00 2001 From: taco-paco Date: Wed, 29 Jul 2026 00:02:53 +0300 Subject: [PATCH 2/6] fix: retry failed intents with ensuring they're safe to retry --- .../src/committor_processor.rs | 39 ++-- .../src/persist/commit_persister.rs | 127 +++++++++++-- .../src/persist/db.rs | 137 +++++++++++++- .../src/persist/mod.rs | 4 +- magicblock-committor-service/src/service.rs | 178 ++++++++++++------ 5 files changed, 391 insertions(+), 94 deletions(-) diff --git a/magicblock-committor-service/src/committor_processor.rs b/magicblock-committor-service/src/committor_processor.rs index 5012b277b..2f491d1bc 100644 --- a/magicblock-committor-service/src/committor_processor.rs +++ b/magicblock-committor-service/src/committor_processor.rs @@ -32,7 +32,7 @@ use crate::{ }, persist::{ CommitStatusRow, IntentPersister, IntentPersisterImpl, - MessageSignatures, + MessageSignatures, RecoveredIntent, }, }; const POISONED_MUTEX_MSG: &str = @@ -154,43 +154,40 @@ impl CommittorProcessor { Ok(signatures) } - /// Fetches pending and failed bundles from DB for recovery + /// Fetches pending and failed bundles from DB for recovery. No filtering. pub async fn load_recovery_intent_bundles( &self, - ) -> CommittorServiceResult> { - // Extract pending and failed bundles satisfying predicate + ) -> CommittorServiceResult> { let now = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_secs(); - let mut bundles = self.persister.pending_intent_bundles( - now.saturating_sub(RECOVERY_MAX_AGE_SECS), - )?; - - if bundles.is_empty() { - return Ok(bundles); - } + let recovered = self + .persister + .recoverable_intents(now.saturating_sub(RECOVERY_MAX_AGE_SECS))?; - // Log info about extracted bundles - { - let accounts_count: usize = bundles + if !recovered.is_empty() { + let accounts_count: usize = recovered .iter() - .map(|bundle| bundle.get_all_committed_pubkeys().len()) + .map(|r| r.bundle.get_all_committed_pubkeys().len()) .sum(); info!( - intent_count = bundles.len(), + intent_count = recovered.len(), accounts_count, "Loaded commit intents from persistence for recovery" ); } - // Extracted bundles are out of data and missing some of the info - self.refresh_intent_bundles(&mut bundles).await?; - - Ok(bundles) + Ok(recovered) } - async fn refresh_intent_bundles( + /// Stamps `payer` and `remote_slot` on recovered bundles with current + /// values so execution (fetching nonces/base accounts) uses a fresh + /// `min_context_slot`. Must run only on bundles that already survived + /// the caller's nonce and delegation-session recovery filters, since it + /// destroys the original scheduling-time `remote_slot` those filters + /// depend on. + pub(crate) async fn refresh_intent_bundles( &self, intent_bundles: &mut [ScheduledIntentBundle], ) -> CommittorServiceResult<()> { diff --git a/magicblock-committor-service/src/persist/commit_persister.rs b/magicblock-committor-service/src/persist/commit_persister.rs index efc816115..16290c760 100644 --- a/magicblock-committor-service/src/persist/commit_persister.rs +++ b/magicblock-committor-service/src/persist/commit_persister.rs @@ -1,5 +1,5 @@ use std::{ - collections::BTreeMap, + collections::{BTreeMap, HashMap}, path::Path, sync::{Arc, Mutex}, }; @@ -140,6 +140,7 @@ impl IntentPersisterImpl { commit_status: CommitStatus::Pending, last_retried_at: created_at, retries_count: 0, + remote_slot: account.remote_slot, } }; @@ -164,6 +165,23 @@ impl IntentPersisterImpl { }) .collect() } + + /// Loads pending/failed bundles for restart recovery, each paired with + /// the `commit_id` (nonce) its committed accounts were persisted with. + /// See [`RecoveredIntent`] for why this isn't part of the + /// `IntentPersister::pending_intent_bundles` trait method. + pub fn recoverable_intents( + &self, + min_created_at: u64, + ) -> CommitPersistResult> { + let commits_db = self.commits_db.lock().expect(POISONED_MUTEX_MSG); + let mut rows = + commits_db.get_pending_commit_statuses(min_created_at)?; + rows.extend(commits_db.get_failed_commit_statuses(min_created_at)?); + drop(commits_db); + + Ok(pending_rows_to_recovered_intents(rows, min_created_at)) + } } impl IntentPersister for IntentPersisterImpl { @@ -481,10 +499,31 @@ impl IntentPersister for Option { } } +/// A recovered bundle paired with the `commit_id` (nonce) each of its +/// committed accounts was persisted with. `commit_id` is committor-internal +/// bookkeeping, not chain-facing state, so it doesn't live on +/// `CommittedAccount`/`ScheduledIntentBundle` themselves - keeping it bundled +/// here (rather than a separate map keyed by bundle id) means a consumer can +/// never observe a bundle without its matching commit_ids. +pub struct RecoveredIntent { + pub bundle: ScheduledIntentBundle, + pub commit_ids: HashMap, +} + fn pending_rows_to_scheduled_intent_bundles( rows: Vec, min_created_at: u64, ) -> Vec { + pending_rows_to_recovered_intents(rows, min_created_at) + .into_iter() + .map(|recovered| recovered.bundle) + .collect() +} + +fn pending_rows_to_recovered_intents( + rows: Vec, + min_created_at: u64, +) -> Vec { let mut grouped_rows = BTreeMap::>::new(); for row in rows { grouped_rows.entry(row.message_id).or_default().push(row); @@ -519,13 +558,13 @@ fn pending_rows_to_scheduled_intent_bundles( Some(rows) } }) - .filter_map(intent_bundle_from_rows) + .filter_map(recovered_intent_from_rows) .collect() } -fn intent_bundle_from_rows( +fn recovered_intent_from_rows( rows: Vec, -) -> Option { +) -> Option { let first = rows.first()?; let message_id = first.message_id; let slot = first.slot; @@ -533,11 +572,15 @@ fn intent_bundle_from_rows( let mut commit_finalize_accounts = Vec::new(); let mut commit_finalize_and_undelegate_accounts = Vec::new(); + let mut commit_ids = HashMap::new(); for row in rows { + let pubkey = row.pubkey; + let commit_id = row.commit_id; let Some((account, undelegate)) = committed_account_from_row(row) else { continue; }; + commit_ids.insert(pubkey, commit_id); if undelegate { commit_finalize_and_undelegate_accounts.push(account); } else { @@ -563,13 +606,16 @@ fn intent_bundle_from_rows( return None; } - Some(ScheduledIntentBundle { - id: message_id, - slot, - blockhash, - sent_transaction: Transaction::default(), - payer: Pubkey::default(), - intent_bundle, + Some(RecoveredIntent { + bundle: ScheduledIntentBundle { + id: message_id, + slot, + blockhash, + sent_transaction: Transaction::default(), + payer: Pubkey::default(), + intent_bundle, + }, + commit_ids, }) } @@ -595,7 +641,7 @@ fn committed_account_from_row( executable: false, rent_epoch: 0, }, - remote_slot: 0, + remote_slot: row.remote_slot, }, row.undelegate, )) @@ -729,11 +775,67 @@ mod tests { let empty_account = rows.iter().find(|r| r.data.is_none()).unwrap(); assert_eq!(empty_account.commit_type, types::CommitType::EmptyAccount); assert_eq!(empty_account.lamports, 1000); + assert_eq!(empty_account.remote_slot, 1); let data_account = rows.iter().find(|r| r.data.is_some()).unwrap(); assert_eq!(data_account.commit_type, types::CommitType::DataAccount); assert_eq!(data_account.lamports, 2000); assert_eq!(data_account.data, Some(vec![1, 2, 3])); + assert_eq!(data_account.remote_slot, 2); + } + + #[test] + fn test_remote_slot_round_trips_through_persistence() { + // Regression test: `remote_slot` (the base chain slot observed when + // the intent was built) must survive create_commit_rows -> DB -> + // committed_account_from_row, since recovery-time staleness checks + // depend on it reflecting the original scheduling slot, not 0. + let message = create_test_message(1, commit_only_bundle()); + let rows = IntentPersisterImpl::create_commit_rows(&message); + assert!(rows.iter().any(|r| r.remote_slot == 1)); + assert!(rows.iter().any(|r| r.remote_slot == 2)); + + let (persister, _temp_file) = create_test_persister(); + persister.start_base_intent(&message).unwrap(); + let stored = persister.get_commit_statuses_by_message(1).unwrap(); + assert!(stored.iter().any(|r| r.remote_slot == 1)); + assert!(stored.iter().any(|r| r.remote_slot == 2)); + + let reconstructed = pending_rows_to_scheduled_intent_bundles(stored, 0); + let remote_slots: Vec = reconstructed[0] + .get_all_committed_accounts() + .iter() + .map(|a| a.remote_slot) + .collect(); + assert!(remote_slots.contains(&1)); + assert!(remote_slots.contains(&2)); + } + + #[test] + fn test_recoverable_intents_pairs_bundle_with_its_commit_ids() { + let owner = Pubkey::new_unique(); + let blockhash = Hash::new_unique(); + let pubkey_a = Pubkey::new_unique(); + let pubkey_b = Pubkey::new_unique(); + + let (persister, _temp_file) = create_test_persister(); + persister + .commits_db + .lock() + .unwrap() + .insert_commit_status_rows(&[ + pending_row(1, pubkey_a, owner, blockhash, false, None), + pending_row(1, pubkey_b, owner, blockhash, false, None), + ]) + .unwrap(); + persister.set_commit_id(1, &pubkey_a, 5).unwrap(); + persister.set_commit_id(1, &pubkey_b, 7).unwrap(); + + let recovered = persister.recoverable_intents(0).unwrap(); + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].bundle.id, 1); + assert_eq!(recovered[0].commit_ids.get(&pubkey_a), Some(&5)); + assert_eq!(recovered[0].commit_ids.get(&pubkey_b), Some(&7)); } #[test] @@ -960,6 +1062,7 @@ mod tests { commit_strategy: Default::default(), last_retried_at: 1, retries_count: 0, + remote_slot: 0, } } diff --git a/magicblock-committor-service/src/persist/db.rs b/magicblock-committor-service/src/persist/db.rs index 56e93e891..9be1d74b3 100644 --- a/magicblock-committor-service/src/persist/db.rs +++ b/magicblock-committor-service/src/persist/db.rs @@ -54,6 +54,10 @@ pub struct CommitStatusRow { pub last_retried_at: u64, /// Number of times the commit was retried pub retries_count: u16, + /// The base chain slot at which the committed account's state was + /// observed when the intent was built (`CommittedAccount::remote_slot`). + /// Defaults to 0 for rows written before this column existed. + pub remote_slot: u64, } #[derive(Debug)] @@ -86,7 +90,8 @@ impl fmt::Display for CommitStatusRow { commit_status: {}, commit_strategy: {}, last_retried_at: {}, - retries_count: {} + retries_count: {}, + remote_slot: {} }}", self.message_id, self.pubkey, @@ -102,7 +107,8 @@ impl fmt::Display for CommitStatusRow { self.commit_status, self.commit_strategy.as_str(), self.last_retried_at, - self.retries_count + self.retries_count, + self.remote_slot ) } } @@ -124,7 +130,8 @@ const ALL_COMMIT_STATUS_COLUMNS: &str = " commit_stage_signature, finalize_stage_signature, last_retried_at, - retries_count + retries_count, + remote_slot "; const SELECT_ALL_COMMIT_STATUS_COLUMNS: &str = r#" @@ -145,7 +152,8 @@ SELECT commit_stage_signature, finalize_stage_signature, last_retried_at, - retries_count + retries_count, + remote_slot FROM commit_status "#; @@ -366,6 +374,7 @@ impl CommittsDb { finalize_stage_signature TEXT, last_retried_at INTEGER NOT NULL, retries_count INTEGER NOT NULL, + remote_slot INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (message_id, commit_id, pubkey) ); CREATE INDEX IF NOT EXISTS idx_commits_pubkey ON commit_status (pubkey); @@ -373,12 +382,32 @@ impl CommittsDb { CREATE INDEX IF NOT EXISTS idx_commits_commit_id ON commit_status (commit_id); COMMIT;", ) { - Ok(_) => Ok(()), + Ok(_) => {} Err(err) => { error!(error = ?err, "Failed to create commit_status table"); - Err(err) + return Err(err); } } + self.migrate_add_remote_slot_column() + } + + /// Adds the `remote_slot` column to `commit_status` for databases created + /// before this column existed. `CREATE TABLE IF NOT EXISTS` above is a + /// no-op against an already-existing table, so a missing column needs an + /// explicit, idempotent `ALTER TABLE`. Existing rows backfill to 0, which + /// is indistinguishable from "never verified" for the recovery filter + /// since real base chain slots are never 0. + fn migrate_add_remote_slot_column(&self) -> Result<()> { + let has_column = self + .conn + .prepare("SELECT 1 FROM pragma_table_info('commit_status') WHERE name = 'remote_slot'")? + .exists([])?; + if has_column { + return Ok(()); + } + self.conn.execute_batch( + "ALTER TABLE commit_status ADD COLUMN remote_slot INTEGER NOT NULL DEFAULT 0;", + ) } // ----------------- @@ -502,7 +531,7 @@ impl CommittsDb { tx.execute( &format!( "INSERT INTO commit_status ({ALL_COMMIT_STATUS_COLUMNS}) VALUES - (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)", + (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)", ), params![ u64_into_i64(commit.message_id), @@ -526,6 +555,7 @@ impl CommittsDb { .map(|s| s.to_string()), u64_into_i64(commit.last_retried_at), commit.retries_count, + u64_into_i64(commit.remote_slot), ], )?; Ok(()) @@ -777,6 +807,10 @@ fn extract_committor_row( let retries_count: i64 = row.get(16)?; retries_count.try_into().unwrap_or_default() }; + let remote_slot: u64 = { + let remote_slot: i64 = row.get(17)?; + i64_into_u64(remote_slot) + }; Ok(CommitStatusRow { message_id, @@ -791,6 +825,7 @@ fn extract_committor_row( commit_type, created_at, commit_strategy, + remote_slot, commit_status, last_retried_at, retries_count, @@ -817,6 +852,93 @@ mod tests { (db, temp_file) } + #[test] + fn test_migrate_add_remote_slot_column_backfills_existing_rows() { + test_utils::init_test_logger(); + let temp_file = NamedTempFile::new().unwrap(); + let db = CommittsDb::new(temp_file.path()).unwrap(); + + // Simulate a pre-migration database: create the table without the + // remote_slot column, exactly as it looked before this column existed. + db.conn + .execute_batch( + " + CREATE TABLE commit_status ( + message_id INTEGER NOT NULL, + pubkey TEXT NOT NULL, + commit_id INTEGER NOT NULL, + delegated_account_owner TEXT NOT NULL, + slot INTEGER NOT NULL, + ephemeral_blockhash TEXT NOT NULL, + undelegate INTEGER NOT NULL, + lamports INTEGER NOT NULL, + data BLOB, + commit_type TEXT NOT NULL, + created_at INTEGER NOT NULL, + commit_strategy TEXT NOT NULL, + commit_status TEXT NOT NULL, + commit_stage_signature TEXT, + finalize_stage_signature TEXT, + last_retried_at INTEGER NOT NULL, + retries_count INTEGER NOT NULL, + PRIMARY KEY (message_id, commit_id, pubkey) + );", + ) + .unwrap(); + + // A real, fully-valid row (not migration-affected placeholder + // strings) so we can round-trip it through the typed read path below + // and catch any parsing issue the migration might introduce, not + // just confirm the new column exists. + let old_row = create_test_row(1, 0); + db.conn + .execute( + "INSERT INTO commit_status ( + message_id, pubkey, commit_id, delegated_account_owner, + slot, ephemeral_blockhash, undelegate, lamports, data, + commit_type, created_at, commit_strategy, commit_status, + commit_stage_signature, finalize_stage_signature, + last_retried_at, retries_count + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)", + params![ + u64_into_i64(old_row.message_id), + old_row.pubkey.to_string(), + u64_into_i64(old_row.commit_id), + old_row.delegated_account_owner.to_string(), + u64_into_i64(old_row.slot), + old_row.ephemeral_blockhash.to_string(), + if old_row.undelegate { 1 } else { 0 }, + u64_into_i64(old_row.lamports), + old_row.data.as_deref(), + old_row.commit_type.as_str(), + u64_into_i64(old_row.created_at), + old_row.commit_strategy.as_str(), + old_row.commit_status.as_str(), + None::, + None::, + u64_into_i64(old_row.last_retried_at), + old_row.retries_count, + ], + ) + .unwrap(); + + // Running create_commit_status_table against the pre-existing table + // (CREATE TABLE IF NOT EXISTS is a no-op) must still add the column. + db.create_commit_status_table().unwrap(); + // And it must be idempotent on repeated calls (e.g. every startup). + db.create_commit_status_table().unwrap(); + + // The row must still be readable through the normal typed query + // path, with every pre-existing field intact and remote_slot + // backfilled to 0. + let migrated_row = db.get_commit_statuses_by_id(1).unwrap(); + let expected_row = CommitStatusRow { + remote_slot: 0, + ..old_row + }; + assert_eq!(migrated_row, vec![expected_row]); + } + // Helper to create a test CommitStatusRow fn create_test_row(message_id: u64, commit_id: u64) -> CommitStatusRow { CommitStatusRow { @@ -835,6 +957,7 @@ mod tests { commit_strategy: CommitStrategy::StateArgs, last_retried_at: 1000, retries_count: 0, + remote_slot: 100, } } diff --git a/magicblock-committor-service/src/persist/mod.rs b/magicblock-committor-service/src/persist/mod.rs index 6d7083055..ee7e41dd1 100644 --- a/magicblock-committor-service/src/persist/mod.rs +++ b/magicblock-committor-service/src/persist/mod.rs @@ -4,7 +4,9 @@ pub mod error; mod types; mod utils; -pub use commit_persister::{IntentPersister, IntentPersisterImpl}; +pub use commit_persister::{ + IntentPersister, IntentPersisterImpl, RecoveredIntent, +}; pub use db::{CommitStatusRow, CommittsDb, MessageSignatures}; pub use types::{ CommitStatus, CommitStatusSignatures, CommitStrategy, CommitType, diff --git a/magicblock-committor-service/src/service.rs b/magicblock-committor-service/src/service.rs index 0c9b6a07e..05cbe71bf 100644 --- a/magicblock-committor-service/src/service.rs +++ b/magicblock-committor-service/src/service.rs @@ -13,7 +13,9 @@ use intent_client::{ ERIntentClient, InternalIntentClientError, ScheduledBaseIntentMeta, }; use magicblock_account_cloner::ChainlinkCloner; -use magicblock_chainlink::{ProdChainlink, ProdInnerChainlink}; +use magicblock_chainlink::{ + errors::ChainlinkResult, ProdChainlink, ProdInnerChainlink, +}; use magicblock_metrics::metrics::{ self, AccountFetchContext, AccountFetchReason, }; @@ -26,11 +28,13 @@ use tokio::{ task::{JoinError, JoinHandle}, }; use tokio_util::sync::CancellationToken; -use tracing::{error, info, instrument, warn}; +use tracing::{error, info, instrument}; use crate::{ committor_processor::CommittorProcessor, error::CommittorServiceResult, intent_execution_manager::BroadcastedIntentExecutionResult, + intent_executor::task_info_fetcher::TaskInfoFetcherResult, + persist::RecoveredIntent, }; const POISONED_MUTEX_MSG: &str = "ServiceInner intents_meta_map mutex poisoned"; @@ -180,20 +184,27 @@ where } async fn reschedule_pending_bundles(&self) -> CommittorServiceResult<()> { - // Fetch pending and failed bundles from DB - let mut bundles = self + // Fetch pending and failed bundles from DB, unfiltered + let recovered = self .processor .load_recovery_intent_bundles() .await .inspect_err(|err| { error!(error = ?err, "Failed to load pending intent bundles for recovery"); })?; + if recovered.is_empty() { + return Ok(()); + } + + // Drop bundles that are no longer safe to replay + let mut bundles = self.retain_recoverable_intents(recovered).await; if bundles.is_empty() { return Ok(()); } - // Retain only recoverable bundles - self.retain_recoverable_intent_bundles(&mut bundles).await; + // Bundles are out of date and missing some info now that they + // survived the recovery filters, which needed the original values. + self.processor.refresh_intent_bundles(&mut bundles).await?; // Schedule without initial persisitance as bundle already exists in db self.process_intent_bundles(bundles, |bundles| { @@ -374,55 +385,116 @@ where Ok(()) } - /// Retains bundles whose accounts are still delegated - async fn retain_recoverable_intent_bundles( + /// Filters recovered intents down to the ones still safe to replay, + /// checking up to `JOIN_CHUNK_SIZE` at a time in parallel. + async fn retain_recoverable_intents( &self, - bundles: &mut Vec, - ) { - let results = join_all( - bundles.iter().map(|b| b.get_all_committed_pubkeys()).map( - |pubkeys| async move { - #[allow(deprecated)] - let result = self - .chainlink - .accounts_delegated_on_base_and_er( - &pubkeys, - AccountFetchContext::internal( - AccountFetchReason::RequestedAccount, - ), - ) - .await; - result - }, - ), - ) - .await; - - let mut results_iter = results.into_iter(); - bundles.retain(|bundle| { - let Some(result) = results_iter.next() else { - error!("Results and bundles must have equal length"); - return false; - }; - match result { - Ok(delegated) if delegated.iter().all(|d| *d) => true, - Ok(_) => { - warn!( - intent_id = bundle.id, - "Skipping recovered commit intent because not all accounts are delegated on base and ER" - ); - false - } - Err(err) => { - error!( - intent_id = bundle.id, - error = ?err, - "Failed to verify recovered commit intent accounts" - ); - false - } - } + recovered: Vec, + ) -> Vec { + const JOIN_CHUNK_SIZE: usize = 10; + + let mut to_keep = Vec::with_capacity(recovered.len()); + for chunk in recovered.chunks(JOIN_CHUNK_SIZE) { + let iter = chunk.iter().map(|recovered| async move { + self.is_intent_retriable(recovered).await + }); + to_keep.extend(join_all(iter).await); + } + + recovered + .into_iter() + .enumerate() + .filter(|(i, _)| to_keep[*i]) + .map(|(_, el)| el.bundle) + .collect() + } + + /// An intent is retriable only if its delegation session is unchanged + /// and its commit nonce hasn't already landed on chain. + async fn is_intent_retriable(&self, recovered: &RecoveredIntent) -> bool { + let pubkeys = recovered.bundle.get_all_committed_pubkeys(); + let result = self + .is_same_delegation_session(&pubkeys, recovered) + .await + .inspect_err(|err| { + error!(intent_id = recovered.bundle.id, error = ?err, "Failed to check delegation session for recovery"); + }) + .unwrap_or(false); + + if !result { + false + } else { + self.is_valid_nonce(&pubkeys, recovered) + .await + .inspect_err(|err| { + error!(intent_id = recovered.bundle.id, error = ?err, "Failed to check commit nonce for recovery"); + }) + .unwrap_or(false) + } + } + + /// True if every committed account is still delegated (or undelegating) + /// at the same `remote_slot` recorded when the intent was scheduled. + async fn is_same_delegation_session( + &self, + pubkeys: &[Pubkey], + recovered: &RecoveredIntent, + ) -> ChainlinkResult { + let recovered_accounts = recovered.bundle.get_all_committed_accounts(); + let current_accounts = self + .chainlink + .fetch_accounts( + pubkeys, + AccountFetchContext::internal( + AccountFetchReason::RequestedAccount, + ), + ) + .await? + .into_iter() + .collect::>>(); + let Some(current_accounts) = current_accounts else { + return Ok(false); + }; + if current_accounts.len() != pubkeys.len() { + return Ok(false); + } + + let same_session = recovered_accounts + .into_iter() + .zip(current_accounts) + .all(|(recovered, actual)| { + // 1. If account delegated using remote_slot we ensure + // that account is within same delegation "session" and wasn't redelegated + // 2. We handle `undelegating` state as well because undelegation could fail + // If that is the case we want to retry undelegate intent + // We need to make sure that is within same session as well + // NOTE: here we rely `validate_not_delegated` & `should_refresh_undelegating_in_bank_account` + // If account delegated - remote_slot doesn't change + // If account undelegating - until it is remote_slot doesn't change + (actual.delegated() || actual.undelegating()) + && recovered.remote_slot == actual.remote_slot() + }); + Ok(same_session) + } + + /// False if any committed account's persisted `commit_id` has already + /// been consumed on chain, or couldn't be verified. + async fn is_valid_nonce( + &self, + pubkeys: &[Pubkey], + recovered: &RecoveredIntent, + ) -> TaskInfoFetcherResult { + let current_nonces = self + .processor + .fetch_current_commit_nonces(pubkeys, 0) + .await?; + + let valid = recovered.commit_ids.iter().all(|(pubkey, commit_id)| { + current_nonces + .get(pubkey) + .is_some_and(|current_nonce| commit_id > current_nonce) }); + Ok(valid) } } From cdfd8d0d297d78385549dd1cf5290da6eb93626d Mon Sep 17 00:00:00 2001 From: taco-paco Date: Wed, 29 Jul 2026 18:43:43 +0300 Subject: [PATCH 3/6] fix: use up to date context slot --- .../src/committor_processor.rs | 8 +++- magicblock-committor-service/src/service.rs | 39 ++++++++++++------- 2 files changed, 32 insertions(+), 15 deletions(-) diff --git a/magicblock-committor-service/src/committor_processor.rs b/magicblock-committor-service/src/committor_processor.rs index 2f491d1bc..871118b9f 100644 --- a/magicblock-committor-service/src/committor_processor.rs +++ b/magicblock-committor-service/src/committor_processor.rs @@ -6,7 +6,7 @@ use std::{ }; use futures_util::future::join_all; -use magicblock_core::traits::ActionsCallbackScheduler; +use magicblock_core::{traits::ActionsCallbackScheduler, Slot}; use magicblock_program::magic_scheduled_base_intent::ScheduledIntentBundle; use magicblock_rpc_client::MagicblockRpcClient; use magicblock_table_mania::{GarbageCollectorConfig, TableMania}; @@ -181,6 +181,10 @@ impl CommittorProcessor { Ok(recovered) } + pub(crate) async fn get_slot(&self) -> CommittorServiceResult { + Ok(self.magic_rpc_client.get_slot().await?) + } + /// Stamps `payer` and `remote_slot` on recovered bundles with current /// values so execution (fetching nonces/base accounts) uses a fresh /// `min_context_slot`. Must run only on bundles that already survived @@ -190,9 +194,9 @@ impl CommittorProcessor { pub(crate) async fn refresh_intent_bundles( &self, intent_bundles: &mut [ScheduledIntentBundle], + slot: u64, ) -> CommittorServiceResult<()> { let payer = self.authority.pubkey(); - let slot = self.magic_rpc_client.get_slot().await?; macro_rules! set_remote_slot { ($field:expr, $slot:expr) => { diff --git a/magicblock-committor-service/src/service.rs b/magicblock-committor-service/src/service.rs index 05cbe71bf..48612caca 100644 --- a/magicblock-committor-service/src/service.rs +++ b/magicblock-committor-service/src/service.rs @@ -185,26 +185,33 @@ where async fn reschedule_pending_bundles(&self) -> CommittorServiceResult<()> { // Fetch pending and failed bundles from DB, unfiltered - let recovered = self - .processor - .load_recovery_intent_bundles() - .await - .inspect_err(|err| { - error!(error = ?err, "Failed to load pending intent bundles for recovery"); - })?; + let slot = self.processor.get_slot(); + let recovered = self.processor.load_recovery_intent_bundles(); + let (slot, recovered) = { + let (slot, recovered) = tokio::join!(slot, recovered); + let slot = slot?; + let recovered = recovered + .inspect_err(|err| { + error!(error = ?err, "Failed to load pending intent bundles for recovery"); + })?; + (slot, recovered) + }; if recovered.is_empty() { return Ok(()); } // Drop bundles that are no longer safe to replay - let mut bundles = self.retain_recoverable_intents(recovered).await; + let mut bundles = + self.retain_recoverable_intents(recovered, slot).await; if bundles.is_empty() { return Ok(()); } // Bundles are out of date and missing some info now that they // survived the recovery filters, which needed the original values. - self.processor.refresh_intent_bundles(&mut bundles).await?; + self.processor + .refresh_intent_bundles(&mut bundles, slot) + .await?; // Schedule without initial persisitance as bundle already exists in db self.process_intent_bundles(bundles, |bundles| { @@ -390,13 +397,14 @@ where async fn retain_recoverable_intents( &self, recovered: Vec, + min_context_slot: u64, ) -> Vec { const JOIN_CHUNK_SIZE: usize = 10; let mut to_keep = Vec::with_capacity(recovered.len()); for chunk in recovered.chunks(JOIN_CHUNK_SIZE) { let iter = chunk.iter().map(|recovered| async move { - self.is_intent_retriable(recovered).await + self.is_intent_retriable(recovered, min_context_slot).await }); to_keep.extend(join_all(iter).await); } @@ -411,7 +419,11 @@ where /// An intent is retriable only if its delegation session is unchanged /// and its commit nonce hasn't already landed on chain. - async fn is_intent_retriable(&self, recovered: &RecoveredIntent) -> bool { + async fn is_intent_retriable( + &self, + recovered: &RecoveredIntent, + min_context_slot: u64, + ) -> bool { let pubkeys = recovered.bundle.get_all_committed_pubkeys(); let result = self .is_same_delegation_session(&pubkeys, recovered) @@ -424,7 +436,7 @@ where if !result { false } else { - self.is_valid_nonce(&pubkeys, recovered) + self.is_valid_nonce(&pubkeys, recovered, min_context_slot) .await .inspect_err(|err| { error!(intent_id = recovered.bundle.id, error = ?err, "Failed to check commit nonce for recovery"); @@ -483,10 +495,11 @@ where &self, pubkeys: &[Pubkey], recovered: &RecoveredIntent, + min_context_slot: u64, ) -> TaskInfoFetcherResult { let current_nonces = self .processor - .fetch_current_commit_nonces(pubkeys, 0) + .fetch_current_commit_nonces(pubkeys, min_context_slot) .await?; let valid = recovered.commit_ids.iter().all(|(pubkey, commit_id)| { From 7583688adf51bbe8b15da27a038a02bb4f9fddfe Mon Sep 17 00:00:00 2001 From: taco-paco Date: Wed, 29 Jul 2026 18:53:00 +0300 Subject: [PATCH 4/6] fix: relax commmit nonce checks. Allows same intent committed twice but covers 2 stage execution --- magicblock-committor-service/src/service.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/magicblock-committor-service/src/service.rs b/magicblock-committor-service/src/service.rs index 48612caca..b20e5a724 100644 --- a/magicblock-committor-service/src/service.rs +++ b/magicblock-committor-service/src/service.rs @@ -505,7 +505,7 @@ where let valid = recovered.commit_ids.iter().all(|(pubkey, commit_id)| { current_nonces .get(pubkey) - .is_some_and(|current_nonce| commit_id > current_nonce) + .is_some_and(|current_nonce| commit_id >= current_nonce) }); Ok(valid) } From 9b7c417fc9578f894e52a6fb67c57e81a646e3e4 Mon Sep 17 00:00:00 2001 From: taco-paco Date: Wed, 29 Jul 2026 19:39:08 +0300 Subject: [PATCH 5/6] fix: pending intents aren't filtered as asumed to be just scheduled --- .../src/committor_processor.rs | 38 ++++++++++++++---- .../src/persist/commit_persister.rs | 39 +++++++++---------- magicblock-committor-service/src/service.rs | 34 +++++++++------- 3 files changed, 71 insertions(+), 40 deletions(-) diff --git a/magicblock-committor-service/src/committor_processor.rs b/magicblock-committor-service/src/committor_processor.rs index 871118b9f..1137a6284 100644 --- a/magicblock-committor-service/src/committor_processor.rs +++ b/magicblock-committor-service/src/committor_processor.rs @@ -154,17 +154,41 @@ impl CommittorProcessor { Ok(signatures) } - /// Fetches pending and failed bundles from DB for recovery. No filtering. + fn recovery_min_created_at() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + .saturating_sub(RECOVERY_MAX_AGE_SECS) + } + + /// Fetches pending bundles from DB for recovery. No filtering - these + /// are the most recent ones, still in-flight, not yet confirmed failed. + pub async fn load_pending_intent_bundles( + &self, + ) -> CommittorServiceResult> { + let bundles = self + .persister + .pending_intent_bundles(Self::recovery_min_created_at())?; + + if !bundles.is_empty() { + info!( + intent_count = bundles.len(), + "Loaded pending commit intents from persistence for recovery" + ); + } + + Ok(bundles) + } + + /// Fetches failed bundles from DB for recovery, for the caller to filter + /// for nonce/delegation-session staleness before replaying. pub async fn load_recovery_intent_bundles( &self, ) -> CommittorServiceResult> { - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); let recovered = self .persister - .recoverable_intents(now.saturating_sub(RECOVERY_MAX_AGE_SECS))?; + .recoverable_intents(Self::recovery_min_created_at())?; if !recovered.is_empty() { let accounts_count: usize = recovered @@ -174,7 +198,7 @@ impl CommittorProcessor { info!( intent_count = recovered.len(), accounts_count, - "Loaded commit intents from persistence for recovery" + "Loaded failed commit intents from persistence for recovery" ); } diff --git a/magicblock-committor-service/src/persist/commit_persister.rs b/magicblock-committor-service/src/persist/commit_persister.rs index 16290c760..0bce97913 100644 --- a/magicblock-committor-service/src/persist/commit_persister.rs +++ b/magicblock-committor-service/src/persist/commit_persister.rs @@ -166,20 +166,17 @@ impl IntentPersisterImpl { .collect() } - /// Loads pending/failed bundles for restart recovery, each paired with - /// the `commit_id` (nonce) its committed accounts were persisted with. - /// See [`RecoveredIntent`] for why this isn't part of the - /// `IntentPersister::pending_intent_bundles` trait method. + /// Loads *failed* bundles for restart recovery, paired with the + /// `commit_id` each committed account was persisted with. pub fn recoverable_intents( &self, min_created_at: u64, ) -> CommitPersistResult> { - let commits_db = self.commits_db.lock().expect(POISONED_MUTEX_MSG); - let mut rows = - commits_db.get_pending_commit_statuses(min_created_at)?; - rows.extend(commits_db.get_failed_commit_statuses(min_created_at)?); - drop(commits_db); - + let rows = self + .commits_db + .lock() + .expect(POISONED_MUTEX_MSG) + .get_failed_commit_statuses(min_created_at)?; Ok(pending_rows_to_recovered_intents(rows, min_created_at)) } } @@ -272,7 +269,7 @@ impl IntentPersister for IntentPersisterImpl { .get_commit_statuses_by_id(message_id) } - /// Returns pending and failed bundles created at or after `min_created_at`. + /// Returns pending bundles created at or after `min_created_at`. /// NOTE: this constructs `ScheduleIntentBundle` only from existing information. /// As persister doesn't save `ScheduleIntentBundle::payer` info, Pubkey::default is used. /// `CommittedAccount` information like slot may also be outdated. @@ -281,11 +278,11 @@ impl IntentPersister for IntentPersisterImpl { &self, min_created_at: u64, ) -> CommitPersistResult> { - let commits_db = self.commits_db.lock().expect(POISONED_MUTEX_MSG); - let mut rows = - commits_db.get_pending_commit_statuses(min_created_at)?; - rows.extend(commits_db.get_failed_commit_statuses(min_created_at)?); - drop(commits_db); + let rows = self + .commits_db + .lock() + .expect(POISONED_MUTEX_MSG) + .get_pending_commit_statuses(min_created_at)?; Ok(pending_rows_to_scheduled_intent_bundles( rows, @@ -818,15 +815,17 @@ mod tests { let pubkey_a = Pubkey::new_unique(); let pubkey_b = Pubkey::new_unique(); + let mut row_a = pending_row(1, pubkey_a, owner, blockhash, false, None); + row_a.commit_status = CommitStatus::FailedProcess(None); + let mut row_b = pending_row(1, pubkey_b, owner, blockhash, false, None); + row_b.commit_status = CommitStatus::FailedProcess(None); + let (persister, _temp_file) = create_test_persister(); persister .commits_db .lock() .unwrap() - .insert_commit_status_rows(&[ - pending_row(1, pubkey_a, owner, blockhash, false, None), - pending_row(1, pubkey_b, owner, blockhash, false, None), - ]) + .insert_commit_status_rows(&[row_a, row_b]) .unwrap(); persister.set_commit_id(1, &pubkey_a, 5).unwrap(); persister.set_commit_id(1, &pubkey_b, 7).unwrap(); diff --git a/magicblock-committor-service/src/service.rs b/magicblock-committor-service/src/service.rs index b20e5a724..851326688 100644 --- a/magicblock-committor-service/src/service.rs +++ b/magicblock-committor-service/src/service.rs @@ -183,26 +183,34 @@ where } } + /// Fetches and reschedules intents safe for retry + /// Note: this doesn't provide intent durability but is a best effort to retry intents async fn reschedule_pending_bundles(&self) -> CommittorServiceResult<()> { // Fetch pending and failed bundles from DB, unfiltered - let slot = self.processor.get_slot(); - let recovered = self.processor.load_recovery_intent_bundles(); - let (slot, recovered) = { - let (slot, recovered) = tokio::join!(slot, recovered); + let (slot, pending, failed) = { + let (slot, pending, failed) = tokio::join!( + self.processor.get_slot(), + self.processor.load_pending_intent_bundles(), + self.processor.load_recovery_intent_bundles(), + ); let slot = slot?; - let recovered = recovered - .inspect_err(|err| { - error!(error = ?err, "Failed to load pending intent bundles for recovery"); - })?; - (slot, recovered) + let pending = pending.inspect_err(|err| { + error!(error = ?err, "Failed to load pending intent bundles for recovery"); + })?; + let failed = failed.inspect_err(|err| { + error!(error = ?err, "Failed to load failed intent bundles for recovery"); + })?; + (slot, pending, failed) }; - if recovered.is_empty() { + if pending.is_empty() && failed.is_empty() { return Ok(()); } - // Drop bundles that are no longer safe to replay - let mut bundles = - self.retain_recoverable_intents(recovered, slot).await; + // Pending bundles go through as-is; failed ones must prove they're + // still safe to replay first. Unite and keep execution order stable. + let mut bundles = self.retain_recoverable_intents(failed, slot).await; + bundles.extend(pending); + bundles.sort_by_key(|b| b.id); if bundles.is_empty() { return Ok(()); } From 12d90414c7853a6930f247160d7c9a748206da25 Mon Sep 17 00:00:00 2001 From: taco-paco Date: Wed, 29 Jul 2026 19:50:13 +0300 Subject: [PATCH 6/6] fix: coderabbit, tests --- .../src/persist/commit_persister.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/magicblock-committor-service/src/persist/commit_persister.rs b/magicblock-committor-service/src/persist/commit_persister.rs index 0bce97913..591258c8a 100644 --- a/magicblock-committor-service/src/persist/commit_persister.rs +++ b/magicblock-committor-service/src/persist/commit_persister.rs @@ -818,7 +818,11 @@ mod tests { let mut row_a = pending_row(1, pubkey_a, owner, blockhash, false, None); row_a.commit_status = CommitStatus::FailedProcess(None); let mut row_b = pending_row(1, pubkey_b, owner, blockhash, false, None); - row_b.commit_status = CommitStatus::FailedProcess(None); + row_b.commit_status = + CommitStatus::FailedFinalize(CommitStatusSignatures { + commit_stage_signature: Signature::new_unique(), + finalize_stage_signature: None, + }); let (persister, _temp_file) = create_test_persister(); persister