From 1c98fb4b61a595b70336412843073b27e8122564 Mon Sep 17 00:00:00 2001 From: Grisha Sobol Date: Fri, 12 Jun 2026 00:06:28 +0200 Subject: [PATCH 1/6] feat(vara.eth/ethexe): cap scheduled tasks processed per announce take_actual_tasks_capped drains at most MAX_SCHEDULE_TASKS_PER_MB (64) due tasks oldest-first; the excess stays in the schedule under original heights and runs in following blocks, keeping per-announce commitment size and gas bounded (#5204, #5203). Co-Authored-By: Claude Opus 4.7 --- ethexe/common/src/lib.rs | 7 ++ ethexe/processor/src/lib.rs | 14 ++- ethexe/runtime/common/src/schedule.rs | 8 +- ethexe/runtime/common/src/transitions.rs | 111 +++++++++++++++++++++++ 4 files changed, 134 insertions(+), 6 deletions(-) diff --git a/ethexe/common/src/lib.rs b/ethexe/common/src/lib.rs index 3e06ccf26ff..2f636c30250 100644 --- a/ethexe/common/src/lib.rs +++ b/ethexe/common/src/lib.rs @@ -112,6 +112,13 @@ pub const DEFAULT_COMMITMENT_DELAY_LIMIT: core::num::NonZero = /// Maximum number of touched programs per MB. pub const MAX_TOUCHED_PROGRAMS_PER_MB: u32 = 128; +/// Maximum number of scheduled tasks processed per MB (announce). Excess +/// due tasks stay in the schedule and run in subsequent blocks, keeping +/// the per-announce commitment bounded (#5203). Protocol constant: every +/// validator must apply the same cap. +pub const MAX_SCHEDULE_TASKS_PER_MB: core::num::NonZero = + core::num::NonZero::new(64).expect("64 != 0"); + // Soft limits for one MB processing. Stops execution if any of them is exceeded. pub const OUTGOING_MESSAGES_SOFT_LIMIT: u32 = 128; pub const OUTGOING_MESSAGES_BYTES_SOFT_LIMIT: u32 = 32 * 1024; diff --git a/ethexe/processor/src/lib.rs b/ethexe/processor/src/lib.rs index 337458ff060..6913f01b0a2 100644 --- a/ethexe/processor/src/lib.rs +++ b/ethexe/processor/src/lib.rs @@ -145,7 +145,7 @@ pub use promise::BoundPromiseSink; use core::num::NonZero; use ethexe_common::{ - CodeAndIdUnchecked, ProgramStates, Schedule, + CodeAndIdUnchecked, MAX_SCHEDULE_TASKS_PER_MB, ProgramStates, Schedule, ecdsa::VerifiedData, events::{BlockRequestEvent, MirrorRequestEvent, mirror::MessageQueueingRequestedEvent}, gear::Message, @@ -388,9 +388,19 @@ impl Processor { } fn process_tasks(&mut self, mut transitions: InBlockTransitions) -> InBlockTransitions { - let tasks = transitions.take_actual_tasks(); + let tasks = transitions.take_actual_tasks_capped(MAX_SCHEDULE_TASKS_PER_MB); let block_height = transitions.block_height(); + let deferred = transitions.due_tasks_len(); + if deferred != 0 { + // Excess stays scheduled and runs in following blocks, keeping + // the per-announce commitment bounded (#5204). + log::warn!( + "Schedule for #{block_height} is capped: {} tasks processed, {deferred} deferred", + tasks.len(), + ); + } + log::trace!("Running schedule for #{block_height}: tasks are {tasks:?}"); let mut handler = ScheduleHandler { diff --git a/ethexe/runtime/common/src/schedule.rs b/ethexe/runtime/common/src/schedule.rs index 1b654f5e4e9..ba88d1de4ba 100644 --- a/ethexe/runtime/common/src/schedule.rs +++ b/ethexe/runtime/common/src/schedule.rs @@ -151,10 +151,10 @@ impl TaskHandler for Handler<'_, S> { /// Used primary for fast sync and tests. /// /// No expiry filtering is applied: every scheduled task found in the dumped -/// states is restored. Committed states never hold a task already expired at -/// the dumped block, and the executor drains the full backlog with no lower -/// bound, so any restored task fires at the first computed block regardless of -/// its expiry. +/// states is restored. A committed schedule may hold overdue tasks (capped +/// processing defers the excess — see `MAX_SCHEDULE_TASKS_PER_MB`), and the +/// executor drains the backlog with no lower bound, so any restored task +/// fires starting from the first computed block, subject to the same cap. #[derive(Default)] pub struct Restorer { schedule: Schedule, diff --git a/ethexe/runtime/common/src/transitions.rs b/ethexe/runtime/common/src/transitions.rs index 90505061eed..7ebbc4d0327 100644 --- a/ethexe/runtime/common/src/transitions.rs +++ b/ethexe/runtime/common/src/transitions.rs @@ -90,6 +90,45 @@ impl InBlockTransitions { due.into_values().flatten().collect() } + /// Like [`Self::take_actual_tasks`], but drains at most `limit` tasks. + /// The remainder stays in the schedule under its original heights — + /// it is still due on following blocks and `remove_task` with the + /// original expiry keeps working for deferred entries. + pub fn take_actual_tasks_capped(&mut self, limit: NonZero) -> Vec { + let limit = limit.get(); + let mut taken = Vec::with_capacity(limit.min(16)); + + while taken.len() < limit { + let Some((&height, _)) = self.schedule.first_key_value() else { + break; + }; + if height > self.block_height { + break; + } + + let tasks = self.schedule.get_mut(&height).expect("peeked above"); + while taken.len() < limit + && let Some(task) = tasks.pop_first() + { + taken.push(task); + } + + if tasks.is_empty() { + self.schedule.remove(&height); + } + } + + taken + } + + /// Number of tasks still due at or before the current block height. + pub fn due_tasks_len(&self) -> usize { + self.schedule + .range(..=self.block_height) + .map(|(_, tasks)| tasks.len()) + .sum() + } + pub fn schedule_task(&mut self, in_blocks: NonZero, task: ScheduledTask) -> u32 { let scheduled_block = self.block_height + u32::from(in_blocks); @@ -313,6 +352,78 @@ mod tests { InBlockTransitions::new(block_height, ProgramStates::default(), schedule) } + #[test] + fn capped_drain_respects_limit_and_order() { + let mut schedule = Schedule::new(); + schedule + .entry(5) + .or_default() + .extend([wake(1, 1), wake(2, 2)]); + schedule + .entry(7) + .or_default() + .extend([wake(3, 3), wake(4, 4), wake(5, 5)]); + schedule.entry(20).or_default().insert(wake(9, 9)); + let mut t = transitions_with_schedule(10, schedule); + + let limit = NonZero::new(4).unwrap(); + let taken = t.take_actual_tasks_capped(limit); + // Oldest height first; BTreeSet order within a height. + assert_eq!(taken, vec![wake(1, 1), wake(2, 2), wake(3, 3), wake(4, 4)]); + + // Leftover stays at its ORIGINAL height and is still removable + // with the original expiry. + assert_eq!(t.due_tasks_len(), 1); + t.remove_task(7, &wake(5, 5)) + .expect("deferred task must stay addressable"); + assert_eq!(t.due_tasks_len(), 0); + + // Future tasks are untouched. + assert_eq!(t.schedule.get(&20).map(|s| s.len()), Some(1)); + } + + #[test] + fn capped_drain_iterated_equals_one_shot() { + let mut schedule = Schedule::new(); + for height in [3u32, 4, 9, 10] { + for msg in 0..5u8 { + schedule + .entry(height) + .or_default() + .insert(wake(height as u8, msg)); + } + } + let mut capped = transitions_with_schedule(10, schedule.clone()); + let mut one_shot = transitions_with_schedule(10, schedule); + + let limit = NonZero::new(3).unwrap(); + let mut collected = Vec::new(); + loop { + let chunk = capped.take_actual_tasks_capped(limit); + if chunk.is_empty() { + break; + } + assert!(chunk.len() <= limit.get()); + collected.extend(chunk); + } + + assert_eq!(collected, one_shot.take_actual_tasks()); + } + + #[test] + fn capped_drain_ignores_future_heights() { + let mut schedule = Schedule::new(); + schedule.entry(11).or_default().insert(wake(1, 1)); + let mut t = transitions_with_schedule(10, schedule); + + assert!( + t.take_actual_tasks_capped(NonZero::new(10).unwrap()) + .is_empty() + ); + assert_eq!(t.due_tasks_len(), 0); + assert_eq!(t.schedule.len(), 1); + } + #[test] fn take_actual_tasks_single_height() { let mut schedule = Schedule::new(); From 84f51c2651bacdc28d6700160928ed28f84efbad Mon Sep 17 00:00:00 2001 From: Grisha Sobol Date: Fri, 12 Jun 2026 00:10:24 +0200 Subject: [PATCH 2/6] fix(vara.eth/ethexe): drop outdated scheduled tasks instead of panicking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Schedule task handlers now pre-check the target (program known, entry present in mailbox/waitlist/stash) and skip with a warning when it is gone — an outdated task must not crash the validator or register a spurious state modification. Kind mismatches and storage corruption stay loud (#5204). Co-Authored-By: Claude Opus 4.7 --- ethexe/runtime/common/src/schedule.rs | 123 ++++++++++++++++++++++++++ ethexe/runtime/common/src/state.rs | 24 +++++ 2 files changed, 147 insertions(+) diff --git a/ethexe/runtime/common/src/schedule.rs b/ethexe/runtime/common/src/schedule.rs index ba88d1de4ba..a78cb338830 100644 --- a/ethexe/runtime/common/src/schedule.rs +++ b/ethexe/runtime/common/src/schedule.rs @@ -19,12 +19,41 @@ pub struct Handler<'a, S: Storage> { pub controller: TransitionController<'a, S>, } +impl Handler<'_, S> { + /// Read-only snapshot of the program state; `None` for unknown programs. + fn peek_state(&self, program_id: ActorId) -> Option { + let hash = self.controller.transitions.state_of(&program_id)?.hash; + let state = self + .controller + .storage + .program_state(hash) + .expect("failed to read state from storage"); + Some(state) + } +} + impl TaskHandler for Handler<'_, S> { fn remove_from_mailbox( &mut self, (program_id, user_id): (ActorId, ActorId), message_id: MessageId, ) -> u64 { + // Outdated task (entity already gone, e.g. after restore edge + // cases): drop instead of panicking the announce (#5204). + let target_exists = self.peek_state(program_id).is_some_and(|state| { + let storage = self.controller.storage; + storage + .query(&state.mailbox_hash) + .expect("failed to query mailbox") + .contains(storage, &user_id, &message_id) + }); + if !target_exists { + log::warn!( + "skipping outdated RemoveFromMailbox(({program_id}, {user_id}), {message_id}): target not found" + ); + return 0; + } + self.controller .update_state(program_id, |state, storage, transitions| { let Expiring { @@ -67,6 +96,20 @@ impl TaskHandler for Handler<'_, S> { } fn send_dispatch(&mut self, (program_id, message_id): (ActorId, MessageId)) -> u64 { + let target_exists = self.peek_state(program_id).is_some_and(|state| { + self.controller + .storage + .query(&state.stash_hash) + .expect("failed to query dispatch stash") + .contains(&message_id) + }); + if !target_exists { + log::warn!( + "skipping outdated SendDispatch(({program_id}, {message_id})): target not found" + ); + return 0; + } + self.controller .update_state(program_id, |state, storage, _| { let dispatch = storage.modify(&mut state.stash_hash, |stash| { @@ -83,6 +126,20 @@ impl TaskHandler for Handler<'_, S> { } fn send_user_message(&mut self, stashed_message_id: MessageId, program_id: ActorId) -> u64 { + let target_exists = self.peek_state(program_id).is_some_and(|state| { + self.controller + .storage + .query(&state.stash_hash) + .expect("failed to query dispatch stash") + .contains(&stashed_message_id) + }); + if !target_exists { + log::warn!( + "skipping outdated SendUserMessage({stashed_message_id}, {program_id}): target not found" + ); + return 0; + } + self.controller .update_state(program_id, |state, storage, transitions| { let (dispatch, user_id) = storage.modify(&mut state.stash_hash, |stash| { @@ -118,6 +175,22 @@ impl TaskHandler for Handler<'_, S> { fn wake_message(&mut self, program_id: ActorId, message_id: MessageId) -> u64 { log::trace!("Running scheduled task wake message {message_id} to {program_id}"); + // Missing waitlist entry plausibly means the message was already + // woken or replied to — the task is outdated, drop it (#5204). + let target_exists = self.peek_state(program_id).is_some_and(|state| { + self.controller + .storage + .query(&state.waitlist_hash) + .expect("failed to query waitlist") + .contains(&message_id) + }); + if !target_exists { + log::warn!( + "skipping outdated WakeMessage({program_id}, {message_id}): target not found" + ); + return 0; + } + self.controller .update_state(program_id, |state, storage, _| { let Expiring { @@ -294,6 +367,56 @@ mod tests { use gear_core::buffer::Payload; use std::collections::{BTreeMap, BTreeSet}; + /// Outdated tasks (target entity gone, or even the whole program + /// unknown) are dropped with a warning — and crucially produce NO + /// state modification, so they add nothing to the commitment. + #[test] + fn outdated_tasks_are_skipped() { + use crate::{InBlockTransitions, TransitionController, state::ProgramState}; + use ethexe_common::{ProgramStates, StateHashWithQueueSize}; + + let storage = MemStorage::default(); + let unknown_pid = ActorId::from(0xDEAD); + let empty_pid = ActorId::from(0xBEEF); + let user_id = ActorId::from(0x10); + let message_id = MessageId::from(0x42); + + // `empty_pid` is known but its state holds no mailbox / waitlist / + // stash entries; `unknown_pid` is absent from the states map. + let state_hash = storage.write_program_state(ProgramState::zero()); + let states: ProgramStates = [( + empty_pid, + StateHashWithQueueSize { + hash: state_hash, + canonical_queue_size: 0, + injected_queue_size: 0, + }, + )] + .into_iter() + .collect(); + + let mut transitions = InBlockTransitions::new(10, states, Schedule::default()); + let mut handler = Handler { + controller: TransitionController { + storage: &storage, + transitions: &mut transitions, + }, + }; + + for pid in [unknown_pid, empty_pid] { + assert_eq!(handler.remove_from_mailbox((pid, user_id), message_id), 0); + assert_eq!(handler.send_dispatch((pid, message_id)), 0); + assert_eq!(handler.send_user_message(message_id, pid), 0); + assert_eq!(handler.wake_message(pid, message_id), 0); + } + + assert_eq!( + handler.controller.transitions.modifications_len(), + 0, + "skipped tasks must not register state modifications" + ); + } + #[test] fn restorer_waitlist() { let program_id = ActorId::from(1); diff --git a/ethexe/runtime/common/src/state.rs b/ethexe/runtime/common/src/state.rs index bd04b07f638..4626305be11 100644 --- a/ethexe/runtime/common/src/state.rs +++ b/ethexe/runtime/common/src/state.rs @@ -666,6 +666,10 @@ impl Waitlist { debug_assert!(r.is_none()) } + pub fn contains(&self, message_id: &MessageId) -> bool { + self.inner.contains_key(message_id) + } + pub fn wake(&mut self, message_id: &MessageId) -> Option> { self.inner .remove(message_id) @@ -721,6 +725,10 @@ impl DispatchStash { debug_assert!(r.is_none()); } + pub fn contains(&self, message_id: &MessageId) -> bool { + self.0.contains_key(message_id) + } + pub fn remove_to_program(&mut self, message_id: &MessageId) -> Dispatch { let Expiring { value: (dispatch, user_id), @@ -864,6 +872,22 @@ impl Mailbox { let _ = self.inner.insert(user_id, hash); } + /// Read-only check that `message_id` sits in `user_id`'s mailbox. + pub fn contains( + &self, + storage: &S, + user_id: &ActorId, + message_id: &MessageId, + ) -> bool { + let maybe_hash: MaybeHashOf = self.inner.get(user_id).cloned().into(); + + storage + .query(&maybe_hash) + .expect("failed to query user mailbox") + .0 + .contains_key(message_id) + } + pub fn remove_and_store_user_mailbox( &mut self, storage: &S, From e966ac7a8c2576e8cb99174caae4f8e43565407f Mon Sep 17 00:00:00 2001 From: Grisha Sobol Date: Fri, 12 Jun 2026 00:29:38 +0200 Subject: [PATCH 3/6] test(vara.eth/ethexe): cover capped and outdated task processing Integration: over-cap backlog defers exactly the tail at its original height and drains next block; a dangling task is skipped without aborting sibling program execution. Proptest: iterated capped drain equals the uncapped due-task sequence, each chunk within the cap. Co-Authored-By: Claude Opus 4.7 --- ethexe/processor/src/tests.rs | 159 ++++++++++++++++++++++++++ ethexe/runtime/common/src/proptest.rs | 35 ++++++ 2 files changed, 194 insertions(+) diff --git a/ethexe/processor/src/tests.rs b/ethexe/processor/src/tests.rs index e210e697f24..a2132c8d107 100644 --- a/ethexe/processor/src/tests.rs +++ b/ethexe/processor/src/tests.rs @@ -2423,3 +2423,162 @@ async fn call_wait_up_to_with_huge_duration() { let task = tasks.into_iter().next().unwrap(); assert!(matches!(task, ScheduledTask::WakeMessage(_, _))); } + +/// Capped task processing: only `MAX_SCHEDULE_TASKS_PER_MB` due tasks run +/// per announce; the excess stays scheduled at its original height and +/// drains on the next block. +#[tokio::test] +async fn schedule_tasks_capped_per_announce() { + init_logger(); + + let mut processor = Processor::new(Database::memory()).expect("failed to create processor"); + let chain = BlockChain::mock(3).setup(&processor.db); + + let cap = MAX_SCHEDULE_TASKS_PER_MB.get(); + let extra = 7usize; + let block1 = chain.blocks[1].to_simple(); + + // Dangling wakes for unknown programs: each is skipped by the tolerant + // handler, but still counts against the per-announce cap. + let height = block1.header.height; + let mut schedule = Schedule::default(); + for i in 0..(cap + extra) as u64 { + schedule + .entry(height) + .or_default() + .insert(ethexe_common::ScheduledTask::WakeMessage( + ActorId::from(0xA000 + i), + MessageId::from(i), + )); + } + + let executable = ExecutableData { + height, + timestamp: block1.header.timestamp, + schedule, + ..Default::default() + }; + let FinalizedBlockTransitions { schedule, .. } = processor + .process_programs(executable, None) + .await + .expect("first block must not fail"); + + let leftover: usize = schedule.values().map(|tasks| tasks.len()).sum(); + assert_eq!( + leftover, extra, + "exactly the over-cap tail must be deferred" + ); + assert!( + schedule.keys().all(|&h| h == height), + "deferred tasks keep their original height" + ); + + // Next block drains the remainder. + let block2 = chain.blocks[2].to_simple(); + let executable = ExecutableData { + height: block2.header.height, + timestamp: block2.header.timestamp, + schedule, + ..Default::default() + }; + let FinalizedBlockTransitions { schedule, .. } = processor + .process_programs(executable, None) + .await + .expect("second block must not fail"); + assert!( + schedule.is_empty(), + "backlog must fully drain under the cap" + ); +} + +/// A dangling (outdated) task must be skipped without aborting the rest +/// of the announce: sibling program execution still proceeds. +#[tokio::test] +async fn outdated_task_does_not_abort_announce() { + init_logger(); + + let (mut processor, chain, [code_id]) = + setup_test_env_and_load_codes([demo_ping::WASM_BINARY]).await; + let block1 = chain.blocks[1].to_simple(); + + let user_id = ActorId::from(10); + let actor_id = ActorId::from(0x10000); + + let create_program_events = vec![ + BlockRequestEvent::Router(RouterRequestEvent::ProgramCreated(ProgramCreatedEvent { + actor_id, + code_id, + })), + BlockRequestEvent::Mirror { + actor_id, + event: MirrorRequestEvent::ExecutableBalanceTopUpRequested( + ExecutableBalanceTopUpRequestedEvent { + value: 1_500_000_000_000, + }, + ), + }, + // First queued message becomes the Init dispatch — spend it. + BlockRequestEvent::Mirror { + actor_id, + event: MirrorRequestEvent::MessageQueueingRequested(MessageQueueingRequestedEvent { + id: MessageId::from(100), + source: user_id, + payload: vec![], + value: 0, + call_reply: false, + }), + }, + BlockRequestEvent::Mirror { + actor_id, + event: MirrorRequestEvent::MessageQueueingRequested(MessageQueueingRequestedEvent { + id: MessageId::from(1), + source: user_id, + payload: b"PING".to_vec(), + value: 0, + call_reply: false, + }), + }, + ]; + + // Dangling siblings at the same and a past height. + let height = block1.header.height; + let mut schedule = Schedule::default(); + schedule + .entry(height) + .or_default() + .insert(ethexe_common::ScheduledTask::WakeMessage( + ActorId::from(0xDEAD), + MessageId::from(0xD1), + )); + schedule + .entry(height.saturating_sub(1)) + .or_default() + .insert(ethexe_common::ScheduledTask::SendUserMessage { + message_id: MessageId::from(0xD2), + to_mailbox: ActorId::from(0xDEAD), + }); + + let executable = ExecutableData { + height, + timestamp: block1.header.timestamp, + schedule, + events: create_program_events, + gas_allowance: Some(DEFAULT_BLOCK_GAS_LIMIT), + ..Default::default() + }; + let finalized = processor + .process_programs(executable, None) + .await + .expect("dangling tasks must not abort the announce"); + + assert!(finalized.schedule.is_empty(), "dangling tasks consumed"); + let payloads: Vec<_> = finalized + .transitions + .iter() + .flat_map(|t| t.messages.iter().map(|m| m.payload.clone())) + .collect(); + assert!( + payloads.contains(&b"PONG".to_vec()), + "sibling PING execution must still produce PONG, got {payloads:?}" + ); +} diff --git a/ethexe/runtime/common/src/proptest.rs b/ethexe/runtime/common/src/proptest.rs index 5ea9ea08908..023906f01ec 100644 --- a/ethexe/runtime/common/src/proptest.rs +++ b/ethexe/runtime/common/src/proptest.rs @@ -1065,6 +1065,41 @@ mod tests { ::proptest::proptest! { #![proptest_config(proptest_config())] + /// Capping only slices the drain into chunks: iterating + /// `take_actual_tasks_capped` yields exactly the uncapped due-task + /// sequence, each chunk within the cap, nothing lost or reordered. + #[test] + fn capped_drain_equals_uncapped_sequence( + schedule in common_schedule_strategy(), + block_height in any::(), + cap in 1usize..6, + ) { + let mut capped = InBlockTransitions::new( + block_height, + ProgramStates::default(), + schedule.clone(), + ); + let mut uncapped = InBlockTransitions::new( + block_height, + ProgramStates::default(), + schedule, + ); + + let limit = core::num::NonZero::new(cap).expect("cap > 0"); + let mut collected = Vec::new(); + loop { + let chunk = capped.take_actual_tasks_capped(limit); + if chunk.is_empty() { + break; + } + prop_assert!(chunk.len() <= cap); + collected.extend(chunk); + } + + prop_assert_eq!(collected, uncapped.take_actual_tasks()); + prop_assert_eq!(capped.due_tasks_len(), 0); + } + #[test] fn finalize_matches_model((transitions, model) in in_block_transitions_with_model_strategy()) { let finalized = transitions.finalize(); From 873ebe489ba38522bfdbb7c9843141b8d7a31a26 Mon Sep 17 00:00:00 2001 From: Grisha Sobol Date: Fri, 12 Jun 2026 11:08:15 +0200 Subject: [PATCH 4/6] fix(vara.eth/ethexe): tie the task cap to the modifications budget Task processing shares PROGRAM_MODIFICATIONS_SOFT_LIMIT with queue processing; a 64-task block exhausted the whole budget and skipped queues entirely. Cap at half the budget (32) so queues always retain headroom (audit finding). Co-Authored-By: Claude Opus 4.7 --- ethexe/common/src/lib.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/ethexe/common/src/lib.rs b/ethexe/common/src/lib.rs index 2f636c30250..7c8293fd0a4 100644 --- a/ethexe/common/src/lib.rs +++ b/ethexe/common/src/lib.rs @@ -116,8 +116,14 @@ pub const MAX_TOUCHED_PROGRAMS_PER_MB: u32 = 128; /// due tasks stay in the schedule and run in subsequent blocks, keeping /// the per-announce commitment bounded (#5203). Protocol constant: every /// validator must apply the same cap. +/// +/// Tasks run before queue processing and their state modifications count +/// toward the same [`PROGRAM_MODIFICATIONS_SOFT_LIMIT`] budget, so the cap +/// is half of it — a saturated task backlog slows queue processing but +/// cannot starve it entirely. pub const MAX_SCHEDULE_TASKS_PER_MB: core::num::NonZero = - core::num::NonZero::new(64).expect("64 != 0"); + core::num::NonZero::new(PROGRAM_MODIFICATIONS_SOFT_LIMIT as usize / 2) + .expect("soft limit must be non-zero"); // Soft limits for one MB processing. Stops execution if any of them is exceeded. pub const OUTGOING_MESSAGES_SOFT_LIMIT: u32 = 128; From 70544946bbd4191a1953a66820a8cb437407dffd Mon Sep 17 00:00:00 2001 From: Grisha Sobol Date: Fri, 12 Jun 2026 11:23:17 +0200 Subject: [PATCH 5/6] docs(vara.eth/ethexe): note deferred-expiry semantics under task backlog Co-Authored-By: Claude Opus 4.7 --- ethexe/common/src/lib.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/ethexe/common/src/lib.rs b/ethexe/common/src/lib.rs index 7c8293fd0a4..56439d9fde8 100644 --- a/ethexe/common/src/lib.rs +++ b/ethexe/common/src/lib.rs @@ -121,6 +121,10 @@ pub const MAX_TOUCHED_PROGRAMS_PER_MB: u32 = 128; /// toward the same [`PROGRAM_MODIFICATIONS_SOFT_LIMIT`] budget, so the cap /// is half of it — a saturated task backlog slows queue processing but /// cannot starve it entirely. +/// +/// Known consequence: while a backlog drains, deferred expiries (e.g. +/// `RemoveFromMailbox`) fire late, so externally visible deadlines such +/// as the mailbox TTL are lower bounds, not exact heights. pub const MAX_SCHEDULE_TASKS_PER_MB: core::num::NonZero = core::num::NonZero::new(PROGRAM_MODIFICATIONS_SOFT_LIMIT as usize / 2) .expect("soft limit must be non-zero"); From 9439d3474bbd5174f435cff1d41cf43587665c7e Mon Sep 17 00:00:00 2001 From: Grisha Sobol Date: Fri, 12 Jun 2026 11:53:05 +0200 Subject: [PATCH 6/6] test(vara.eth/ethexe): drain capped task backlog in wait/wake tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit many_waits and cross_height_wake_drain schedule more wakes than MAX_SCHEDULE_TASKS_PER_MB allows per announce — drive process_tasks the way consecutive blocks would until the backlog is empty. Co-Authored-By: Claude Opus 4.7 --- ethexe/processor/src/tests.rs | 27 ++++++++++++++++++++------- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/ethexe/processor/src/tests.rs b/ethexe/processor/src/tests.rs index a2132c8d107..b098c56c012 100644 --- a/ethexe/processor/src/tests.rs +++ b/ethexe/processor/src/tests.rs @@ -998,12 +998,19 @@ async fn many_waits() { // Check all messages wake up and reply with "Hello, world!" in wake block. // Hack: change block height to wake up tasks. - let transitions = handler + let mut transitions = handler .transitions .tap_mut(|ts| *ts.block_height_mut() = wake_block.header.height); - let mut transitions = processor.process_tasks(transitions); - // Hack: nullify modifications to avoid modifications limit. - transitions.modifications_mut().clear(); + // Task processing is capped per announce (MAX_SCHEDULE_TASKS_PER_MB); + // drain the whole backlog the way consecutive blocks would. + loop { + transitions = processor.process_tasks(transitions); + // Hack: nullify modifications to avoid modifications limit. + transitions.modifications_mut().clear(); + if transitions.due_tasks_len() == 0 { + break; + } + } let transitions = processor .process_queues( transitions, @@ -1151,11 +1158,17 @@ async fn cross_height_wake_drain() { // Jump past the scheduled wake height (block1 + blocks_to_wait + 5): // `process_tasks` must still drain the wakes despite the height gap. - let transitions = handler + let mut transitions = handler .transitions .tap_mut(|ts| *ts.block_height_mut() = wake_block.header.height); - let mut transitions = processor.process_tasks(transitions); - transitions.modifications_mut().clear(); + // Drain the capped backlog the way consecutive blocks would. + loop { + transitions = processor.process_tasks(transitions); + transitions.modifications_mut().clear(); + if transitions.due_tasks_len() == 0 { + break; + } + } let transitions = processor .process_queues( transitions,