Skip to content

hotfix: load failed intents on restart & reschedule #1371

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

Open
taco-paco wants to merge 1 commit into
base: master
Choose a base branch
from
Open

hotfix: load failed intents on restart & reschedule #1371

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
40 changes: 21 additions & 19 deletions magicblock-committor-service/src/committor_processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,19 +174,21 @@ impl CommittorProcessor {
Ok(signatures)
}

pub async fn pending_intent_bundles(
pub async fn load_recovery_intent_bundles(
&self,
) -> CommittorServiceResult<Vec<ScheduledIntentBundle>> {
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)?);
Comment on lines +183 to +185

@coderabbitai coderabbitai Bot Jun 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift

Don't union independently filtered recovery rows.

Line 184 and Line 185 can recover a partial bundle. If one message_id has both Pending and Failed* rows with different created_at values, one query can include it while the other excludes it, and rows_to_scheduled_intent_bundles then rebuilds only the surviving subset of accounts. Compute recovery eligibility once across the full recoverable-status set and load all eligible rows together.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@magicblock-committor-service/src/committor_processor.rs` around lines 183 -
185, The recovery path in committor_processor’s row loading is incorrectly
unioning separately filtered Pending and Failed* results, which can reconstruct
only part of a bundle for the same message_id. Update the logic around
self.persister.get_pending_commit_statuses and
self.persister.get_failed_commit_statuses so recovery eligibility is computed
once across the full recoverable-status set, then load all rows for eligible
message_ids together before passing them to rows_to_scheduled_intent_bundles.

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,
Expand All @@ -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"
);
}

Expand Down Expand Up @@ -263,7 +265,7 @@ impl CommittorProcessor {
}
}

fn pending_rows_to_scheduled_intent_bundles(
fn rows_to_scheduled_intent_bundles(
rows: Vec<CommitStatusRow>,
payer: Pubkey,
recovery_base_slot: u64,
Expand All @@ -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;
}
Expand All @@ -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;
}
Expand All @@ -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;
};
Expand Down Expand Up @@ -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 {
Expand All @@ -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)> {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -513,15 +515,15 @@ 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,
recovery_time,
);
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,
Expand Down
26 changes: 26 additions & 0 deletions magicblock-committor-service/src/persist/commit_persister.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,10 @@ pub trait IntentPersister: Send + Sync + Clone + 'static {
&self,
min_created_at: u64,
) -> CommitPersistResult<Vec<CommitStatusRow>>;
fn get_failed_commit_statuses(
&self,
min_created_at: u64,
) -> CommitPersistResult<Vec<CommitStatusRow>>;
fn get_commit_status_by_message(
&self,
message_id: u64,
Expand Down Expand Up @@ -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<Vec<CommitStatusRow>> {
self.commits_db
.lock()
.expect(POISONED_MUTEX_MSG)
.get_failed_commit_statuses(min_created_at)
}
Comment on lines +264 to +272

@coderabbitai coderabbitai Bot Jun 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Avoid panicking on the DB mutex in this new method.

Line 270 uses .expect(POISONED_MUTEX_MSG), so a poisoned mutex aborts the committor instead of returning a persistence error from this recovery path. Please map the lock failure into CommitPersistResult or justify the panic invariant explicitly. As per path instructions, {magicblock-*,programs,storage-proto}/**: Treat any usage of .unwrap() or .expect() in production Rust code as a MAJOR issue.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@magicblock-committor-service/src/persist/commit_persister.rs` around lines
264 - 272, The new recovery helper get_failed_commit_statuses currently panics
on a poisoned DB mutex via .expect(POISONED_MUTEX_MSG), which violates the
non-panicking persistence path requirement. Update get_failed_commit_statuses in
commit_persister.rs to handle the lock acquisition failure explicitly by mapping
it into a CommitPersistResult error instead of aborting the committor. Reuse the
existing CommitPersistResult and commits_db access pattern used by the other
persistence methods so the failure is reported consistently, and only keep a
panic if you add an explicit invariant comment justifying it.

Source: Path instructions


fn get_commit_status_by_message(
&self,
message_id: u64,
Expand Down Expand Up @@ -415,6 +429,18 @@ impl<T: IntentPersister> IntentPersister for Option<T> {
}
}

fn get_failed_commit_statuses(
&self,
min_created_at: u64,
) -> CommitPersistResult<Vec<CommitStatusRow>> {
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,
Expand Down
128 changes: 128 additions & 0 deletions magicblock-committor-service/src/persist/db.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -585,6 +585,35 @@ impl CommittsDb {
extract_committor_rows(&mut rows)
}

pub(crate) fn get_failed_commit_statuses(
&self,
min_created_at: u64,
) -> CommitPersistResult<Vec<CommitStatusRow>> {
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,
Expand Down Expand Up @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion magicblock-committor-service/src/service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
}
Expand Down
Loading