diff --git a/magicblock-committor-service/src/intent_engine/intent_execution_engine.rs b/magicblock-committor-service/src/intent_engine/intent_execution_engine.rs index e78e51d91..7972bd917 100644 --- a/magicblock-committor-service/src/intent_engine/intent_execution_engine.rs +++ b/magicblock-committor-service/src/intent_engine/intent_execution_engine.rs @@ -85,6 +85,17 @@ impl BroadcastedIntentExecutionResult { .successful_transaction_strategies, } } + + fn poisoned(id: u64) -> Self { + Self { + inner: Err(IntentExecutorError::PoisonedIntentError.into()), + id, + patched_errors: Arc::default(), + callbacks_report: Vec::new(), + #[cfg(feature = "dev-context-only-utils")] + successful_transaction_strategies: vec![], + } + } } impl Deref for BroadcastedIntentExecutionResult { @@ -248,10 +259,19 @@ where }, else => { // Shouldn't be possible: - // 1. If no executors spawned -> we can receive - // 2. If can't receive -> there are MAX_EXECUTORS running executors - // We can't receive new message as there's no available Executor - // that could pick up the task. + // 1. If no executors spawned -> nothing occupies any + // pubkey's front -> blocked_intents is empty -> we can receive + // 2. If can't receive -> blocked_intents is non-empty -> + // something occupies a front -> an executor is running + // + // running_executors isn't 1-1 with the executors semaphore: + // Self::execute drops its permit while sleeping between + // retries, so a retrying intent still holds a slot here + // without holding a permit. That's fine for (2) - it still + // counts as "an executor is running" for as long as it + // hasn't given up. And if it does give up, failed() evicts + // the whole poisoned cascade from blocked_intents instead + // of leaving it stuck, so capacity always frees up. unreachable!("next_scheduled_intent") } }; @@ -297,34 +317,74 @@ where error!(intent_id = intent.intent_id, error = ?err, "Failed to execute intent bundle"); }).is_err(); Self::execution_metrics(instant.elapsed(), &intent, &result.inner); - let broadcasted_result = - BroadcastedIntentExecutionResult::new(intent.intent_id, result); - if let Err(err) = result_sender.send(broadcasted_result) { - warn!(error = ?err, "No result listeners"); - } - - // Lock intent's pubkeys. This will prevent future intents from execution - // if future intent overlaps with failed one(current) + let mut scheduler = inner_scheduler.lock().expect(POISONED_INNER_MSG); if is_err { + // Poison this intent's pubkeys and evict any successor that's + // reachable from them, so a terminal failure can't leave the + // scheduler permanently stuck on dead accounts. let pubkeys = intent.get_all_committed_pubkeys(); - warn!(pubkeys = ?pubkeys, "Intents for following pubkeys are now blocked."); - return; + warn!(pubkeys = ?pubkeys, "Intents for following pubkeys are now permanently poisoned for the life of this process."); + // SAFETY: Self::execute is called ONLY after IntentScheduler + // successfully is able to schedule execution of some Intent + // that means that the same Intent is SAFE to mark as failed + let poisoned_intents = scheduler + .failed(&intent) + .expect("Valid completion of previously scheduled message"); + metrics::set_committor_poisoned_keys_count( + scheduler.poisoned_keys_count() as i64, + ); + drop(scheduler); + Self::broadcast_result( + BroadcastedIntentExecutionResult::new(intent.intent_id, result), + &result_sender, + ); + Self::report_poisoned_intents(poisoned_intents, &result_sender); + } else { + // Remove executed task from Scheduler to unblock other intents + // SAFETY: Self::execute is called ONLY after IntentScheduler + // successfully is able to schedule execution of some Intent + // that means that the same Intent is SAFE to complete + scheduler + .complete(&intent) + .expect("Valid completion of previously scheduled message"); + drop(scheduler); + Self::broadcast_result( + BroadcastedIntentExecutionResult::new(intent.intent_id, result), + &result_sender, + ); } - // That would - // Remove executed task from Scheduler to unblock other intents - // SAFETY: Self::execute is called ONLY after IntentScheduler - // successfully is able to schedule execution of some Intent - // that means that the same Intent is SAFE to complete - inner_scheduler - .lock() - .expect(POISONED_INNER_MSG) - .complete(&intent) - .expect("Valid completion of previously scheduled message"); // Free worker drop(execution_permit); } + fn broadcast_result( + broadcasted_result: BroadcastedIntentExecutionResult, + result_sender: &broadcast::Sender, + ) { + if result_sender.receiver_count() != 0 + && let Err(err) = result_sender.send(broadcasted_result) + { + warn!(error = ?err, "No result listeners"); + }; + } + + fn report_poisoned_intents( + poisoned_intents: Vec, + result_sender: &broadcast::Sender, + ) { + metrics::inc_committor_cascade_voided_intents_count_by( + poisoned_intents.len() as u64, + ); + for intent in poisoned_intents { + warn!(poisoned_intent = ?intent.intent_id, "Intent poisoned"); + Self::broadcast_result( + BroadcastedIntentExecutionResult::poisoned(intent.intent_id), + result_sender, + ); + } + } + /// Executes an intent, retrying plausibly-transient failures with a /// fresh executor. Returns the final result together with whichever /// execution permit is still held (`None` only if the executors @@ -457,7 +517,7 @@ mod tests { use solana_signature::Signature; use solana_signer::SignerError; use solana_transaction_error::TransactionError; - use tokio::time::sleep; + use tokio::time::{sleep, timeout}; use super::*; use crate::{ @@ -707,26 +767,126 @@ mod tests { ); } - /// Tests that errors from executor propagated gracefully + /// A failed head reports its own real error, and every successor + /// voided by the poisoning cascade gets its own `PoisonedIntentError` + /// report - not silence. Previously the cascaded successors were left + /// in `blocked_intents` forever with no report at all, which hung this + /// exact test (10 intents sharing one pubkey, waiting for 10 results) + /// indefinitely. The bounded `timeout` here turns any regression back + /// into a fast failure instead of another indefinite hang. #[tokio::test] async fn test_multiple_failures() { let (sender, worker, _db) = setup_engine(true); // Worker that always fails let result_subscriber = worker.spawn(); let mut result_receiver = result_subscriber.subscribe(); - // Send several messages that will fail + // All intents share one pubkey: the head executes and fails, and + // the cascade voids every successor queued behind it. const NUM_FAILURES: usize = 10; + let pubkey = pubkey!("1111111111111111111111111111111111111111111"); for i in 0..NUM_FAILURES { - let msg = - create_test_intent(i as u64, &[Pubkey::new_unique()], false); + let msg = create_test_intent(i as u64, &[pubkey], false); sender.schedule(vec![msg]).unwrap(); } - // Verify all failures are processed and semaphore slots released + let mut results = Vec::with_capacity(NUM_FAILURES); for _ in 0..NUM_FAILURES { - let result = result_receiver.recv().await.unwrap(); + let result = timeout( + Duration::from_secs(5), + result_receiver.recv(), + ) + .await + .expect( + "must not hang waiting for poisoned successors to be reported", + ) + .unwrap(); + results.push(result); + } + + // The head (id 0) executed and reports its own real error. + let head = results.iter().find(|r| r.id == 0).unwrap(); + assert!( + matches!( + head.inner.as_ref().unwrap_err().as_ref(), + ExecutorError::FailedToCommitError { .. } + ), + "head should report the real executor error, got {:?}", + head.inner + ); + + // Every other intent was queued behind the head, never executed, + // and was voided by the cascade instead. + for id in 1..NUM_FAILURES as u64 { + let voided = results.iter().find(|r| r.id == id).unwrap(); + assert!( + matches!( + voided.inner.as_ref().unwrap_err().as_ref(), + ExecutorError::PoisonedIntentError + ), + "intent {id} should be reported as poisoned, got {:?}", + voided.inner + ); + } + } + + /// A poisoned pubkey rejects future intents silently (no broadcast - + /// they never entered the scheduler), while unrelated pubkeys keep + /// executing and reporting normally, proving poisoning doesn't leak + /// across unrelated accounts. + #[tokio::test] + async fn test_poisoned_pubkey_rejects_future_intents_silently() { + let (sender, worker, _db) = setup_engine(true); // Worker that always fails + let result_subscriber = worker.spawn(); + let mut result_receiver = result_subscriber.subscribe(); + + let poisoned_pubkey = + pubkey!("1111111111111111111111111111111111111111111"); + let head = create_test_intent(0, &[poisoned_pubkey], false); + let successor = create_test_intent(1, &[poisoned_pubkey], false); + sender.schedule(vec![head]).unwrap(); + sender.schedule(vec![successor]).unwrap(); + + // Head fails for real, successor is voided by the cascade. + for _ in 0..2 { + let result = + timeout(Duration::from_secs(5), result_receiver.recv()) + .await + .expect("must not hang") + .unwrap(); assert!(result.is_err()); } + + // A brand new intent on the now-poisoned pubkey is rejected at + // admission - it never reaches an executor, so no broadcast for + // it ever arrives. + let rejected = create_test_intent(2, &[poisoned_pubkey], false); + sender.schedule(vec![rejected]).unwrap(); + let silence = + timeout(Duration::from_millis(300), result_receiver.recv()).await; + assert!( + silence.is_err(), + "poisoned pubkey must not produce a broadcast for a rejected intent" + ); + + // An intent on an unrelated pubkey is unaffected by the poisoning + // and still executes (and fails for real, not as poisoned). + let unrelated_pubkey = + pubkey!("21111111111111111111111111111111111111111111"); + let unrelated = create_test_intent(3, &[unrelated_pubkey], false); + sender.schedule(vec![unrelated]).unwrap(); + let result = timeout(Duration::from_secs(5), result_receiver.recv()) + .await + .expect("must not hang") + .unwrap(); + assert_eq!(result.id, 3); + assert!( + matches!( + result.inner.as_ref().unwrap_err().as_ref(), + ExecutorError::FailedToCommitError { .. } + ), + "unrelated pubkey should fail for real, not as poisoned: {:?}", + result.inner + ); } /// Transient failures are retried with a fresh executor until success diff --git a/magicblock-committor-service/src/intent_engine/intent_scheduler.rs b/magicblock-committor-service/src/intent_engine/intent_scheduler.rs index 6b28075aa..c1a7cbf63 100644 --- a/magicblock-committor-service/src/intent_engine/intent_scheduler.rs +++ b/magicblock-committor-service/src/intent_engine/intent_scheduler.rs @@ -1,9 +1,9 @@ -use std::collections::{HashMap, VecDeque, hash_map::Entry}; +use std::collections::{BTreeSet, HashMap, HashSet, VecDeque, hash_map::Entry}; use magicblock_program::outbox_intent_bundles::OutboxIntentBundle; use solana_pubkey::Pubkey; use thiserror::Error; -use tracing::error; +use tracing::{error, warn}; pub(crate) const POISONED_INNER_MSG: &str = "Mutex on CommitSchedulerInner is poisoned."; @@ -26,9 +26,17 @@ struct IntentMeta { /// - Key: IntentID /// - Value: Intent metadata including original intent /// +/// 3. `poisoned_keys`: Pubkeys touched by a failed or voided intent - once +/// poisoned, a pubkey rejects every future intent for the lifetime of +/// this scheduler (only a process restart clears it, since the outbox +/// will naturally retry whatever was poisoned) +/// - Key: Pubkey +/// /// # Scheduling Logic /// /// 1. On intent arrival: +/// - Check if any required pubkey is poisoned: if so, poison the rest +/// of this intent's pubkeys too and reject it (it will never execute) /// - Check if any required pubkey exists in `blocked_keys` /// - If conflicted: Add intent to all relevant pubkey queues /// - Else: Start executing immediately @@ -43,6 +51,18 @@ struct IntentMeta { /// has all of its pubkeys unblocked, /// i.e they are first at corresponding queues /// +/// 4. On intent failure: +/// - Poison all of the failed intent's pubkeys, then walk every +/// successor reachable from them (transitively, via shared pubkeys) +/// and void it too - each voided intent poisons its own full pubkey +/// set the same way, so the cascade can't stop partway through a +/// dependency chain +/// - Intents that merely share a pubkey but aren't reachable (queued +/// *before* the point the cascade reaches, or on an unrelated key) +/// are untouched and keep executing normally +/// - See `poisoned_test` for the full algorithm writeup and worked +/// examples +/// /// Some examples/edge cases: /// (1) Assume `t1`: /// executing: `[a1, a2, a3] [b1, b2, b3]` - 1 @@ -64,6 +84,7 @@ struct IntentMeta { pub(crate) struct IntentScheduler { blocked_keys: HashMap>, blocked_intents: HashMap, + poisoned_keys: HashSet, } impl IntentScheduler { @@ -71,21 +92,25 @@ impl IntentScheduler { Self { blocked_keys: HashMap::new(), blocked_intents: HashMap::new(), + poisoned_keys: HashSet::new(), } } /// Returns [`ScheduledIntentBundle`] if intent can be executed, /// otherwise consumes it and enqueues + // TODO(edwin): tweak return type to reflect Poisoned, ScheduleResult pub fn schedule( &mut self, intent_bundle: OutboxIntentBundle, ) -> Option { let intent_id = intent_bundle.intent_id; + let pubkeys = intent_bundle.get_all_committed_pubkeys(); + if pubkeys.is_empty() { + return Some(intent_bundle); + }; - // To check duplicate scheduling its enough to check: - // 1. currently blocked - // 2. currently executing - if self.blocked_intents.contains_key(&intent_id) { + // Check that id is not duplicated + if self.is_duplicate(intent_id) { // This is critical error as we shouldn't schedule duplicate Intents! // this requires investigation error!( @@ -94,28 +119,21 @@ impl IntentScheduler { ); return None; } - let duplicate_executing = self.blocked_keys.iter().any(|(_, queue)| { - if let Some(executing_id) = queue.front() { - &intent_id == executing_id - } else { - false - } - }); - if duplicate_executing { - // This is critical error as we shouldn't schedule duplicate Intents! - // this requires investigation - error!( + + // Check if intent is poisoned by existing poisonous keys + let is_poisoned = + pubkeys.iter().any(|el| self.poisoned_keys.contains(el)); + if is_poisoned { + // Intent got poisoned by others + warn!( intent_id, - "CRITICAL! Attempt to schedule already scheduled intent" + pubkeys = ?pubkeys, + "Intent got poisoned" ); + self.poisoned_keys.extend(pubkeys); return None; } - let pubkeys = intent_bundle.get_all_committed_pubkeys(); - if pubkeys.is_empty() { - return Some(intent_bundle); - }; - // Check if there are any conflicting keys let is_conflicting = pubkeys .iter() @@ -143,21 +161,32 @@ impl IntentScheduler { } } - /// Completes Intent, cleaning up data after itself and allowing Intents to move forward - /// NOTE: This doesn't unblock intent, hence Self::intents_blocked will return old value. - /// NOTE: this shall be called on executing intents to finilize their execution. - pub fn complete( - &mut self, - intent_bundle: &OutboxIntentBundle, - ) -> IntentSchedulerResult<()> { - // Release data for completed intent - let intent_id = intent_bundle.intent_id; - let pubkeys = intent_bundle.get_all_committed_pubkeys(); - if pubkeys.is_empty() { - // This means BaseAction, it doesn't have to be scheduled - return Ok(()); - }; + /// Returns if same IntentId is scheduled + /// To check duplicate scheduling its enough to check: + /// 1. currently blocked + /// 2. currently executing + /// + /// NOTE: under assumption that outer system doesn't schedule duplcates + /// this can be ommitted to reduce execution time + fn is_duplicate(&self, intent_id: IntentID) -> bool { + if self.blocked_intents.contains_key(&intent_id) { + true + } else { + self.blocked_keys.iter().any(|(_, queue)| { + if let Some(executing_id) = queue.front() { + &intent_id == executing_id + } else { + false + } + }) + } + } + fn validate_executing( + &self, + intent_id: IntentID, + pubkeys: &[Pubkey], + ) -> IntentSchedulerResult<()> { if self.blocked_intents.contains_key(&intent_id) { return Err(IntentSchedulerError::CompletingBlockedIntentError); } @@ -166,7 +195,7 @@ impl IntentScheduler { let mut all_front = true; // Some of front queues contain intent id let mut some_front = false; - for pubkey in &pubkeys { + for pubkey in pubkeys { if let Some(blocked_intents) = self.blocked_keys.get(pubkey) { // SAFETY: if entry exists it means that queue not empty // This is ensured during scheduling as we always insert el-t in the queue @@ -206,8 +235,29 @@ impl IntentScheduler { .filter(|(_, queue)| queue.front() == Some(&intent_id)) .count(); if found_in_front != pubkeys.len() { - return Err(IntentSchedulerError::CorruptedIntentError); + Err(IntentSchedulerError::CorruptedIntentError) + } else { + Ok(()) } + } + + /// Completes Intent, cleaning up data after itself and allowing Intents to move forward + /// NOTE: This doesn't unblock intent, hence Self::intents_blocked will return old value. + /// NOTE: this shall be called on executing intents to finalize their execution. + pub fn complete( + &mut self, + intent_bundle: &OutboxIntentBundle, + ) -> IntentSchedulerResult<()> { + // Release data for completed intent + let intent_id = intent_bundle.intent_id; + let pubkeys = intent_bundle.get_all_committed_pubkeys(); + if pubkeys.is_empty() { + // This means BaseAction, it doesn't have to be scheduled + return Ok(()); + }; + + // Validate that requested intent is executing indeed + self.validate_executing(intent_id, &pubkeys)?; // After all the checks we may safely complete pubkeys.iter().for_each(|pubkey| { @@ -232,6 +282,219 @@ impl IntentScheduler { Ok(()) } + /// Marks `intent_bundle` as failed: poisons its pubkeys, voids every + /// intent reachable from it, and returns the voided intents. + /// + /// For curious below is AI generated slop from my notes and thoughts + /// # Poisoning: how a failed intent's dependents are found and voided + /// + /// ## Why + /// + /// When an executing intent fails, everything queued behind it may be + /// relying on effects that never happened. If we leave those successors + /// sitting in `blocked_intents` forever, [`Self::pop_next_scheduled_intent`] + /// can never make progress on the pubkeys they hold, `intents_blocked()` + /// never shrinks, and (upstream, in the engine) the scheduler eventually + /// panics once capacity is exhausted and no executor remains to drain it. + /// This exists to unstick that: it finds every intent that can no longer + /// safely run, removes it from the scheduler, and reports it so a fresh + /// attempt can be made later (nothing is discarded — the intent's outbox + /// record is untouched and gets picked up again by the normal recovery + /// scan, e.g. on restart). + /// + /// ## Definitions + /// + /// - **Dependency.** Intent `B` depends on intent `A` if `B` was queued + /// behind `A` because they share a pubkey. The FIFO blocking scheme + /// exists precisely because `B` may assume `A`'s effects on that + /// pubkey already landed — e.g. `A` transfers funds *into* an account, + /// `B` spends *from* it. + /// - **Reachable.** `B` is reachable from `A` if there is a chain of + /// dependencies `A -> X1 -> X2 -> ... -> B`, where each arrow is one + /// "queued directly behind, on a shared pubkey" edge. Sharing a pubkey + /// alone is not enough — the edge only exists in the successor + /// direction (see the worked example below, where `I1` shares `a2` + /// with the chain but is *not* reachable from it). + /// - **Isolated.** Two intents are isolated from each other if neither + /// is reachable from the other. Isolated intents may still share a + /// pubkey; they just never depend on one another through it. + /// - **Voided intent.** An intent reachable from a failed intent. It + /// will never execute, is removed from `blocked_intents`, and is + /// returned by `failed()`. + /// - **Poisoned key.** A pubkey no longer eligible for *new* scheduling + /// (`schedule()` rejects any incoming intent that touches it). A key + /// is poisoned either because an intent that actually executed + /// failed on it, or because a voided intent touched it — see + /// "Why poisoning is unconditional" below for why the latter applies + /// even when the key still has isolated, healthy intents on it. + /// `poisoned_keys` lives only in this `IntentScheduler` instance and + /// is cleared by process restart, not by anything within `failed()` + /// or `complete()`. + /// + /// ## "Lemma": reachable intents are ID-ordered + /// + /// *If `B` is reachable from `A`, then `A.intent_id < B.intent_id`.* + /// + /// **Handwaving Proof.** It's enough to show this for one dependency edge + /// (`A -> B` directly); the general case follows by chaining edges, + /// since `<` is transitive. `schedule()` only ever appends to the back + /// of a pubkey's queue, in call order, so a queue's contents are always + /// sorted ascending by id. `B` depends on `A` means `B` was scheduled + /// while `A` already occupied their shared pubkey's queue — i.e. `B` + /// arrived, and therefore was assigned its id, *after* `A` did. Were it + /// the other way around (`B.intent_id < A.intent_id`), `B` would have been admitted + /// to that queue first, and `A` — arriving later — would have had to + /// queue behind `B` instead, contradicting `B` depending on `A`. So + /// `A.intent_id < B.intent_id`. ∎ + /// + /// This is what makes the worklist walk in `failed()` well-founded: it + /// only ever looks *forward* (`drain(pos..)`, never backward) through a + /// queue, so it can never re-visit or accidentally cross into something + /// isolated — anything positioned before a reachable intent is, by this + /// lemma, not reachable itself. + /// + /// ## Algorithm + /// + /// Given a failed intent `F` (validated to be at the front of every one + /// of its own pubkey queues): + /// + /// 1. **Seed.** For each of `F`'s own pubkeys: mark it poisoned, and + /// remove its queue entirely. Everything behind `F` in that queue + /// (i.e. everything except `F` itself) is reachable from `F` by + /// definition — add it to the worklist. + /// 2. **Propagate.** Pop an intent `V` from the worklist and remove it + /// from `blocked_intents` (skip if already removed — reachable via + /// another edge already processed). For each of `V`'s *other* + /// pubkeys: find `V`'s position in that queue (binary search — the + /// queue is sorted, per the lemma) and drain from that position to + /// the end. Everything drained besides `V` itself is newly + /// discovered as reachable — add it to the worklist. Mark the + /// pubkey poisoned. Record `V` as voided. + /// 3. Repeat step 2 until the worklist is empty. + /// + /// Termination is immediate: each intent id is removed from + /// `blocked_intents` at most once, so the worklist strictly shrinks. + /// + /// ## Why poisoning is unconditional + /// + /// A pubkey is marked poisoned as soon as a reachable intent touches + /// it — even if, after the drain, isolated intents are still sitting + /// in its queue and will go on to `complete()` successfully. This can + /// look surprising (see the worked example: `a2` gets poisoned while + /// `I0`/`I1` are still healthily using it), but the two things it's + /// protecting are different: + /// + /// - Intents *already in the queue* are protected by voiding, which is + /// precise (isolated intents like `I1` are never touched, per the + /// lemma above). + /// - Poisoning the key protects intents *not yet submitted*. A future + /// caller may construct a new intent on `a2` assuming the un-landed + /// chain's effects already happened (e.g. assuming `I3`'s deposit + /// landed before spending from `a2`). A simple balance check would + /// just fail cleanly on-chain if that assumption is wrong — but + /// intents can carry arbitrary actions/callbacks, and there's no + /// general guarantee that arbitrary program logic fails safely + /// against unexpected state. The scheduler can't tell in advance + /// which future intents are safe, so every key touched by a voided + /// intent is treated as unsafe until a human clears it (or the + /// process restarts). + /// + /// ## Worked example + /// + /// ```text + /// I0 = [a1, a2] executing, front of a1 and a2 + /// I2 = [b1, b2] executing, front of b1 and b2 <- this one fails + /// I1 = [a1, a2] queued behind I0 + /// I3 = [a2, b1] queued behind I1 on a2, behind I2 on b1 + /// I4 = [a1, a2] queued behind I3 on a2, behind I1 on a1 + /// + /// a1 a2 b1 b2 + /// pos0: I0 I0 I2 I2 + /// pos1: I1 I1 I3 . + /// pos2: I4 I3 . . + /// pos3: . I4 . . + /// ``` + /// + /// `I2` fails. Seed: `poisoned_keys = {b1, b2}`, worklist = `{I3}` + /// (`I2` itself is discarded, not voided — it already executed). + /// + /// Processing `I3` (`[a2, b1]`): `b1` is already poisoned, its queue is + /// already gone. `a2`: `I3` is at position 2 in `[I0, I1, I3, I4]`; + /// draining from there removes `I3, I4`, leaving `[I0, I1]`; `I4` goes + /// to the worklist; `a2` is marked poisoned (even though `I0, I1` + /// remain). + /// + /// Processing `I4` (`[a1, a2]`): `a2` already poisoned/drained, `I4` no + /// longer there, skip. `a1`: `I4` is at position 2 in `[I0, I1, I4]`; + /// draining removes just `I4`, leaving `[I0, I1]`; `a1` marked + /// poisoned. + /// + /// Result: `poisoned_keys = {a1, a2, b1, b2}`, `poisoned = [I3, I4]`. + /// `I0` and `I1` are never touched, remain in `blocked_keys`, and + /// `complete()` normally — but `a1`/`a2` reject any *new* intent from + /// here on, per "why poisoning is unconditional" above. `I1` is + /// isolated from `I2`'s failure (no dependency chain reaches it — it's + /// positioned *before* `I3` on `a2`, not after), which the lemma + /// guarantees the drain-from-position walk can never touch. + pub fn failed( + &mut self, + intent_bundle: &OutboxIntentBundle, + ) -> IntentSchedulerResult> { + // Release data for completed intent + let intent_id = intent_bundle.intent_id; + let pubkeys = intent_bundle.get_all_committed_pubkeys(); + if pubkeys.is_empty() { + // This means Action only intent, it can't poison anything + return Ok(vec![]); + }; + + // Validate that requested intent is executing indeed + self.validate_executing(intent_id, &pubkeys)?; + + // Poison intents + let mut worklist = BTreeSet::new(); + for pubkey in pubkeys { + self.poisoned_keys.insert(pubkey); + // SAFETY: validate_executing just confirmed intent_id is at the + // front of every one of these pubkeys' queues, so each queue + // is non-empty and must exist in blocked_keys. + let queue = + self.blocked_keys.remove(&pubkey).expect("front-checked"); + worklist.extend(queue.into_iter().skip(1)); + } + + let mut poisoned = Vec::new(); + while let Some(intent_id) = worklist.pop_first() { + let Some(meta) = self.blocked_intents.remove(&intent_id) else { + continue; + }; + + let pubkeys = meta.intent.get_all_committed_pubkeys(); + for pubkey in &pubkeys { + let Entry::Occupied(mut val) = self.blocked_keys.entry(*pubkey) + else { + continue; + }; + let Ok(pos) = val.get_mut().binary_search(&intent_id) else { + // Queue was already drained + continue; + }; + + // Remove items in queue starting with intent_id. All following intens are poisoned + let poisoned_iter = val.get_mut().drain(pos..).skip(1); + worklist.extend(poisoned_iter); + if val.get().is_empty() { + val.remove(); + } + } + + self.poisoned_keys.extend(pubkeys); + poisoned.push(meta.intent); + } + + Ok(poisoned) + } + // Returns [`ScheduledBaseIntent`] that can be executed pub fn pop_next_scheduled_intent(&mut self) -> Option { // TODO(edwin): optimize. Create counter im IntentMeta & update @@ -281,6 +544,10 @@ impl IntentScheduler { pub fn intents_blocked(&self) -> usize { self.blocked_intents.len() } + + pub fn poisoned_keys_count(&self) -> usize { + self.poisoned_keys.len() + } } #[derive(Error, Debug)] @@ -291,6 +558,8 @@ pub enum IntentSchedulerError { CorruptedIntentError, #[error("Attempt to complete blocked message")] CompletingBlockedIntentError, + #[error("Intent touched poisoned pubkeys")] + IntentPoisonedError(Vec), } pub type IntentSchedulerResult = Result; @@ -846,6 +1115,352 @@ mod intent_bundle_test { } } +#[cfg(test)] +mod poisoned_test { + use solana_pubkey::pubkey; + + use super::*; + use crate::test_utils; + + fn setup() { + test_utils::init_test_logger(); + } + + /// # Case 1 — poison propagates through a chain, but spares an + /// unrelated *executing* intent + /// + /// ```text + /// F = [a1, a2] executing, front of a1 and a2 + /// G = [b1, b2] executing, front of b1 and b2 <- this one fails + /// X = [a2, b1] queued behind F on a2, behind G on b1 + /// Y = [a1, a2] queued behind F on a1, behind X on a2 + /// + /// a1 a2 b1 b2 + /// pos0: F F G G + /// pos1: Y X X . + /// pos2: . Y . . + /// ``` + /// + /// `G` fails. Poisoning `b1` reaches `X` (shares `b1`), which in turn + /// reaches `Y` (shares `a2`) — both get voided. `F` must be left + /// completely alone: it shares pubkeys with the voided chain, but not + /// a dependency edge — it's positioned *before* `X` on `a2`, not + /// after, so the drain-from-position walk never reaches it. + #[test] + fn test_case_1_poison_chain_spares_the_other_executing_intent() { + setup(); + let mut scheduler = IntentScheduler::new(); + let a1 = pubkey!("1111111111111111111111111111111111111111111"); + let a2 = pubkey!("21111111111111111111111111111111111111111111"); + let b1 = pubkey!("31111111111111111111111111111111111111111111"); + let b2 = pubkey!("41111111111111111111111111111111111111111111"); + + let executing_a = create_test_intent(1, &[a1, a2], false); + let executing_b = create_test_intent(2, &[b1, b2], false); + assert!(scheduler.schedule(executing_a.clone()).is_some()); + assert!(scheduler.schedule(executing_b.clone()).is_some()); + + let x = create_test_intent(3, &[a2, b1], false); + let y = create_test_intent(4, &[a1, a2], false); + assert!(scheduler.schedule(x.clone()).is_none()); + assert!(scheduler.schedule(y.clone()).is_none()); + + let poisoned = scheduler.failed(&executing_b).unwrap(); + let mut ids: Vec<_> = poisoned.iter().map(|i| i.intent_id).collect(); + ids.sort(); + assert_eq!( + ids, + vec![x.intent_id, y.intent_id], + "both blocked intents on the chain must be voided" + ); + + // The unrelated executing intent is untouched: it can still complete(). + assert!(scheduler.complete(&executing_a).is_ok()); + } + + /// # Case 2 — reachable intents are ID-ordered + /// + /// ```text + /// F = [a1, a2] executing, front of a1 and a2 + /// I1 = [b1, b2] executing, front of b1 and b2 + /// X = [a2, b1] queued behind F on a2, behind I1 on b1 + /// I2 = [a1, a2] queued behind F on a1, behind X on a2 + /// + /// a1 a2 b1 b2 + /// pos0: F F I1 I1 + /// pos1: I2 X X . + /// pos2: . I2 . . + /// ``` + /// + /// `I1` reaches `X` (shares `b1`), which reaches `I2` (shares `a2`) — + /// so `I1` and `I2` are *not* isolated from each other. Reachable + /// intents must be ID-ordered: `I1.intent_id < I2.intent_id`. + /// + /// Handwaving proof: suppose instead `I2.intent_id < I1.intent_id`. Then `I2` would + /// have been admitted first and would occupy `a2`'s queue ahead of + /// whatever comes to share it with `I1` — but `I1` is what's supposed + /// to reach `I2`, not the other way around. Contradiction. + #[test] + fn test_case_2_reachable_intents_are_id_ordered() { + setup(); + let mut scheduler = IntentScheduler::new(); + let a1 = pubkey!("1111111111111111111111111111111111111111111"); + let a2 = pubkey!("21111111111111111111111111111111111111111111"); + let b1 = pubkey!("31111111111111111111111111111111111111111111"); + let b2 = pubkey!("41111111111111111111111111111111111111111111"); + + let executing_a = create_test_intent(1, &[a1, a2], false); + let i1 = create_test_intent(2, &[b1, b2], false); + assert!(scheduler.schedule(executing_a).is_some()); + assert!(scheduler.schedule(i1.clone()).is_some()); + + let bridge = create_test_intent(3, &[a2, b1], false); + let i2 = create_test_intent(4, &[a1, a2], false); + assert!(scheduler.schedule(bridge).is_none()); + assert!(scheduler.schedule(i2.clone()).is_none()); + + assert!( + i1.intent_id < i2.intent_id, + "reachable intents must be ID-ordered" + ); + + // Empirically: I2 is reachable from I1 (not isolated) - failing I1 + // must void I2 too. + let poisoned = scheduler.failed(&i1).unwrap(); + assert!(poisoned.iter().any(|p| p.intent_id == i2.intent_id)); + } + + /// # Case 3 — which intents to exclude + /// + /// ```text + /// F = [a1, a2] executing, front of a1 and a2 + /// I1 = [b1, b2] executing, front of b1 and b2 <- fails + /// I2 = [a1, a2] queued behind F on a1 and a2 + /// I3 = [a2, b1] queued behind I2 on a2, behind I1 on b1 + /// I4 = [a1, a2] queued behind I2 on a1, behind I3 on a2 + /// + /// a1 a2 b1 b2 + /// pos0: F F I1 I1 + /// pos1: I2 I2 I3 . + /// pos2: I4 I3 . . + /// pos3: . I4 . . + /// ``` + /// + /// `I1` fails: `[a1, a2, b1, b2]` all become poisoned (every pubkey + /// touched by a voided intent, unconditionally). `I3` and `I4` are + /// voided. `I2` is *not*: it was queued for `a2` before `I3` was, so + /// it's isolated from the chain — no `id`-based heuristic needed, the + /// drain-from-position walk simply never reaches it — and it survives + /// to execute normally. + #[test] + fn test_case_3_isolated_intent_survives_the_cascade() { + setup(); + let mut scheduler = IntentScheduler::new(); + let a1 = pubkey!("1111111111111111111111111111111111111111111"); + let a2 = pubkey!("21111111111111111111111111111111111111111111"); + let b1 = pubkey!("31111111111111111111111111111111111111111111"); + let b2 = pubkey!("41111111111111111111111111111111111111111111"); + + let executing_a = create_test_intent(1, &[a1, a2], false); + let i1 = create_test_intent(2, &[b1, b2], false); + assert!(scheduler.schedule(executing_a.clone()).is_some()); + assert!(scheduler.schedule(i1.clone()).is_some()); + + let i2 = create_test_intent(3, &[a1, a2], false); + let i3 = create_test_intent(4, &[a2, b1], false); + let i4 = create_test_intent(5, &[a1, a2], false); + assert!(scheduler.schedule(i2.clone()).is_none()); + assert!(scheduler.schedule(i3.clone()).is_none()); + assert!(scheduler.schedule(i4.clone()).is_none()); + + let poisoned = scheduler.failed(&i1).unwrap(); + let mut ids: Vec<_> = poisoned.iter().map(|p| p.intent_id).collect(); + ids.sort(); + assert_eq!( + ids, + vec![i3.intent_id, i4.intent_id], + "I2 must not be voided" + ); + + // Every pubkey touched by a voided intent rejects new scheduling, + // even a1/a2 where I2 is still healthily queued. + for pk in [a1, a2, b1, b2] { + let probe = create_test_intent(100, &[pk], false); + assert!( + scheduler.schedule(probe).is_none(), + "{pk} must reject new scheduling after the cascade" + ); + } + + // I2 survives and executes normally once executing_a completes. + assert!(scheduler.complete(&executing_a).is_ok()); + let next = scheduler.pop_next_scheduled_intent().unwrap(); + assert_eq!(next.intent_id, i2.intent_id); + } + + /// # Case 4 — a chain of single-key overlaps cascades end to end + /// + /// Unlike Case 1/3, there's no parallel branch here to survive: each + /// intent overlaps the next through exactly one shared pubkey, so the + /// cascade has nowhere to stop until it consumes the whole chain. + /// + /// ```text + /// I1 = [b2] executing, front of b2 (single key) + /// I2 = [b1, b2] front of b1, queued behind I1 on b2 + /// I3 = [a2, b1] front of a2, queued behind I2 on b1 + /// I4 = [a1, a2] front of a1, queued behind I3 on a2 + /// + /// a1 a2 b1 b2 + /// pos0: I4 I3 I2 I1 + /// pos1: . I4 I3 I2 + /// pos2: . . . . + /// pos3: . . . . + /// ``` + /// + /// `I1` fails. `I2` is reachable via `b2`, `I3` via `b1`, `I4` via + /// `a2` — every link in the chain gets voided, and every pubkey the + /// chain ever touched (`a1, a2, b1, b2`) is poisoned. + #[test] + fn test_case_4_full_chain_cascades_through_single_key_overlaps() { + setup(); + let mut scheduler = IntentScheduler::new(); + let a1 = pubkey!("1111111111111111111111111111111111111111111"); + let a2 = pubkey!("21111111111111111111111111111111111111111111"); + let b1 = pubkey!("31111111111111111111111111111111111111111111"); + let b2 = pubkey!("41111111111111111111111111111111111111111111"); + + let i1 = create_test_intent(1, &[b2], false); + let i2 = create_test_intent(2, &[b1, b2], false); + let i3 = create_test_intent(3, &[a2, b1], false); + let i4 = create_test_intent(4, &[a1, a2], false); + assert!(scheduler.schedule(i1.clone()).is_some()); + assert!(scheduler.schedule(i2.clone()).is_none()); + assert!(scheduler.schedule(i3.clone()).is_none()); + assert!(scheduler.schedule(i4.clone()).is_none()); + + let poisoned = scheduler.failed(&i1).unwrap(); + let mut ids: Vec<_> = poisoned.iter().map(|p| p.intent_id).collect(); + ids.sort(); + assert_eq!( + ids, + vec![i2.intent_id, i3.intent_id, i4.intent_id], + "the whole chain must be voided, nothing left to survive" + ); + + for pk in [a1, a2, b1, b2] { + let probe = create_test_intent(100, &[pk], false); + assert!( + scheduler.schedule(probe).is_none(), + "{pk} must reject new scheduling after the cascade" + ); + } + } + + /// # Case 5 — a diamond: two chains fan out and re-converge on one intent + /// + /// `X` is reachable from the failed intent through *two* independent + /// paths. The worklist has to discover it twice and void it once. + /// + /// ```text + /// F = [p1, p2] executing, front of p1 and p2 <- this one fails + /// A = [p1, q1] queued behind F on p1, front of q1 + /// B = [p2, q2] queued behind F on p2, front of q2 + /// X = [q1, q2] queued behind A on q1, behind B on q2 + /// + /// p1 p2 q1 q2 + /// pos0: F F A B + /// pos1: A B X X + /// + /// F + /// / \ + /// A B + /// \ / + /// X + /// ``` + /// + /// `F` fails. `A` and `B` are each poisoned directly; walking `A` + /// reaches `X` via `q1`, and walking `B` reaches `X` via `q2` — two + /// discovery paths landing on the same intent. The worklist is a + /// `BTreeSet`, so `X`'s id is only ever present once: it's + /// popped and voided exactly once, and by the time it's processed both + /// `q1` and `q2` are already drained, so both lookups on `X`'s own + /// pubkeys hit the harmless `Entry::Vacant` bail-out. + #[test] + fn test_case_5_diamond_merge_is_voided_exactly_once() { + setup(); + let mut scheduler = IntentScheduler::new(); + let p1 = pubkey!("1111111111111111111111111111111111111111111"); + let p2 = pubkey!("21111111111111111111111111111111111111111111"); + let q1 = pubkey!("31111111111111111111111111111111111111111111"); + let q2 = pubkey!("41111111111111111111111111111111111111111111"); + + let f = create_test_intent(1, &[p1, p2], false); + let a = create_test_intent(2, &[p1, q1], false); + let b = create_test_intent(3, &[p2, q2], false); + let x = create_test_intent(4, &[q1, q2], false); + assert!(scheduler.schedule(f.clone()).is_some()); + assert!(scheduler.schedule(a.clone()).is_none()); + assert!(scheduler.schedule(b.clone()).is_none()); + assert!(scheduler.schedule(x.clone()).is_none()); + + let poisoned = scheduler.failed(&f).unwrap(); + let mut ids: Vec<_> = poisoned.iter().map(|p| p.intent_id).collect(); + ids.sort(); + assert_eq!( + ids, + vec![a.intent_id, b.intent_id, x.intent_id], + "X must be voided exactly once despite two discovery paths" + ); + + for pk in [p1, p2, q1, q2] { + let probe = create_test_intent(100, &[pk], false); + assert!( + scheduler.schedule(probe).is_none(), + "{pk} must reject new scheduling after the cascade" + ); + } + } + + /// `schedule()`'s admission-time poison check extends `poisoned_keys` + /// with a rejected intent's *entire* pubkey set, same as `failed()` + /// does for a voided intent. This is intentional, not a special case: + /// a rejected intent is an atomic bundle that will never execute, so + /// every pubkey it touches is equally hypothetical, whether or not the + /// scheduler happened to have already materialized it into + /// `blocked_keys` before the rejection was detected. + #[test] + fn test_rejection_poisons_the_rejected_intents_full_pubkey_set() { + setup(); + let mut scheduler = IntentScheduler::new(); + let a1 = pubkey!("1111111111111111111111111111111111111111111"); + let c1 = pubkey!("21111111111111111111111111111111111111111111"); + + // F touches only a1, executes immediately (nothing queued behind + // it), then fails - poisons exactly {a1}. + let f = create_test_intent(1, &[a1], false); + let executed_f = scheduler.schedule(f).unwrap(); + let voided = scheduler.failed(&executed_f).unwrap(); + assert!(voided.is_empty()); + + // Z touches the poisoned a1 *and* c1 in one atomic bundle. Since + // Z can never execute, its effect on c1 never happens either - + // rejecting Z must poison c1 too, not just a1. + let z = create_test_intent(2, &[a1, c1], false); + assert!(scheduler.schedule(z).is_none()); + + // W touches only c1. A caller could construct W assuming Z's + // (never-landed) effect on c1 already happened - e.g. undelegating + // c1 assuming funds Z was supposed to deposit. W must be rejected. + let w = create_test_intent(3, &[c1], false); + assert!( + scheduler.schedule(w).is_none(), + "c1 was touched by Z, which will never execute; schedule() \ + must poison it so a future intent can't assume Z's effects \ + already landed" + ); + } +} + // Helper function to create test intents #[cfg(test)] pub(crate) fn create_test_intent( diff --git a/magicblock-committor-service/src/intent_executor/error.rs b/magicblock-committor-service/src/intent_executor/error.rs index ba0a6368f..70bdf5308 100644 --- a/magicblock-committor-service/src/intent_executor/error.rs +++ b/magicblock-committor-service/src/intent_executor/error.rs @@ -60,6 +60,11 @@ pub enum IntentExecutorError { EmptyIntentError, #[error("Failed to fit in single TX")] FailedToFitError, + /// This doesn't mean permanent failure + /// This means that some ancestor failed + /// to execute with multiple retries + #[error("Intent got poisoned")] + PoisonedIntentError, #[error("SignerError: {0}")] SignerError(#[from] SignerError), #[error("OutboxClientError: {0}")] @@ -120,6 +125,7 @@ impl IntentExecutorError { match self { Self::EmptyIntentError | Self::FailedToFitError + | Self::PoisonedIntentError | Self::SignerError(_) | Self::OutboxClientError(_) // The underlying tx may have already landed; blindly @@ -162,6 +168,7 @@ impl IntentExecutorError { } IntentExecutorError::EmptyIntentError | IntentExecutorError::FailedToFitError + | IntentExecutorError::PoisonedIntentError | IntentExecutorError::SignerError(_) | IntentExecutorError::OutboxClientError(_) | IntentExecutorError::GetPendingSignatureStatusError(_) => None, diff --git a/magicblock-committor-service/src/intent_executor/intent_executor_factory.rs b/magicblock-committor-service/src/intent_executor/intent_executor_factory.rs index bde857525..a363b11aa 100644 --- a/magicblock-committor-service/src/intent_executor/intent_executor_factory.rs +++ b/magicblock-committor-service/src/intent_executor/intent_executor_factory.rs @@ -48,7 +48,7 @@ pub struct ExecutorConfig { pub actions_timeout: Duration, } -/// Dummy struct to simplify signature of CommitSchedulerWorker +/// Dummy struct to simplify signature of IntentExecutionEngine pub struct IntentExecutorBuilderImpl { /// Base-layer signing identity — the engine's authority keypair. pub authority: Keypair, diff --git a/magicblock-committor-service/src/intent_executor/mod.rs b/magicblock-committor-service/src/intent_executor/mod.rs index b18abba3e..a1587d2a0 100644 --- a/magicblock-committor-service/src/intent_executor/mod.rs +++ b/magicblock-committor-service/src/intent_executor/mod.rs @@ -154,6 +154,10 @@ impl IntentExecutionResult { pub fn is_err(&self) -> bool { self.inner.is_err() } + + pub fn is_ok(&self) -> bool { + self.inner.is_ok() + } } #[derive(Default)] diff --git a/magicblock-committor-service/src/intent_executor/utils.rs b/magicblock-committor-service/src/intent_executor/utils.rs index 5dc36f917..462b073d7 100644 --- a/magicblock-committor-service/src/intent_executor/utils.rs +++ b/magicblock-committor-service/src/intent_executor/utils.rs @@ -236,7 +236,8 @@ fn is_terminal_execution_error(err: &IntentExecutorError) -> bool { | IntentExecutorError::GetPendingSignatureStatusError(_) | IntentExecutorError::PendingSignatureResolutionError(_) | IntentExecutorError::FailedCommitPreparationError(_) - | IntentExecutorError::FailedFinalizePreparationError(_) => false, + | IntentExecutorError::FailedFinalizePreparationError(_) + | IntentExecutorError::PoisonedIntentError => false, IntentExecutorError::FailedToCommitError { err, .. } | IntentExecutorError::FailedToFinalizeError { err, .. } => { is_terminal_strategy_error(err) diff --git a/magicblock-metrics/src/metrics/mod.rs b/magicblock-metrics/src/metrics/mod.rs index ea038b3d6..3f3795886 100644 --- a/magicblock-metrics/src/metrics/mod.rs +++ b/magicblock-metrics/src/metrics/mod.rs @@ -348,6 +348,16 @@ lazy_static::lazy_static! { "committor_executors_busy_count", "Number of busy intent executors" ).unwrap(); + static ref COMMITTOR_POISONED_KEYS_COUNT: IntGauge = IntGauge::new( + "committor_poisoned_keys_count", + "Number of pubkeys currently poisoned in the intent scheduler (only cleared by a process restart)" + ).unwrap(); + + static ref COMMITTOR_CASCADE_VOIDED_INTENTS_COUNT: IntCounter = IntCounter::new( + "committor_cascade_voided_intents_count", + "Total number of intents voided by a poisoning cascade after a dependency failed" + ).unwrap(); + static ref COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM: HistogramVec = HistogramVec::new( HistogramOpts::new( "committor_intent_execution_time_histogram_v2", @@ -602,6 +612,8 @@ pub(crate) fn register() { register!(COMMITTOR_INTENTS_BACKLOG_COUNT); register!(COMMITTOR_FAILED_INTENTS_COUNT); register!(COMMITTOR_EXECUTORS_BUSY_COUNT); + register!(COMMITTOR_POISONED_KEYS_COUNT); + register!(COMMITTOR_CASCADE_VOIDED_INTENTS_COUNT); register!(COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM); register!(COMMITTOR_INTENT_CU_USAGE); register!(COMMITTOR_INTENT_TASK_PREPARATION_TIME); @@ -769,6 +781,14 @@ pub fn set_committor_executors_busy_count(value: i64) { COMMITTOR_EXECUTORS_BUSY_COUNT.set(value) } +pub fn set_committor_poisoned_keys_count(value: i64) { + COMMITTOR_POISONED_KEYS_COUNT.set(value) +} + +pub fn inc_committor_cascade_voided_intents_count_by(by: u64) { + COMMITTOR_CASCADE_VOIDED_INTENTS_COUNT.inc_by(by) +} + pub fn observe_committor_intent_execution_time_histogram( seconds: f64, kind: &impl LabelValue,