diff --git a/guide/lua.md b/guide/lua.md index 4f59b43..94b942b 100644 --- a/guide/lua.md +++ b/guide/lua.md @@ -27,7 +27,15 @@ Each channel runs the script independently. It executes top-level initialization ## Events and payloads -A Source event has `type = "source"`, a timestamp and `payload`. A timer event has `type = "timer"`, a timestamp and no payload. Events and nested payload values are deeply read-only; modifying them is a sandbox violation. +| Field | Source event | Timer event | +| --- | --- | --- | +| `type` | `"source"` | `"timer"` | +| `timestamp` | Dispatch time, in integer milliseconds since Pipeline startup | Dispatch time, on the same timeline | +| `payload` | Read-only Source payload | Absent (`nil`) | +| `id` | Absent (`nil`) | Timer name (string), or `nil` for the anonymous timer | +| `eligibleAt` | Absent (`nil`) | Scheduled deadline, in integer milliseconds since Pipeline startup | + +Events and nested payload values are deeply read-only; modifying them is a sandbox violation. `event.timestamp` is a monotonic integer millisecond count since Pipeline startup, sampled when the channel dispatches the event. It is not Unix time. Use the wall-clock functions below for a real-world date or timestamp. @@ -87,19 +95,28 @@ Registry lookup, Builder receiver/argument/type/range errors and timer argument ## Timers and state -Each VM has one one-shot timer slot: +Each VM has an anonymous one-shot timer slot plus independently named one-shot timers: ```lua setTimeout(durationMs) -clearTimerTask() +setTimeout(durationMs, id) +clearTimeout() +clearTimeout(id) hasTimeout() +hasTimeout(id) ``` -The delay must be a nonnegative Lua integer. It starts at the call, on a monotonic timeline. Setting again replaces the pending timer; clearing an empty slot succeeds. `hasTimeout()` observes the current pending slot immediately. The slot is removed before its timer event enters `main`, so it reports false during that event unless another timer was scheduled. +The delay must be a nonnegative Lua integer. It starts at the call, on a monotonic timeline. Setting again replaces the pending timer with the same id; different ids remain independent. `clearTimeout` is idempotent, and `hasTimeout` observes the selected timer immediately. A timer is removed before its event enters `main`, so it reports false during that event unless another timer was scheduled. The id must be a non-empty UTF-8 string of at most 128 bytes. The maximum accepted integer is `9223372036854775807`; if conversion or deadline arithmetic cannot represent it, the catchable error is `setTimeout delay is out of range`. + +These functions are allowed at top level and in `main`. A zero delay schedules eligibility on the next event-loop iteration. Execution may be later because the channel is busy. Periodic behavior explicitly schedules the next one-shot timer from the current timer event. Negative, non-integer, string, or missing delays raise `setTimeout delay must be a non-negative integer`; a delay that cannot be represented by the monotonic deadline or `eligibleAt` calculation raises `setTimeout delay is out of range`. + +Omitting `id` or passing `nil` selects the anonymous timer. Invalid ids raise the catchable error `timer id must be a non-empty string of at most 128 bytes`. + +`event.eligibleAt` is the timer's deadline in integer milliseconds since Pipeline startup, on the same monotonic timeline as `event.timestamp`. The difference `event.timestamp - event.eligibleAt` is its dispatch delay. Timers with the same deadline run in registration order. Among due timers and readable Source input, the earlier deadline or first observed Source readiness runs first; ties use their registration or observation order. Source readiness keeps its place until one record is consumed, so a timer that repeatedly reschedules itself with zero delay cannot indefinitely prevent readable Source input from running. Sink backpressure and lifecycle work can still delay either kind of event. -These functions are allowed at top level and in `main`. A zero delay schedules eligibility on the next event-loop iteration. Execution may be later because the channel is busy. Periodic behavior explicitly schedules the next one-shot timer from the current timer event. Invalid delays raise `setTimeout delay must be a non-negative integer`. +Named timers share the VM's batch completion behavior: the first successful `emit` in an event can complete all pending Source records in that VM. A timer id does not provide independent reliable completion for records with that key. -Lua state, Builders, snapshots and timers count toward the VM's memory allowance. Top-level initialization and each `main` invocation are bounded by the Runner's Lua CPU limit. State lives only in the current VM. A script/resource/sandbox failure invalidates the VM and its timers; rebuilding executes top-level code again. Relevant Document updates and process restarts also recreate state. Script replacement can briefly hold both old and new VMs, each with its own configured allowance. +Lua state, Builders, snapshots and timers count toward the VM's memory allowance. Each pending timer is charged 2 KiB plus three times its id length in UTF-8 bytes (zero id bytes for the anonymous timer). The fixed charge includes an allowance for the timer and its indexes; it is an estimate, not an exact allocation or process-memory measurement. There is no separate timer-count limit. Top-level initialization and each `main` invocation are bounded by the Runner's Lua CPU limit. State lives only in the current VM. A script/resource/sandbox failure invalidates the VM and its timers; rebuilding executes top-level code again. Relevant Document updates and process restarts also recreate state. Script replacement can briefly hold both old and new VMs, each with its own configured allowance. ## Wall-clock time and dates diff --git a/src/lua/event.rs b/src/lua/event.rs index ad2dc24..60360d8 100644 --- a/src/lua/event.rs +++ b/src/lua/event.rs @@ -43,6 +43,8 @@ pub(super) enum ProcessEvent { }, Timer { timestamp: KernelTimestampMillis, + id: Option>, + eligible_at: KernelTimestampMillis, }, } @@ -167,9 +169,15 @@ pub(super) fn project( project_source_message(lua, payload, Rc::clone(&fatal_fault))?, )?; } - ProcessEvent::Timer { timestamp } => { + ProcessEvent::Timer { + timestamp, + id, + eligible_at, + } => { backing.raw_set("type", "timer")?; backing.raw_set("timestamp", timestamp.0)?; + backing.raw_set("id", id.as_deref())?; + backing.raw_set("eligibleAt", eligible_at.0)?; } } install_readonly_backing(lua, backing, fatal_fault) diff --git a/src/lua/memory.rs b/src/lua/memory.rs new file mode 100644 index 0000000..9887dfb --- /dev/null +++ b/src/lua/memory.rs @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +//! Shared native-memory budget for one thread-affine Lua VM. +//! +//! Sandbox initialization creates one budget and shares it with Payload and +//! timer owners. Each owner supplies its memory charge and releases it when +//! its storage is dropped. The budget reduces the Lua allocator's allowance +//! so native charges and Lua allocations share the same VM limit. +//! +//! The Lua reference is weak: reservations may outlive the interpreter during +//! teardown without keeping it alive. Failed reservations leave both the +//! charge and allocator allowance unchanged. This module owns no payload or +//! timer storage and does not define their accounting estimates. + +use super::{LuaApiFailure, LuaApiResult, LuaVmFatalFault, record_fatal_fault}; +use mlua::{Lua, WeakLua}; +use std::cell::Cell; +use std::fmt; +use std::rc::Rc; + +pub(super) struct LuaNativeMemoryBudget { + lua: WeakLua, + limit_bytes: usize, + used_bytes: Cell, + fatal_fault: Rc>>, +} + +impl LuaNativeMemoryBudget { + pub(super) fn new( + lua: &Lua, + limit_bytes: usize, + fatal_fault: Rc>>, + ) -> Self { + Self { + lua: lua.weak(), + limit_bytes, + used_bytes: Cell::new(0), + fatal_fault, + } + } + + pub(super) fn used_bytes(&self) -> usize { + self.used_bytes.get() + } + + pub(super) fn replace(&self, previous_bytes: usize, next_bytes: usize) -> LuaApiResult<()> { + let used_bytes = self.used_bytes.get(); + let retained_bytes = used_bytes + .checked_sub(previous_bytes) + .ok_or(LuaApiFailure::InternalInvariantViolation)?; + let next_used_bytes = retained_bytes + .checked_add(next_bytes) + .ok_or(LuaApiFailure::MemoryExceeded)?; + let lua = self + .lua + .try_upgrade() + .ok_or(LuaApiFailure::InternalInvariantViolation)?; + let next_lua_limit = self + .limit_bytes + .checked_sub(next_used_bytes) + .ok_or(LuaApiFailure::MemoryExceeded)?; + if lua.used_memory() > next_lua_limit { + return Err(LuaApiFailure::MemoryExceeded); + } + lua.set_memory_limit(next_lua_limit) + .map_err(|_| LuaApiFailure::InternalInvariantViolation)?; + self.used_bytes.set(next_used_bytes); + Ok(()) + } + + pub(super) fn release(&self, bytes: usize) { + let Some(next_used_bytes) = self.used_bytes.get().checked_sub(bytes) else { + record_fatal_fault( + &self.fatal_fault, + LuaVmFatalFault::InternalInvariantViolation, + ); + return; + }; + self.used_bytes.set(next_used_bytes); + let Some(lua) = self.lua.try_upgrade() else { + return; + }; + let Some(next_lua_limit) = self.limit_bytes.checked_sub(next_used_bytes) else { + record_fatal_fault( + &self.fatal_fault, + LuaVmFatalFault::InternalInvariantViolation, + ); + return; + }; + if lua.set_memory_limit(next_lua_limit).is_err() { + record_fatal_fault( + &self.fatal_fault, + LuaVmFatalFault::InternalInvariantViolation, + ); + } + } +} +impl fmt::Debug for LuaNativeMemoryBudget { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LuaNativeMemoryBudget") + .field("limit_bytes", &self.limit_bytes) + .field("used_bytes", &self.used_bytes.get()) + .finish_non_exhaustive() + } +} diff --git a/src/lua/mod.rs b/src/lua/mod.rs index 4d6938b..0b6c2dc 100644 --- a/src/lua/mod.rs +++ b/src/lua/mod.rs @@ -35,7 +35,7 @@ //! module owns all `mlua` installation, fixed-size argument rendering, and //! callback invocation without knowing the destination state. -use self::metrics::VmMetrics; +use self::{memory::LuaNativeMemoryBudget, metrics::VmMetrics}; use crate::config::ScriptVmLimits; use crate::contracts::core::DIAGNOSTIC_TEXT_MAXIMUM_BYTES; use crate::identifiers::SinkContractId; @@ -60,12 +60,13 @@ mod bytes; mod emit; mod event; mod json; +mod memory; pub(crate) mod metrics; mod payload; mod timer; pub(crate) use emit::EmitBoundary; -pub(crate) use timer::{TimerSchedule, TimerSlotInactive}; +pub(crate) use timer::{TimerEvent, TimerSchedule, TimerSlotInactive}; const HOOK_INSTRUCTION_INTERVAL: u32 = 1_000; const INTERNAL_CPU_LIMIT_ERROR: &str = "Tenon Lua CPU time limit exceeded"; @@ -459,7 +460,7 @@ pub(crate) struct LuaVm { emit_slot: emit::EmitSlot, timer_slot: timer::TimerSlot, cpu_time_limit: Duration, - payload_memory: Rc, + native_memory_budget: Rc, metrics: Option, } @@ -510,6 +511,37 @@ impl LuaVm { print: LuaPrintCallback, stop_requested: impl Fn() -> bool + 'static, metrics: Option, + ) -> Result { + Self::load_observed_at( + source, + limits, + max_record_bytes, + source_payload_root, + sink_payload_roots, + print, + stop_requested, + metrics, + Instant::now(), + Rc::new(Cell::new(0)), + ) + } + + /// Uses the Channel timeline and ordering sequence across VM replacements. + #[expect( + clippy::too_many_arguments, + reason = "VM construction receives runtime, contract, observation, and Channel timeline inputs" + )] + pub(crate) fn load_observed_at( + source: &str, + limits: ScriptVmLimits, + max_record_bytes: NonZeroU64, + source_payload_root: MessageDescriptor, + sink_payload_roots: HashMap, + print: LuaPrintCallback, + stop_requested: impl Fn() -> bool + 'static, + metrics: Option, + timer_origin: Instant, + event_order: Rc>, ) -> Result { let lua = Lua::new_with( StdLib::STRING | StdLib::TABLE | StdLib::MATH | StdLib::UTF8 | StdLib::OS, @@ -526,6 +558,8 @@ impl LuaVm { limits.memory_bytes().get(), max_record_bytes, print, + timer_origin, + event_order, ) .map_err(|source| LuaVmError::with_source(LuaVmErrorKind::InitializationFailed, source))?; @@ -557,7 +591,7 @@ impl LuaVm { emit_slot: sandbox.emit_slot, timer_slot: sandbox.timer_slot, cpu_time_limit: limits.cpu_time(), - payload_memory: sandbox.payload_memory, + native_memory_budget: sandbox.native_memory_budget, metrics, }; vm.publish_memory(); @@ -583,31 +617,37 @@ impl LuaVm { Ok(self.call_main_ready(&event)) } - pub(crate) fn call_timer(&mut self, timestamp_millis: i64) -> LuaMainOutcome { + pub(crate) fn call_timer_event( + &mut self, + timestamp_millis: i64, + timer: TimerEvent, + ) -> LuaMainOutcome { if let Some(outcome) = self.terminal_outcome() { return outcome; } self.call_main_ready(&event::ProcessEvent::Timer { timestamp: event::KernelTimestampMillis(timestamp_millis), + id: timer.id, + eligible_at: event::KernelTimestampMillis(timer.eligible_at), }) } - pub(crate) fn scheduled_timer(&self) -> Option { + pub(crate) fn next_timer_event(&self) -> Option { if self.terminal_error_kind().is_some() { None } else { - self.timer_slot.schedule() + self.timer_slot.next_event() } } - pub(crate) fn begin_timer(&self) -> Result<(), TimerSlotInactive> { + pub(crate) fn begin_timer(&self) -> Result { self.timer_slot.begin() } fn publish_memory(&self) { if let Some(metrics) = &self.metrics { - metrics.publish_memory(self.lua.used_memory() + self.payload_memory.used_bytes()); + metrics.publish_memory(self.lua.used_memory() + self.native_memory_budget.used_bytes()); } } @@ -768,10 +808,14 @@ struct Sandbox { protected_names: Rc>>>, emit_slot: emit::EmitSlot, timer_slot: timer::TimerSlot, - payload_memory: Rc, + native_memory_budget: Rc, } impl Sandbox { + #[expect( + clippy::too_many_arguments, + reason = "Sandbox installation binds VM resource owners and the Channel timeline" + )] fn install( lua: &Lua, fatal_fault: Rc>>, @@ -780,6 +824,8 @@ impl Sandbox { memory_limit_bytes: usize, max_record_bytes: NonZeroU64, print: LuaPrintCallback, + timer_origin: Instant, + event_order: Rc>, ) -> mlua::Result { let native_globals = lua.globals(); let guards = SandboxGuards { @@ -789,7 +835,7 @@ impl Sandbox { let environment_values = lua.create_table()?; let environment = lua.create_table()?; let emit_slot = emit::EmitSlot::default(); - let timer_slot = timer::TimerSlot::default(); + let timer_slot = timer::TimerSlot::new(timer_origin, event_order); install_environment_metatable( lua, @@ -869,22 +915,28 @@ impl Sandbox { Rc::clone(&guards.fatal_fault), Rc::clone(&execution_budget), )?; - timer::install( + let native_memory_budget = Rc::new(LuaNativeMemoryBudget::new( + lua, + memory_limit_bytes, + Rc::clone(&guards.fatal_fault), + )); + payload::install( lua, &environment_values, Rc::clone(&guards.protected_names), Rc::clone(&guards.fatal_fault), Rc::clone(&execution_budget), - timer_slot.clone(), + payload_registry, + Rc::clone(&native_memory_budget), )?; - let payload_memory = payload::install( + timer::install( lua, &environment_values, Rc::clone(&guards.protected_names), Rc::clone(&guards.fatal_fault), Rc::clone(&execution_budget), - payload_registry, - memory_limit_bytes, + timer_slot.clone(), + Rc::clone(&native_memory_budget), )?; environment_values.raw_set("_G", environment.clone())?; @@ -905,7 +957,7 @@ impl Sandbox { protected_names: guards.protected_names, emit_slot, timer_slot, - payload_memory, + native_memory_budget, }) } } diff --git a/src/lua/payload.rs b/src/lua/payload.rs index b77c4a8..2dc216b 100644 --- a/src/lua/payload.rs +++ b/src/lua/payload.rs @@ -35,19 +35,19 @@ use super::{ ExecutionBudget, LuaApiFailure, LuaApiResult, LuaVmFatalFault, - create_catchable_api_wrapper_factory, finish_api_call, protect_name, record_fatal_fault, + create_catchable_api_wrapper_factory, finish_api_call, memory::LuaNativeMemoryBudget, + protect_name, }; use crate::identifiers::SinkContractId; use mlua::{ AnyUserData, Function, Lua, LuaString, MetaMethod, MultiValue, Table, UserData, UserDataFields, - UserDataMethods, Value as LuaValue, WeakLua, + UserDataMethods, Value as LuaValue, }; use prost_reflect::{ DynamicMessage, FieldDescriptor, Kind, MapKey, MessageDescriptor, Value as ProtobufValue, }; use std::cell::{Cell, RefCell}; use std::collections::{HashMap, HashSet}; -use std::fmt; use std::rc::Rc; use std::str; @@ -107,7 +107,7 @@ struct PayloadBinding { struct LuaPayloadRuntime { fatal_fault: Rc>>, execution_budget: Rc>>, - memory_budget: Rc, + memory_budget: Rc, wrapper_factory: Function, } @@ -315,16 +315,12 @@ pub(super) fn install( fatal_fault: Rc>>, execution_budget: Rc>>, registry: FrozenPayloadRegistry, - memory_limit_bytes: usize, -) -> mlua::Result> { + memory_budget: Rc, +) -> mlua::Result<()> { let runtime = Rc::new(LuaPayloadRuntime { - fatal_fault: Rc::clone(&fatal_fault), + fatal_fault, execution_budget, - memory_budget: Rc::new(PayloadMemoryBudget::new( - lua, - memory_limit_bytes, - fatal_fault, - )), + memory_budget, wrapper_factory: create_catchable_api_wrapper_factory(lua)?, }); let prototypes = BuilderPrototypeCatalog::compile(lua, ®istry, Rc::clone(&runtime))?; @@ -349,7 +345,7 @@ pub(super) fn install( let registry = lua.create_userdata(RegistryUserData { get_builder })?; environment_values.raw_set("registry", registry)?; protect_name(&protected_names, "registry"); - Ok(Rc::clone(&context.runtime.memory_budget)) + Ok(()) } fn get_builder( @@ -922,98 +918,14 @@ fn require_no_arguments(arguments: &MultiValue, message: &'static str) -> LuaApi } } -pub(super) struct PayloadMemoryBudget { - lua: WeakLua, - limit_bytes: usize, - used_bytes: Cell, - fatal_fault: Rc>>, -} - -impl fmt::Debug for PayloadMemoryBudget { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter - .debug_struct("PayloadMemoryBudget") - .field("limit_bytes", &self.limit_bytes) - .field("used_bytes", &self.used_bytes.get()) - .finish_non_exhaustive() - } -} - -impl PayloadMemoryBudget { - pub(super) fn used_bytes(&self) -> usize { - self.used_bytes.get() - } - - fn new(lua: &Lua, limit_bytes: usize, fatal_fault: Rc>>) -> Self { - Self { - lua: lua.weak(), - limit_bytes, - used_bytes: Cell::new(0), - fatal_fault, - } - } - - fn replace(&self, previous_bytes: usize, next_bytes: usize) -> LuaApiResult<()> { - let used_bytes = self.used_bytes.get(); - let retained_bytes = used_bytes - .checked_sub(previous_bytes) - .ok_or(LuaApiFailure::InternalInvariantViolation)?; - let next_used_bytes = retained_bytes - .checked_add(next_bytes) - .ok_or(LuaApiFailure::MemoryExceeded)?; - let lua = self - .lua - .try_upgrade() - .ok_or(LuaApiFailure::InternalInvariantViolation)?; - let next_lua_limit = self - .limit_bytes - .checked_sub(next_used_bytes) - .ok_or(LuaApiFailure::MemoryExceeded)?; - if lua.used_memory() > next_lua_limit { - return Err(LuaApiFailure::MemoryExceeded); - } - lua.set_memory_limit(next_lua_limit) - .map_err(|_| LuaApiFailure::InternalInvariantViolation)?; - self.used_bytes.set(next_used_bytes); - Ok(()) - } - - fn release(&self, bytes: usize) { - let Some(next_used_bytes) = self.used_bytes.get().checked_sub(bytes) else { - record_fatal_fault( - &self.fatal_fault, - LuaVmFatalFault::InternalInvariantViolation, - ); - return; - }; - self.used_bytes.set(next_used_bytes); - let Some(lua) = self.lua.try_upgrade() else { - return; - }; - let Some(next_lua_limit) = self.limit_bytes.checked_sub(next_used_bytes) else { - record_fatal_fault( - &self.fatal_fault, - LuaVmFatalFault::InternalInvariantViolation, - ); - return; - }; - if lua.set_memory_limit(next_lua_limit).is_err() { - record_fatal_fault( - &self.fatal_fault, - LuaVmFatalFault::InternalInvariantViolation, - ); - } - } -} - #[derive(Debug)] struct PayloadMemoryReservation { - budget: Rc, + budget: Rc, bytes: usize, } impl PayloadMemoryReservation { - fn reserve(budget: Rc, bytes: usize) -> LuaApiResult { + fn reserve(budget: Rc, bytes: usize) -> LuaApiResult { budget.replace(0, bytes)?; Ok(Self { budget, bytes }) } diff --git a/src/lua/tests.rs b/src/lua/tests.rs index 1e8dedf..5c158b9 100644 --- a/src/lua/tests.rs +++ b/src/lua/tests.rs @@ -52,6 +52,31 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, OnceLock}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +// General VM tests inject an anonymous timer event without scheduling a task. +// Timer lifecycle tests pass the event returned by begin_timer instead. +impl LuaVm { + fn call_timer(&mut self, timestamp_millis: i64) -> LuaMainOutcome { + let now = Instant::now(); + self.call_timer_event( + timestamp_millis, + super::TimerEvent { + schedule: super::TimerSchedule { + scheduled_at: now, + delay: Duration::ZERO, + }, + deadline: now, + eligible_at: timestamp_millis, + id: None, + sequence: 0, + }, + ) + } + + fn scheduled_timer(&self) -> Option { + self.next_timer_event().map(|timer| timer.schedule) + } +} + const TEST_MEMORY_LIMIT_BYTES: usize = 4 * 1024 * 1024; const TEST_CPU_TIME_LIMIT: Duration = Duration::from_millis(100); @@ -101,12 +126,12 @@ impl LuaMainOutcomeTestExt for LuaMainOutcome { } } -fn limits() -> io::Result { +pub(super) fn limits() -> io::Result { ScriptVmLimits::try_new(non_zero(TEST_MEMORY_LIMIT_BYTES)?, TEST_CPU_TIME_LIMIT) .map_err(io::Error::other) } -fn load_vm(source: &str, limits: ScriptVmLimits) -> Result { +pub(super) fn load_vm(source: &str, limits: ScriptVmLimits) -> Result { load_vm_with_contracts(source, limits, HashMap::new()) } @@ -2049,7 +2074,8 @@ fn exposes_only_the_fixed_library_allowlist() -> io::Result<()> { assert(type(utf8.codepoint) == "function") assert(type(print) == "function") assert(type(setTimeout) == "function") - assert(type(clearTimerTask) == "function") + assert(type(clearTimeout) == "function") + assert(clearTimerTask == nil) assert(type(hasTimeout) == "function") assert(type(currentTimeMillis) == "function") assert(type(os.date) == "function") @@ -2567,7 +2593,7 @@ fn top_level_zero_timeout_waits_for_the_next_event_loop_turn() -> io::Result<()> assert(not hasTimeout()) setTimeout(25) assert(hasTimeout()) - clearTimerTask() + clearTimeout() assert(not hasTimeout()) setTimeout(0) assert(hasTimeout()) @@ -2597,11 +2623,13 @@ fn top_level_zero_timeout_waits_for_the_next_event_loop_turn() -> io::Result<()> assert!(lua_has_timeout(&vm).map_err(test_error)?); assert_eq!(vm.scheduled_timer(), Some(schedule)); - assert!(vm.begin_timer().is_ok()); + let event = vm + .begin_timer() + .map_err(|_| io::Error::other("timer missing"))?; assert!(!lua_has_timeout(&vm).map_err(test_error)?); assert!(vm.scheduled_timer().is_none()); assert!(vm.begin_timer().is_err()); - vm.call_timer(1) + vm.call_timer_event(1, event) .into_result_without_emit_boundaries() .map_err(test_error)?; assert_eq!( @@ -2647,7 +2675,7 @@ fn every_new_vm_schedules_its_own_top_level_timeout() -> io::Result<()> { fn timer_apis_share_one_slot_in_top_level_and_source_main() -> io::Result<()> { let mut vm = load_vm( r#" - clearTimerTask() + clearTimeout() function main(event) if event.timestamp == 1 then @@ -2658,7 +2686,7 @@ fn timer_apis_share_one_slot_in_top_level_and_source_main() -> io::Result<()> { assert(hasTimeout()) elseif event.timestamp == 2 then assert(hasTimeout()) - clearTimerTask() + clearTimeout() assert(not hasTimeout()) end end @@ -2738,7 +2766,11 @@ fn set_timeout_accepts_all_non_negative_lua_integers() -> io::Result<()> { setTimeout(0) assert(hasTimeout()) - setTimeout(9223372036854775807) + local max_ok, max_reason = pcall(setTimeout, 9223372036854775807) + if not max_ok then + assert(max_reason == "setTimeout delay is out of range") + setTimeout(0) + end assert(hasTimeout()) function main(event) @@ -2751,9 +2783,240 @@ fn set_timeout_accepts_all_non_negative_lua_integers() -> io::Result<()> { let Some(schedule) = vm.scheduled_timer() else { return Err(io::Error::other("maximum timeout schedule is missing")); }; + assert!( + schedule.delay == Duration::ZERO + || schedule.delay == Duration::from_millis(i64::MAX.unsigned_abs()) + ); + Ok(()) +} + +#[test] +fn named_timers_replace_clear_and_expose_their_id() -> io::Result<()> { + let mut vm = load_vm( + r#" + setTimeout(20, "flush") + setTimeout(10, "heartbeat") + assert(hasTimeout("flush")) + assert(hasTimeout("heartbeat")) + clearTimeout("flush") + assert(not hasTimeout("flush")) + function main(event) + assert(event.type == "timer") + assert(event.id == "heartbeat") + assert(event.eligibleAt >= 10) + assert(not hasTimeout("heartbeat")) + end + "#, + limits()?, + ) + .map_err(test_error)?; + let event = vm + .begin_timer() + .map_err(|_| io::Error::other("timer missing"))?; + vm.call_timer_event(10, event) + .into_result_without_emit_boundaries() + .map_err(test_error) +} + +#[test] +fn zero_delay_timers_are_dispatched_in_registration_order() -> io::Result<()> { + let mut vm = load_vm( + r#" + setTimeout(0, "first") + setTimeout(0, "second") + setTimeout(0) + function main(event) + assert(event.type == "timer") + end + "#, + limits()?, + ) + .map_err(test_error)?; + let first = vm + .begin_timer() + .map_err(|_| io::Error::other("first timer missing"))?; + assert_eq!(first.id.as_deref(), Some("first")); + vm.call_timer_event(0, first) + .into_result_without_emit_boundaries() + .map_err(test_error)?; + let second = vm + .begin_timer() + .map_err(|_| io::Error::other("second timer missing"))?; + assert_eq!(second.id.as_deref(), Some("second")); + vm.call_timer_event(0, second) + .into_result_without_emit_boundaries() + .map_err(test_error)?; + let anonymous = vm + .begin_timer() + .map_err(|_| io::Error::other("anonymous timer missing"))?; + assert_eq!(anonymous.id, None); + Ok(()) +} + +#[test] +fn named_timer_ids_are_validated_without_changing_existing_timers() -> io::Result<()> { + load_vm( + r#" + local valid = string.rep("é", 64) + setTimeout(0, valid) + setTimeout(0) + for _, id in ipairs({"", true, 1, string.rep("é", 65), string.char(255)}) do + for _, call in ipairs({ + function() setTimeout(0, id) end, + function() clearTimeout(id) end, + function() hasTimeout(id) end + }) do + local ok, reason = pcall(call) + assert(not ok) + assert(reason == "timer id must be a non-empty string of at most 128 bytes") + end + end + assert(hasTimeout(valid)) + assert(hasTimeout(nil)) + clearTimeout(valid) + assert(not hasTimeout(valid)) + assert(hasTimeout()) + assert(select('#', clearTimeout(nil)) == 0) + assert(select('#', clearTimeout("missing")) == 0) + assert(not hasTimeout()) + function main(event) end + "#, + limits()?, + ) + .map_err(test_error)?; + Ok(()) +} + +#[test] +fn timer_memory_charge_is_fixed_per_pending_timer_plus_utf8_id_bytes() -> io::Result<()> { + let mut vm = load_vm( + r#" + setTimeout(0, "é") + function main(event) + setTimeout(0) + setTimeout(0, "é") + end + "#, + limits()?, + ) + .map_err(test_error)?; + let budget = Rc::clone(&vm.native_memory_budget); + // The public charge is 2 KiB per timer plus three copies of its UTF-8 id. + assert_eq!(budget.used_bytes(), 2048 + 3 * 2); + call_source(&mut vm, 1, empty_source_payload())? + .into_result_without_emit_boundaries() + .map_err(test_error)?; + assert_eq!(budget.used_bytes(), 2 * 2048 + 3 * 2); + let anonymous = vm + .begin_timer() + .map_err(|_| io::Error::other("anonymous timer missing"))?; + assert_eq!(anonymous.id, None); + assert_eq!(budget.used_bytes(), 2048 + 3 * 2); + let named = vm + .begin_timer() + .map_err(|_| io::Error::other("named timer missing"))?; + assert_eq!(named.id.as_deref(), Some("é")); + assert_eq!(budget.used_bytes(), 0); + Ok(()) +} + +#[test] +fn timer_memory_is_reused_on_replace_and_released_on_clear_dispatch_and_drop() -> io::Result<()> { + let mut vm = load_vm( + r#" + function main(event) + for i = 1, 128 do setTimeout(0, tostring(i)) end + if event.timestamp == 2 then + for i = 1, 128 do clearTimeout(tostring(i)) end + end + end + "#, + limits()?, + ) + .map_err(test_error)?; + let budget = Rc::clone(&vm.native_memory_budget); + assert_eq!(budget.used_bytes(), 0); + call_source(&mut vm, 1, empty_source_payload())? + .into_result_without_emit_boundaries() + .map_err(test_error)?; + let allocated = budget.used_bytes(); + assert!(allocated > 0); + call_source(&mut vm, 1, empty_source_payload())? + .into_result_without_emit_boundaries() + .map_err(test_error)?; + assert_eq!(budget.used_bytes(), allocated); + call_source(&mut vm, 2, empty_source_payload())? + .into_result_without_emit_boundaries() + .map_err(test_error)?; + assert_eq!(budget.used_bytes(), 0); + call_source(&mut vm, 1, empty_source_payload())? + .into_result_without_emit_boundaries() + .map_err(test_error)?; + for _ in 0..128 { + vm.begin_timer() + .map_err(|_| io::Error::other("timer missing"))?; + } + assert_eq!(budget.used_bytes(), 0); + call_source(&mut vm, 1, empty_source_payload())? + .into_result_without_emit_boundaries() + .map_err(test_error)?; + drop(vm); + assert_eq!(budget.used_bytes(), 0); + Ok(()) +} + +#[test] +fn timer_memory_exhaustion_escapes_protected_calls() -> io::Result<()> { + let mut vm = load_vm( + r#" + function main(event) + pcall(function() + for i = 1, 100000 do setTimeout(0, tostring(i)) end + end) + error("memory failure was swallowed") + end + "#, + limits()?, + ) + .map_err(test_error)?; + let Err(error) = + call_source(&mut vm, 1, empty_source_payload())?.into_result_without_emit_boundaries() + else { + return Err(io::Error::other("timer budget must be enforced")); + }; + assert_eq!(error.kind(), LuaVmErrorKind::MemoryExceeded); + assert!(vm.next_timer_event().is_none()); + Ok(()) +} + +#[test] +fn timer_delay_max_integer_rejects_overflow_and_preserves_the_old_timer() -> io::Result<()> { + let origin = Instant::now() + .checked_sub(Duration::from_millis(2)) + .ok_or_else(|| io::Error::other("test origin out of range"))?; + let vm = LuaVm::load_observed_at( + r#" + setTimeout(7) + local ok, message = pcall(setTimeout, 9223372036854775807) + assert(not ok) + assert(message == "setTimeout delay is out of range") + assert(hasTimeout()) + function main(event) end + "#, + limits()?, + NonZeroU64::new(262_144).ok_or_else(|| io::Error::other("record limit"))?, + lua_source_contract()?, + HashMap::new(), + None, + || false, + None, + origin, + Rc::new(std::cell::Cell::new(0)), + ) + .map_err(test_error)?; assert_eq!( - schedule.delay, - Duration::from_millis(i64::MAX.unsigned_abs()) + vm.scheduled_timer().map(|timer| timer.delay), + Some(Duration::from_millis(7)) ); Ok(()) } @@ -2798,8 +3061,13 @@ fn failed_main_does_not_expose_its_timer_schedule() -> io::Result<()> { let Some(_schedule) = vm.scheduled_timer() else { return Err(io::Error::other("startup timeout schedule is missing")); }; - assert!(vm.begin_timer().is_ok()); - let Err(error) = vm.call_timer(1).into_result_without_emit_boundaries() else { + let event = vm + .begin_timer() + .map_err(|_| io::Error::other("timer missing"))?; + let Err(error) = vm + .call_timer_event(1, event) + .into_result_without_emit_boundaries() + else { return Err(io::Error::other("failing main should be rejected")); }; assert_eq!(error.kind(), LuaVmErrorKind::MainFailed); @@ -2813,7 +3081,7 @@ fn protects_predefined_globals_and_libraries() -> io::Result<()> { "_VERSION = \"changed\"; function main(event) end", "_G._VERSION = \"changed\"; function main(event) end", "setTimeout = nil; function main(event) end", - "clearTimerTask = nil; function main(event) end", + "clearTimeout = nil; function main(event) end", "hasTimeout = nil; function main(event) end", "currentTimeMillis = nil; function main(event) end", "os = {}; function main(event) end", @@ -4063,9 +4331,11 @@ fn main_receives_exactly_one_timer_event_argument() -> io::Result<()> { for _ in pairs(event) do count = count + 1 end - assert(count == 2) + assert(count == 3) assert(event.type == "timer") assert(event.timestamp == 27) + assert(event.id == nil) + assert(event.eligibleAt == 27) assert(event.payload == nil) end "##, diff --git a/src/lua/timer.rs b/src/lua/timer.rs index 70f1cd2..92fa3fa 100644 --- a/src/lua/timer.rs +++ b/src/lua/timer.rs @@ -17,26 +17,29 @@ * under the License. */ -//! Single-slot timer state shared by one Lua VM and its event-loop owner. -//! -//! Lua calls only mutate this logical slot. They never run `main` or create an -//! asynchronous task. The owning Flow Channel reads the one schedule after -//! a successful top-level load or `main` call and uses its remaining duration -//! for the next Submission Queue wait. Replacement and cancellation mutate -//! that same fact before the owner chooses its next event, so no update queue, -//! stale callback, or generation token exists. +//! One-shot timers shared by one Lua VM and its event-loop owner. use super::{ ExecutionBudget, LuaApiFailure, LuaVmFatalFault, create_catchable_api_wrapper_factory, - finish_api_call, protect_name, + finish_api_call, memory::LuaNativeMemoryBudget, protect_name, }; -use mlua::{Function, Lua, Table, Value}; +use mlua::{Function, Lua, MultiValue, Table, Value}; use std::cell::{Cell, RefCell}; -use std::collections::HashSet; +use std::collections::{BTreeMap, HashSet}; use std::rc::Rc; +use std::str; use std::time::{Duration, Instant}; const DELAY_ERROR: &str = "setTimeout delay must be a non-negative integer"; +const DELAY_RANGE_ERROR: &str = "setTimeout delay is out of range"; +const TIMER_ID_ERROR: &str = "timer id must be a non-empty string of at most 128 bytes"; +const MAX_TIMER_ID_BYTES: usize = 128; +const TIMER_ID_STORAGE_COPIES: usize = 3; +// Charge each pending timer a fixed 2 KiB for its entry and indexes, plus +// its three owned id copies. This includes a coarse container allowance; +// it is not an exact allocation or RSS measurement and does not depend on +// BTreeMap's internal node layout. +const TIMER_ENTRY_CHARGE_BYTES: usize = 2 * 1024; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) struct TimerSchedule { @@ -44,42 +47,250 @@ pub(crate) struct TimerSchedule { pub(crate) delay: Duration, } +#[derive(Clone, Debug)] +pub(crate) struct TimerEvent { + pub(crate) schedule: TimerSchedule, + pub(crate) deadline: Instant, + pub(crate) eligible_at: i64, + pub(crate) id: Option>, + pub(crate) sequence: u64, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) struct TimerSlotInactive; -#[derive(Clone, Debug, Default)] +#[derive(Debug, Default)] +struct TimerState { + anonymous: Option, + named: BTreeMap, + order: BTreeMap<(Instant, u64), Option>, +} + +impl TimerState { + fn release_empty_storage(&mut self) { + if self.order.is_empty() { + // Empty BTreeMaps can retain their last leaves. Release them when + // no pending timer remains to carry the container allowance. + self.named = BTreeMap::new(); + self.order = BTreeMap::new(); + } + } +} + +#[derive(Debug)] +struct TimerEntry { + schedule: TimerSchedule, + deadline: Instant, + eligible_at: i64, + id: Option>, + sequence: u64, + _memory: TimerMemoryReservation, +} + +#[derive(Debug)] +struct TimerMemoryReservation { + budget: Rc, + bytes: usize, +} + +impl TimerMemoryReservation { + fn reserve(budget: Rc, bytes: usize) -> Result { + budget.replace(0, bytes)?; + Ok(Self { budget, bytes }) + } +} + +impl Drop for TimerMemoryReservation { + fn drop(&mut self) { + self.budget.release(self.bytes); + } +} + +#[derive(Clone, Debug)] pub(super) struct TimerSlot { - schedule: Rc>>, + state: Rc>, + origin: Instant, + sequence: Rc>, } impl TimerSlot { - pub(super) fn schedule(&self) -> Option { - self.schedule.get() + pub(super) fn new(origin: Instant, sequence: Rc>) -> Self { + Self { + state: Rc::new(RefCell::new(TimerState::default())), + origin, + sequence, + } + } + + pub(super) fn next_event(&self) -> Option { + let state = self.state.borrow(); + let (_, id) = state.order.first_key_value()?; + let entry = match id { + None => state.anonymous.as_ref()?, + Some(id) => state.named.get(id)?, + }; + Some(TimerEvent { + schedule: entry.schedule, + deadline: entry.deadline, + eligible_at: entry.eligible_at, + id: entry.id.clone(), + sequence: entry.sequence, + }) + } + + pub(super) fn next_event_sequence(&self) -> Result { + let sequence = self.sequence.get().checked_add(1).ok_or(())?; + self.sequence.set(sequence); + Ok(sequence) } - pub(super) fn begin(&self) -> Result<(), TimerSlotInactive> { - if self.schedule.take().is_none() { - return Err(TimerSlotInactive); + pub(super) fn begin(&self) -> Result { + let mut state = self.state.borrow_mut(); + let (order_key, key) = state + .order + .first_key_value() + .map(|(key, id)| (*key, id.clone())) + .ok_or(TimerSlotInactive)?; + let entry = match key { + None => state.anonymous.take(), + Some(id) => state.named.remove(&id), } - Ok(()) + .ok_or(TimerSlotInactive)?; + state.order.remove(&order_key); + state.release_empty_storage(); + let event = TimerEvent { + schedule: entry.schedule, + deadline: entry.deadline, + eligible_at: entry.eligible_at, + id: entry.id, + sequence: entry.sequence, + }; + Ok(event) } - pub(super) fn has_timeout(&self) -> bool { - self.schedule.get().is_some() + pub(super) fn has_timeout(&self, id: Option<&str>) -> bool { + let state = self.state.borrow(); + match id { + None => state.anonymous.is_some(), + Some(id) => state.named.contains_key(id), + } } - fn set_timeout(&self, delay: Duration) { - self.schedule.set(Some(TimerSchedule { - scheduled_at: Instant::now(), + fn set_timeout( + &self, + budget: Rc, + delay: Duration, + id: Option, + ) -> Result<(), LuaApiFailure> { + let scheduled_at = Instant::now(); + let deadline = checked_deadline(scheduled_at, delay)?; + let eligible_at = i64::try_from( + deadline + .checked_duration_since(self.origin) + .ok_or(LuaApiFailure::Api(DELAY_RANGE_ERROR))? + .as_millis(), + ) + .map_err(|_| LuaApiFailure::Api(DELAY_RANGE_ERROR))?; + let mut state = self.state.borrow_mut(); + let sequence = self + .next_event_sequence() + .map_err(|_| LuaApiFailure::Api(DELAY_RANGE_ERROR))?; + let schedule = TimerSchedule { + scheduled_at, delay, - })); + }; + let target = id.as_deref(); + let old_key = match target { + None => state + .anonymous + .as_ref() + .map(|current| (current.deadline, current.sequence)), + Some(id) => state + .named + .get(id) + .map(|current| (current.deadline, current.sequence)), + }; + if let Some(old_key) = old_key { + let expected_id = id.clone(); + if state.order.get(&old_key) != Some(&expected_id) { + return Err(LuaApiFailure::InternalInvariantViolation); + } + { + let current = match target { + None => state + .anonymous + .as_mut() + .ok_or(LuaApiFailure::InternalInvariantViolation)?, + Some(id) => state + .named + .get_mut(id) + .ok_or(LuaApiFailure::InternalInvariantViolation)?, + }; + // The same id has the same charge, so keep its reservation. + current.schedule = schedule; + current.deadline = deadline; + current.eligible_at = eligible_at; + current.id = id.as_ref().map(|value| value.clone().into_boxed_str()); + current.sequence = sequence; + } + state.order.remove(&old_key); + state.order.insert((deadline, sequence), expected_id); + return Ok(()); + } + + let id_bytes = id.as_ref().map_or(Ok(0), |value| { + value + .len() + .checked_mul(TIMER_ID_STORAGE_COPIES) + .ok_or(LuaApiFailure::MemoryExceeded) + })?; + let bytes = TIMER_ENTRY_CHARGE_BYTES + .checked_add(id_bytes) + .ok_or(LuaApiFailure::MemoryExceeded)?; + let memory = TimerMemoryReservation::reserve(budget, bytes)?; + let entry = TimerEntry { + schedule, + deadline, + eligible_at, + id: id.clone().map(String::into_boxed_str), + sequence, + _memory: memory, + }; + + if let Some(id) = id { + state.order.insert((deadline, sequence), Some(id.clone())); + state.named.insert(id, entry); + } else { + state.order.insert((deadline, sequence), None); + state.anonymous = Some(entry); + } + Ok(()) } - fn clear(&self) { - self.schedule.set(None); + fn clear(&self, id: Option<&str>) { + let mut state = self.state.borrow_mut(); + match id { + None => { + if let Some(entry) = state.anonymous.take() { + state.order.remove(&(entry.deadline, entry.sequence)); + } + } + Some(id) => { + if let Some(entry) = state.named.remove(id) { + state.order.remove(&(entry.deadline, entry.sequence)); + } + } + } + state.release_empty_storage(); } } +fn checked_deadline(scheduled_at: Instant, delay: Duration) -> Result { + scheduled_at + .checked_add(delay) + .ok_or(LuaApiFailure::Api(DELAY_RANGE_ERROR)) +} + pub(super) fn install( lua: &Lua, environment_values: &Table, @@ -87,24 +298,22 @@ pub(super) fn install( fatal_fault: Rc>>, execution_budget: Rc>>, slot: TimerSlot, + memory_budget: Rc, ) -> mlua::Result<()> { let api_wrapper_factory = create_catchable_api_wrapper_factory(lua)?; let set_slot = slot.clone(); + let set_budget = Rc::clone(&memory_budget); let set_fault = Rc::clone(&fatal_fault); - let set_budget = Rc::clone(&execution_budget); - let native_set = lua.create_function(move |lua, value: Value| { - let result = match value { - Value::Integer(delay) if delay >= 0 => u64::try_from(delay) - .map(Duration::from_millis) - .map_err(|_| LuaApiFailure::Api(DELAY_ERROR)) - .map(|delay| { - set_slot.set_timeout(delay); - Value::Nil - }), - _ => Err(LuaApiFailure::Api(DELAY_ERROR)), - }; - finish_api_call(lua, result, &set_fault, &set_budget) + let set_execution_budget = Rc::clone(&execution_budget); + let native_set = lua.create_function(move |lua, mut arguments: MultiValue| { + let result = parse_delay(&mut arguments).and_then(|delay| { + let id = parse_optional_id(arguments.pop_front())?; + set_slot + .set_timeout(Rc::clone(&set_budget), delay, id) + .map(|()| Value::Nil) + }); + finish_api_call(lua, result, &set_fault, &set_execution_budget) })?; environment_values.raw_set( "setTimeout", @@ -113,19 +322,163 @@ pub(super) fn install( protect_name(&protected_names, "setTimeout"); let clear_slot = slot.clone(); - environment_values.raw_set( - "clearTimerTask", - lua.create_function(move |_, ()| { - clear_slot.clear(); - Ok(()) - })?, - )?; - protect_name(&protected_names, "clearTimerTask"); + let clear_fault = Rc::clone(&fatal_fault); + let clear_execution_budget = Rc::clone(&execution_budget); + let native_clear = lua.create_function(move |lua, mut arguments: MultiValue| { + let result = parse_optional_id(arguments.pop_front()).map(|id| { + clear_slot.clear(id.as_deref()); + Value::Nil + }); + finish_api_call(lua, result, &clear_fault, &clear_execution_budget) + })?; + // Preserve the cancellation API's zero return values after catchable error handling. + let clear_timeout: Function = lua + .load("local clear = ...; return function(...) clear(...) end") + .call(api_wrapper_factory.call::(native_clear)?)?; + environment_values.raw_set("clearTimeout", clear_timeout)?; + protect_name(&protected_names, "clearTimeout"); + let has_slot = slot; + let has_fault = Rc::clone(&fatal_fault); + let has_execution_budget = Rc::clone(&execution_budget); + let native_has = lua.create_function(move |lua, mut arguments: MultiValue| { + let result = parse_optional_id(arguments.pop_front()) + .map(|id| Value::Boolean(has_slot.has_timeout(id.as_deref()))); + finish_api_call(lua, result, &has_fault, &has_execution_budget) + })?; environment_values.raw_set( "hasTimeout", - lua.create_function(move |_, ()| Ok(slot.has_timeout()))?, + api_wrapper_factory.call::(native_has)?, )?; protect_name(&protected_names, "hasTimeout"); Ok(()) } + +fn parse_delay(arguments: &mut MultiValue) -> Result { + match arguments.pop_front() { + Some(Value::Integer(delay)) if delay >= 0 => u64::try_from(delay) + .map(Duration::from_millis) + .map_err(|_| LuaApiFailure::Api(DELAY_RANGE_ERROR)), + _ => Err(LuaApiFailure::Api(DELAY_ERROR)), + } +} + +fn parse_optional_id(value: Option) -> Result, LuaApiFailure> { + match value { + None | Some(Value::Nil) => Ok(None), + Some(Value::String(value)) => { + let bytes = value.as_bytes(); + if bytes.is_empty() || bytes.len() > MAX_TIMER_ID_BYTES { + return Err(LuaApiFailure::Api(TIMER_ID_ERROR)); + } + let value = str::from_utf8(&bytes).map_err(|_| LuaApiFailure::Api(TIMER_ID_ERROR))?; + Ok(Some(value.to_owned())) + } + Some(_) => Err(LuaApiFailure::Api(TIMER_ID_ERROR)), + } +} + +#[cfg(test)] +mod tests { + use super::{DELAY_RANGE_ERROR, LuaApiFailure, checked_deadline}; + use std::time::{Duration, Instant}; + + #[test] + fn equal_deadlines_use_sequence_and_dispatch_releases_all_memory() -> std::io::Result<()> { + let vm = super::super::tests::load_vm( + "setTimeout(0, 'z'); setTimeout(0, 'a'); setTimeout(0); function main(event) end", + super::super::tests::limits()?, + ) + .map_err(|error| std::io::Error::other(error.to_string()))?; + // Equal clock readings are valid on a coarse monotonic clock. Keep the + // real registrations and reservations, and give them that same reading. + let deadline = Instant::now(); + let eligible_at = i64::try_from(deadline.duration_since(vm.timer_slot.origin).as_millis()) + .map_err(std::io::Error::other)?; + { + let mut state = vm.timer_slot.state.borrow_mut(); + let mut keys = Vec::new(); + for (id, entry) in &mut state.named { + entry.deadline = deadline; + entry.schedule.scheduled_at = deadline; + entry.eligible_at = eligible_at; + keys.push(((deadline, entry.sequence), Some(id.clone()))); + } + let anonymous = state + .anonymous + .as_mut() + .ok_or_else(|| std::io::Error::other("anonymous timer missing"))?; + anonymous.deadline = deadline; + anonymous.schedule.scheduled_at = deadline; + anonymous.eligible_at = eligible_at; + keys.push(((deadline, anonymous.sequence), None)); + state.order = keys.into_iter().collect(); + } + for expected in [Some("z"), Some("a"), None] { + let event = vm + .begin_timer() + .map_err(|_| std::io::Error::other("pending timer missing"))?; + assert_eq!(event.id.as_deref(), expected); + } + let state = vm.timer_slot.state.borrow(); + assert!(state.order.is_empty()); + assert!(state.named.is_empty()); + assert_eq!(vm.native_memory_budget.used_bytes(), 0); + Ok(()) + } + + #[test] + fn failed_registration_preserves_the_existing_timer_and_memory_charge() -> std::io::Result<()> { + let limits = super::super::tests::limits()?; + let vm = super::super::tests::load_vm( + "setTimeout(0, 'existing'); function main(event) end", + limits, + ) + .map_err(|error| std::io::Error::other(error.to_string()))?; + let budget = std::rc::Rc::clone(&vm.native_memory_budget); + let existing_charge = budget.used_bytes(); + // Leave one byte less than the new timer needs. Registration must + // fail without changing either index or its existing reservation. + let remaining = 2 * 1024 + 3 * "first".len() - 1; + let retained = + limits.memory_bytes().get() - vm.lua.used_memory() - existing_charge - remaining; + let reservation = + super::TimerMemoryReservation::reserve(std::rc::Rc::clone(&budget), retained) + .map_err(|_| std::io::Error::other("initial reservation failed"))?; + assert!(matches!( + vm.timer_slot.set_timeout( + std::rc::Rc::clone(&budget), + Duration::ZERO, + Some("first".to_owned()) + ), + Err(LuaApiFailure::MemoryExceeded) + )); + assert_eq!(budget.used_bytes(), existing_charge + retained); + assert!(vm.timer_slot.has_timeout(Some("existing"))); + assert!(!vm.timer_slot.has_timeout(Some("first"))); + assert_eq!(vm.timer_slot.state.borrow().named.len(), 1); + assert_eq!(vm.timer_slot.state.borrow().order.len(), 1); + // Replacement needs no extra budget, even with too little for a new id. + vm.timer_slot + .set_timeout( + std::rc::Rc::clone(&budget), + Duration::ZERO, + Some("existing".to_owned()), + ) + .map_err(|_| std::io::Error::other("replacement failed"))?; + assert_eq!(budget.used_bytes(), existing_charge + retained); + drop(reservation); + assert_eq!(budget.used_bytes(), existing_charge); + vm.timer_slot.clear(Some("existing")); + assert_eq!(budget.used_bytes(), 0); + Ok(()) + } + + #[test] + fn deadline_overflow_is_reported_as_the_catchable_range_error() { + assert!(matches!( + checked_deadline(Instant::now(), Duration::MAX), + Err(LuaApiFailure::Api(DELAY_RANGE_ERROR)) + )); + } +} diff --git a/src/pipeline/channel/flow_channel/mod.rs b/src/pipeline/channel/flow_channel/mod.rs index e2f02be..9141211 100644 --- a/src/pipeline/channel/flow_channel/mod.rs +++ b/src/pipeline/channel/flow_channel/mod.rs @@ -69,7 +69,9 @@ //! every Channel has installed the same replacement. use std::borrow::Cow; +use std::cell::Cell; use std::fmt; +use std::rc::Rc; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -141,6 +143,13 @@ enum ChannelWork { SourceRecord, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct SourceReadiness { + observed_at: Instant, + order: u64, + generation: u64, +} + /// Thread-affine Lua and Source resources prepared before the old writer handoff. pub(crate) struct PreparedFlowChannel { ingress: IngressQueuePair, @@ -155,6 +164,7 @@ pub(crate) struct PreparedFlowChannel { bell: Arc, control: Arc, commands: ChannelCommandInbox, + event_order: Rc>, } impl PreparedFlowChannel { @@ -183,6 +193,9 @@ impl PreparedFlowChannel { control: self.control, commands: self.commands, candidate: None, + source_readiness: Cell::new(None), + next_event_order: self.event_order, + vm_generation: Cell::new(0), }; channel.metrics.bind(); channel.bind_route_generation(); @@ -207,6 +220,9 @@ pub(crate) struct FlowChannel { control: Arc, commands: ChannelCommandInbox, candidate: Option, + source_readiness: Cell>, + next_event_order: Rc>, + vm_generation: Cell, } impl FlowChannel { @@ -240,11 +256,16 @@ impl FlowChannel { let ingress = IngressQueuePair::open(submission_path, completion_path, &bell, &source_region)?; let (commands, command_control) = control_pair(bell.interrupter()); + let event_order = Rc::new(Cell::new(0)); let execution_control = Arc::clone(&control); let (lua_vm, lua_vm_diagnostics) = spec - .load_vm(diagnostics.clone(), &metrics, move || { - startup_aborted() || execution_control.is_stopping() - }) + .load_vm( + diagnostics.clone(), + &metrics, + move || startup_aborted() || execution_control.is_stopping(), + pipeline_started_at, + Rc::clone(&event_order), + ) .map_err(|source| FlowChannelError::LuaVmLoad { kind: source.kind(), })?; @@ -263,6 +284,7 @@ impl FlowChannel { bell, control, commands, + event_order, }, wake, command_control, @@ -338,9 +360,9 @@ impl FlowChannel { /// dispatch set by construction: a new fact must be added here, and its /// `match` arm in [`Self::run_events`] then cannot be omitted. /// - /// A due Lua timer outranks a ready Source record, because a Source event - /// must observe the script state the timer already produced. A timer that is - /// still waiting is not work here: its exact remaining duration becomes the + /// A due Lua timer and a ready Source record are ordered by their eligible + /// time and the Channel-local sequence tie-breaker. A timer that is still + /// waiting is not work here: its exact remaining duration becomes the /// park's only timeout instead. fn pending_work(&self) -> Result, FlowChannelError> { match self.control.directive() { @@ -359,25 +381,74 @@ impl FlowChannel { if self.commands.has_pending() { return Ok(Some(ChannelWork::Command)); } - if matches!(self.timer_readiness()?, Some(TimerReadiness::Due)) { - return Ok(Some(ChannelWork::TimerDue)); - } - if self + let timer = self.lua_vm()?.next_timer_event(); + let source_ready = self .ingress .readable() - .map_err(|source| FlowChannelError::ChannelWait { source })? - { + .map_err(|source| FlowChannelError::ChannelWait { source })?; + if source_ready { + self.observe_source_readiness()?; + } else { + self.source_readiness.set(None); + } + let timer_due = timer.as_ref().is_some_and(|timer| { + timer_readiness_at(timer.schedule, Instant::now()) == TimerReadiness::Due + }); + if timer_due && source_ready { + let source = self + .source_readiness + .get() + .ok_or(FlowChannelError::InternalInvariantViolation)?; + let timer = timer.ok_or(FlowChannelError::InternalInvariantViolation)?; + let timer_first = timer.deadline < source.observed_at + || (timer.deadline == source.observed_at && timer.sequence <= source.order); + return Ok(Some(if timer_first { + ChannelWork::TimerDue + } else { + ChannelWork::SourceRecord + })); + } + if timer_due { + return Ok(Some(ChannelWork::TimerDue)); + } + if source_ready { return Ok(Some(ChannelWork::SourceRecord)); } Ok(None) } - /// Reports the Lua timer's readiness from the current instant. + fn observe_source_readiness(&self) -> Result<(), FlowChannelError> { + let generation = self.vm_generation.get(); + if self + .source_readiness + .get() + .is_some_and(|readiness| readiness.generation == generation) + { + return Ok(()); + } + let order = self + .next_event_order() + .ok_or(FlowChannelError::InternalInvariantViolation)?; + self.source_readiness.set(Some(SourceReadiness { + observed_at: Instant::now(), + order, + generation, + })); + Ok(()) + } + + fn next_event_order(&self) -> Option { + let next = self.next_event_order.get().checked_add(1)?; + self.next_event_order.set(next); + Some(next) + } + + /// Reports the earliest Lua timer's readiness from the current instant. fn timer_readiness(&self) -> Result, FlowChannelError> { Ok(self .lua_vm()? - .scheduled_timer() - .map(|timer| timer_readiness_at(timer, Instant::now()))) + .next_timer_event() + .map(|timer| timer_readiness_at(timer.schedule, Instant::now()))) } /// Processes at most one already committed Source record without waiting. @@ -392,9 +463,12 @@ impl FlowChannel { /// resource, or private invariant failure. fn process_available_source(&mut self) -> Result, FlowChannelError> { let Some(record) = self.ingress.try_receive(&self.metrics)? else { + self.source_readiness.set(None); return Ok(None); }; - self.process_source_record(record).map(Some) + let result = self.process_source_record(record); + self.source_readiness.set(None); + result.map(Some) } fn prepare_replacement( @@ -414,16 +488,19 @@ impl FlowChannel { "a replacement uses one complete target registry" ); let replacement_control = Arc::clone(&self.control); - let (vm, vm_diagnostics) = - match spec.load_vm(self.diagnostics.clone(), &self.metrics, move || { - replacement_control.is_stopping() - }) { - Ok(prepared) => prepared, - Err(source) => { - drop(routes); - return session.candidate_failed(source.kind()); - } - }; + let (vm, vm_diagnostics) = match spec.load_vm( + self.diagnostics.clone(), + &self.metrics, + move || replacement_control.is_stopping(), + self.pipeline_started_at, + Rc::clone(&self.next_event_order), + ) { + Ok(prepared) => prepared, + Err(source) => { + drop(routes); + return session.candidate_failed(source.kind()); + } + }; PreparedDefinition::Replacement { spec, vm, @@ -495,6 +572,7 @@ impl FlowChannel { self.spec = spec; self.lua_vm = Some(vm); self.lua_vm_diagnostics = Some(vm_diagnostics); + self.reset_event_observations(); } self.routes = bind_routes( routes, @@ -525,6 +603,7 @@ impl FlowChannel { } fn drain_source_records(&mut self) -> Result { + self.source_readiness.set(None); loop { if self.is_stopping() { return Ok(CompletionProgress::Stopped); @@ -954,32 +1033,40 @@ impl FlowChannel { } fn reload_lua_vm(&mut self) -> Result { + self.reset_event_observations(); drop(self.lua_vm.take()); drop(self.lua_vm_diagnostics.take()); let reload_control = Arc::clone(&self.control); - let (lua_vm, lua_vm_diagnostics) = - match self - .spec - .load_vm(self.diagnostics.clone(), &self.metrics, move || { - reload_control.is_stopping() - }) { - Ok(prepared) => prepared, - Err(source) - if self.is_stopping() && source.kind() == LuaVmErrorKind::ExecutionStopped => - { - return Ok(LuaReloadProgress::Stopped); - } - Err(source) => { - return Err(FlowChannelError::LuaVmLoad { - kind: source.kind(), - }); - } - }; + let (lua_vm, lua_vm_diagnostics) = match self.spec.load_vm( + self.diagnostics.clone(), + &self.metrics, + move || reload_control.is_stopping(), + self.pipeline_started_at, + Rc::clone(&self.next_event_order), + ) { + Ok(prepared) => prepared, + Err(source) + if self.is_stopping() && source.kind() == LuaVmErrorKind::ExecutionStopped => + { + return Ok(LuaReloadProgress::Stopped); + } + Err(source) => { + return Err(FlowChannelError::LuaVmLoad { + kind: source.kind(), + }); + } + }; self.lua_vm = Some(lua_vm); self.lua_vm_diagnostics = Some(lua_vm_diagnostics); Ok(LuaReloadProgress::Reloaded) } + fn reset_event_observations(&self) { + self.source_readiness.set(None); + self.vm_generation + .set(self.vm_generation.get().wrapping_add(1)); + } + fn lua_vm_mut(&mut self) -> Result<&mut LuaVm, FlowChannelError> { self.lua_vm .as_mut() @@ -1006,11 +1093,12 @@ impl FlowChannel { return Ok(FlowChannelStep::Stopped); } } - self.lua_vm_mut()? + let timer = self + .lua_vm_mut()? .begin_timer() .map_err(|_| FlowChannelError::InternalInvariantViolation)?; let timestamp = self.timestamp_millis()?; - let outcome = self.lua_vm_mut()?.call_timer(timestamp); + let outcome = self.lua_vm_mut()?.call_timer_event(timestamp, timer); self.finish_lua_outcome(outcome, LuaFailureCompletion::Timer) } diff --git a/src/pipeline/channel/flow_channel/tests/mod.rs b/src/pipeline/channel/flow_channel/tests/mod.rs index 43a1f10..1064972 100644 --- a/src/pipeline/channel/flow_channel/tests/mod.rs +++ b/src/pipeline/channel/flow_channel/tests/mod.rs @@ -494,7 +494,7 @@ fn source_decode_failure_without_older_pending_preserves_lua_state() -> io::Resu function main(event) count = count + 1 if event.payload.deviceId == "before" then - clearTimerTask() + clearTimeout() elseif event.payload.deviceId == "after" then if count ~= 2 or hasTimeout() then error("Lua state was unexpectedly rebuilt") @@ -837,7 +837,11 @@ fn source_session_finish_keeps_the_original_timer_and_lua_state_after_interrupti process_one_source(&mut channel)?, FlowChannelStep::EventProcessed ); - let original_timer = channel.lua_vm().map_err(channel_error)?.scheduled_timer(); + let original_timer = channel + .lua_vm() + .map_err(channel_error)? + .next_timer_event() + .map(|timer| timer.schedule); assert!(original_timer.is_some()); // A non-stop interruption must only recheck the same Completion prefix. wake.wake().map_err(io::Error::other)?; @@ -872,7 +876,11 @@ fn source_session_finish_keeps_the_original_timer_and_lua_state_after_interrupti }; assert_eq!(result??, completion(71, expected)); assert_eq!( - channel.lua_vm().map_err(channel_error)?.scheduled_timer(), + channel + .lua_vm() + .map_err(channel_error)? + .next_timer_event() + .map(|timer| timer.schedule), original_timer ); assert_eq!( @@ -886,7 +894,8 @@ fn source_session_finish_keeps_the_original_timer_and_lua_state_after_interrupti channel .lua_vm() .map_err(channel_error)? - .scheduled_timer() + .next_timer_event() + .map(|timer| timer.schedule) .is_none() ); assert!(source.try_completion()?.is_none()); @@ -1006,7 +1015,7 @@ fn ready_source_runs_before_a_future_timer() -> io::Result<()> { if event.type == "timer" then error("Future timer ran early") end - clearTimerTask() + clearTimeout() emit() end "#, @@ -1022,6 +1031,157 @@ fn ready_source_runs_before_a_future_timer() -> io::Result<()> { channel.stop() } +#[test] +fn timer_source_ties_and_vm_transitions_preserve_event_order() -> io::Result<()> { + let mut source = SourceQueueFixture::new()?; + source.submit(53, source_payload("ready")?)?; + let (diagnostics, _records) = interested_channel(0); + let spec = channel_spec( + "setTimeout(0); setTimeout(0); function main(event) end", + SourceDelivery::AtLeastOnce, + [], + )?; + let (mut channel, _wake, commands) = open_channel( + source.queue_paths(), + source.bells()?, + Instant::now(), + spec, + diagnostics, + HashMap::new(), + Arc::new(FlowChannelControl::new()), + || false, + ) + .map_err(channel_error)?; + let timer = channel + .lua_vm() + .map_err(channel_error)? + .next_timer_event() + .ok_or_else(|| io::Error::other("timer missing"))?; + for (order, expected) in [ + (timer.sequence - 1, super::ChannelWork::SourceRecord), + (timer.sequence + 1, super::ChannelWork::TimerDue), + ] { + channel.source_readiness.set(Some(super::SourceReadiness { + observed_at: timer.deadline, + order, + generation: channel.vm_generation.get(), + })); + assert_eq!( + channel.pending_work().map_err(channel_error)?, + Some(expected) + ); + } + channel.reload_lua_vm().map_err(channel_error)?; + assert!(channel.source_readiness.get().is_none()); + let replacement = channel + .lua_vm() + .map_err(channel_error)? + .next_timer_event() + .ok_or_else(|| io::Error::other("replacement timer missing"))?; + assert!(replacement.sequence > timer.sequence); + assert_eq!( + channel.pending_work().map_err(channel_error)?, + Some(super::ChannelWork::TimerDue) + ); + let old_readiness = channel.source_readiness.get(); + assert!(old_readiness.is_some()); + let (events, observations) = mpsc::sync_channel(8); + let ticket = commands + .begin_replacement( + 0, + ChannelDefinitionChange::Replace(channel.spec.clone()), + HashMap::new(), + events, + ) + .map_err(io::Error::other)?; + let Some(super::ChannelCommand::Replace(request)) = channel.commands.try_take() else { + return Err(io::Error::other("replacement request missing")); + }; + channel + .prepare_replacement(request) + .map_err(channel_error)?; + assert!(matches!( + observations + .recv_timeout(WAIT_LIMIT) + .map_err(io::Error::other)?, + FlowChannelReplacementEvent::Prepared { .. } + )); + assert_eq!(channel.source_readiness.get(), old_readiness); + ticket.cutover().map_err(io::Error::other)?; + thread::scope(|scope| -> io::Result<()> { + let activation = scope.spawn(move || -> io::Result<()> { + assert!(matches!( + observations + .recv_timeout(WAIT_LIMIT) + .map_err(io::Error::other)?, + FlowChannelReplacementEvent::CutoverComplete { .. } + )); + ticket.activate().map_err(io::Error::other) + }); + channel.advance_replacement().map_err(channel_error)?; + activation + .join() + .map_err(|_| io::Error::other("activation thread panicked"))??; + Ok(()) + })?; + assert!(channel.source_readiness.get().is_none()); + assert_eq!( + channel.pending_work().map_err(channel_error)?, + Some(super::ChannelWork::TimerDue) + ); + assert!(channel.source_readiness.get().is_some()); + channel.control.request_drain(); + assert_eq!( + channel.drain_source_records().map_err(channel_error)?, + super::CompletionProgress::Completed + ); + assert!(channel.source_readiness.get().is_none()); + assert_eq!( + source.wait_completion()?, + completion(53, IngressCompletionStatus::Retry) + ); + Ok(()) +} + +#[test] +fn recurring_zero_delay_timer_yields_to_continuously_ready_source() -> io::Result<()> { + let mut source = SourceQueueFixture::new()?; + for record_id in 53..=55 { + source.submit(record_id, source_payload("ready")?)?; + } + let spec = channel_spec( + r#" + local sourceCount = 0 + setTimeout(0) + + function main(event) + if event.type == "timer" then + if sourceCount < 3 then + setTimeout(0) + end + return + end + sourceCount = sourceCount + 1 + emit() + if sourceCount >= 3 then + clearTimeout() + end + end + "#, + SourceDelivery::AtLeastOnce, + [], + )?; + let channel = RunningChannel::start(&source, spec, HashMap::new())?; + + for record_id in 53..=55 { + assert_eq!( + source.wait_completion()?, + completion(record_id, IngressCompletionStatus::Ok) + ); + } + channel.stop() +} + #[test] fn recurring_timer_emits_wall_clock_payloads_without_source_input() -> io::Result<()> { let source = SourceQueueFixture::new()?; diff --git a/src/pipeline/channel/flow_channel_spec.rs b/src/pipeline/channel/flow_channel_spec.rs index bf37484..9ec080c 100644 --- a/src/pipeline/channel/flow_channel_spec.rs +++ b/src/pipeline/channel/flow_channel_spec.rs @@ -23,9 +23,12 @@ //! Lua VM. Runtime routes remain separate live owners; this value only //! verifies that their Sink Contract identities match the frozen Payload registry. +use std::cell::Cell; use std::collections::HashMap; use std::num::NonZeroU64; +use std::rc::Rc; use std::sync::Arc; +use std::time::Instant; use prost_reflect::MessageDescriptor; @@ -86,9 +89,11 @@ impl FlowChannelSpec { diagnostics: ChannelDiagnosticPublisher, metrics: &ChannelMetrics, stop_requested: impl Fn() -> bool + 'static, + timer_origin: Instant, + event_order: Rc>, ) -> Result<(LuaVm, LuaDiagnosticPublisher), LuaVmError> { let lua_vm_diagnostics = diagnostics.lua_vm(); - let lua_vm = LuaVm::load_observed( + let lua_vm = LuaVm::load_observed_at( &self.lua_source, self.lua_limits, self.max_record_bytes, @@ -97,6 +102,8 @@ impl FlowChannelSpec { lua_vm_diagnostics.clone().into_print_callback(), stop_requested, metrics.lua(), + timer_origin, + event_order, ) .inspect_err(|error| { if error.kind() != LuaVmErrorKind::ExecutionStopped {