diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs
index 996b9d90ab3..b73504153c1 100644
--- a/crates/core/src/host/scheduler.rs
+++ b/crates/core/src/host/scheduler.rs
@@ -3,6 +3,7 @@ use super::module_host::{
};
use super::{FunctionArgs, ModuleHost};
use crate::db::relational_db::RelationalDB;
+use crate::error::DBError;
use crate::host::module_host::{CallProcedureParams, ModuleInfo};
use crate::host::wasm_common::module_host_actor::{InstanceCommon, WasmInstance};
use crate::host::{InvalidProcedureArguments, InvalidReducerArguments, NoSuchModule};
@@ -21,7 +22,7 @@ use spacetimedb_datastore::traits::IsolationLevel;
use spacetimedb_lib::scheduler::ScheduleAt;
use spacetimedb_lib::{hash_bytes, Hash, TimeDuration, Timestamp};
use spacetimedb_primitives::{ColId, TableId};
-use spacetimedb_sats::bsatn::ToBsatn as _;
+use spacetimedb_sats::bsatn::{EncodeError as BsatnEncodeError, ToBsatn as _};
use spacetimedb_sats::AlgebraicValue;
use spacetimedb_table::table::RowRef;
use std::panic;
@@ -335,6 +336,34 @@ pub(crate) enum CallScheduledFunctionError {
NoSuchModule(#[from] NoSuchModule),
}
+#[derive(thiserror::Error, Debug)]
+enum ScheduledFunctionParameterError {
+ #[error("could not read scheduled row: {0}")]
+ Datastore(#[from] DBError),
+ #[error("could not encode scheduled row as BSATN: {0}")]
+ BsatnEncoding(#[from] BsatnEncodeError),
+ #[error("Reducer `{0}` not found")]
+ ReducerNotFound(String),
+ #[error("Procedure `{0}` not found")]
+ ProcedureNotFound(String),
+ #[error(transparent)]
+ InvalidReducerArguments(#[from] InvalidReducerArguments),
+ #[error(transparent)]
+ InvalidProcedureArguments(#[from] InvalidProcedureArguments),
+}
+
+impl ScheduledFunctionParameterError {
+ fn is_internal(&self) -> bool {
+ match self {
+ Self::Datastore(_) | Self::BsatnEncoding(_) => true,
+ Self::ReducerNotFound(_)
+ | Self::ProcedureNotFound(_)
+ | Self::InvalidReducerArguments(_)
+ | Self::InvalidProcedureArguments(_) => false,
+ }
+ }
+}
+
impl SchedulerActor {
async fn run(mut self) {
let mut closing = false;
@@ -568,12 +597,11 @@ fn prepare_scheduled_procedure_call(
Ok(None) => return ScheduledProcedureStep::Done(CallScheduledFunctionResult { reschedule: None }, false),
Ok(Some(params)) => params,
Err(err) => {
- // All we can do here is log an error.
- // This can fail because the schedule row could not be read from the datastore,
- // the row could not be BSATN-encoded, the scheduled function no longer exists,
- // or its arguments do not match the current function definition.
- // TODO: Use a typed error to log internal failures at error! and stale/invalid schedules at warn! or lower.
- log::error!("could not determine scheduled procedure or its parameters: {err:#}");
+ if err.is_internal() {
+ log::error!("could not determine scheduled procedure or its parameters: {err:#}");
+ } else {
+ log::warn!("could not determine scheduled procedure or its parameters: {err:#}");
+ }
let reschedule = id.zip(invocation.as_ref()).and_then(|(id, invocation)| {
delete_scheduled_function_row(
module_info,
@@ -638,12 +666,11 @@ fn call_scheduled_reducer_until_done(
Ok(None) => return (CallScheduledFunctionResult { reschedule: None }, false),
Ok(Some(params)) => params,
Err(err) => {
- // All we can do here is log an error.
- // This can fail because the schedule row could not be read from the datastore,
- // the row could not be BSATN-encoded, the scheduled function no longer exists,
- // or its arguments do not match the current function definition.
- // TODO: Use a typed error to log internal failures at error! and stale/invalid schedules at warn! or lower.
- log::error!("could not determine scheduled reducer or its parameters: {err:#}");
+ if err.is_internal() {
+ log::error!("could not determine scheduled reducer or its parameters: {err:#}");
+ } else {
+ log::warn!("could not determine scheduled reducer or its parameters: {err:#}");
+ }
let reschedule = id.zip(invocation.as_ref()).and_then(|(id, invocation)| {
delete_scheduled_function_row(
module_info,
@@ -924,7 +951,7 @@ fn reducer_call_params_for_queued_item(
db: &RelationalDB,
tx: &MutTxId,
item: QueueItem,
-) -> anyhow::Result