fixes

Author: Shubham8287

Cargo.lock

@@ -287,7 +287,7 @@ pub async fn call_from_database<S: ControlStateDelegate + NodeDelegate>(
         .await
         .map_err(client_connected_error_to_response)?;
 
-    let result = module
+    let mut result = module
         .call_reducer_from_database(
             caller_identity,
             Some(connection_id),
@@ -301,6 +301,14 @@ pub async fn call_from_database<S: ControlStateDelegate + NodeDelegate>(
         )
         .await;
 
+    if let Ok(rcr) = result.as_mut()
+        && let Some(tx_offset) = rcr.tx_offset.as_mut()
+        && let Some(mut durable_offset) = module.durable_tx_offset()
+    {
+        let tx_offset = tx_offset.await.map_err(|_| log_and_500("transaction aborted"))?;
+        durable_offset.wait_for(tx_offset).await.map_err(log_and_500)?;
+    }
+
     module
         .call_identity_disconnected(caller_identity, connection_id)
         .await
@@ -311,7 +319,7 @@ pub async fn call_from_database<S: ControlStateDelegate + NodeDelegate>(
             let (status, body) = match rcr.outcome {
                 ReducerOutcome::Committed => (StatusCode::OK, "".into()),
                 ReducerOutcome::Deduplicated => (StatusCode::OK, "deduplicated".into()),
-                // 422 = reducer ran but returned Err; IDC runtime uses this to distinguish
+                // 422 = reducer ran but returned Err; the IDC actor uses this to distinguish
                 // reducer failures from transport errors (which it retries).
                 ReducerOutcome::Failed(errmsg) => (StatusCode::UNPROCESSABLE_ENTITY, *errmsg),
                 ReducerOutcome::BudgetExceeded => {

crates/client-api/src/routes/database.rs

@@ -1,4 +1,4 @@
-use super::idc_runtime::{IdcRuntime, IdcRuntimeConfig, IdcRuntimeStarter, IdcSender};
+use super::idc_actor::{IdcActor, IdcActorConfig, IdcActorSender, IdcActorStarter};
 use super::module_host::{EventStatus, ModuleHost, ModuleInfo, NoSuchModule};
 use super::scheduler::SchedulerStarter;
 use super::wasmtime::WasmtimeRuntime;
@@ -133,11 +133,12 @@ impl HostRuntimes {
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 pub struct ReducerCallResult {
     pub outcome: ReducerOutcome,
     pub energy_used: EnergyQuanta,
     pub execution_duration: Duration,
+    pub tx_offset: Option<TransactionOffset>,
 }
 
 impl ReducerCallResult {
@@ -710,7 +711,7 @@ async fn make_module_host(
     runtimes: Arc<HostRuntimes>,
     replica_ctx: Arc<ReplicaContext>,
     scheduler: Scheduler,
-    idc_sender: IdcSender,
+    idc_sender: IdcActorSender,
     program: Program,
     energy_monitor: Arc<dyn EnergyMonitor>,
     unregister: impl Fn() + Send + Sync + 'static,
@@ -764,7 +765,7 @@ struct LaunchedModule {
     module_host: ModuleHost,
     scheduler: Scheduler,
     scheduler_starter: SchedulerStarter,
-    idc_starter: IdcRuntimeStarter,
+    idc_starter: IdcActorStarter,
 }
 
 struct ModuleLauncher<F> {
@@ -801,7 +802,7 @@ impl<F: Fn() + Send + Sync + 'static> ModuleLauncher<F> {
         .await
         .map(Arc::new)?;
         let (scheduler, scheduler_starter) = Scheduler::open(replica_ctx.relational_db().clone());
-        let (idc_starter, idc_sender) = IdcRuntime::open();
+        let (idc_starter, idc_sender) = IdcActor::open();
         let (program, module_host) = make_module_host(
             self.runtimes.clone(),
             replica_ctx.clone(),
@@ -889,9 +890,9 @@ struct Host {
     /// Handle to the task responsible for cleaning up old views.
     /// The task is aborted when [`Host`] is dropped.
     view_cleanup_task: AbortHandle,
-    /// IDC runtime: delivers outbound inter-database messages from `st_outbound_msg`.
+    /// IDC actor: delivers outbound inter-database messages from `st_outbound_msg`.
     /// Stopped when [`Host`] is dropped.
-    _idc_runtime: IdcRuntime,
+    _idc_actor: IdcActor,
 }
 
 impl Host {
@@ -1112,9 +1113,9 @@ impl Host {
         let disk_metrics_recorder_task = tokio::spawn(metric_reporter(replica_ctx.clone())).abort_handle();
         let view_cleanup_task = spawn_view_cleanup_loop(replica_ctx.relational_db().clone());
 
-        let idc_runtime = idc_starter.start(
+        let idc_actor = idc_starter.start(
             replica_ctx.relational_db().clone(),
-            IdcRuntimeConfig {
+            IdcActorConfig {
                 sender_identity: replica_ctx.database_identity,
             },
             module_host.downgrade(),
@@ -1129,7 +1130,7 @@ impl Host {
             disk_metrics_recorder_task,
             tx_metrics_recorder_task,
             view_cleanup_task,
-            _idc_runtime: idc_runtime,
+            _idc_actor: idc_actor,
         })
     }
 
@@ -1202,7 +1203,7 @@ impl Host {
     ) -> anyhow::Result<UpdateDatabaseResult> {
         let replica_ctx = &self.replica_ctx;
         let (scheduler, scheduler_starter) = Scheduler::open(self.replica_ctx.relational_db().clone());
-        let (_idc_starter, idc_sender) = IdcRuntime::open();
+        let (_idc_starter, idc_sender) = IdcActor::open();
 
         let (program, module) = make_module_host(
             runtimes,

crates/core/src/host/host_controller.rs

@@ -1,4 +1,4 @@
-/// Inter-Database Communication (IDC) Runtime
+/// Inter-Database Communication (IDC) Actor
 ///
 /// Background tokio task that:
 /// 1. Loads undelivered entries from `st_outbound_msg` on startup, resolving delivery data from outbox tables.
@@ -29,47 +29,47 @@ const MAX_BACKOFF: Duration = Duration::from_secs(30);
 /// How long to wait before polling again when there is no work.
 const POLL_INTERVAL: Duration = Duration::from_millis(500);
 
-/// A sender that notifies the IDC runtime of a new outbox row.
+/// A sender that notifies the IDC actor of a new outbox row.
 ///
-/// Sending `()` wakes the runtime to deliver pending messages immediately
+/// Sending `()` wakes the actor to deliver pending messages immediately
 /// rather than waiting for the next poll cycle.
-pub type IdcSender = mpsc::UnboundedSender<()>;
+pub type IdcActorSender = mpsc::UnboundedSender<()>;
 
-/// The identity of this (sender) database, set when the IDC runtime is started.
-pub struct IdcRuntimeConfig {
+/// The identity of this (sender) database, set when the IDC actor is started.
+pub struct IdcActorConfig {
     pub sender_identity: Identity,
 }
 
-/// A handle that, when dropped, stops the IDC runtime background task.
-pub struct IdcRuntime {
+/// A handle that, when dropped, stops the IDC actor background task.
+pub struct IdcActor {
     _abort: tokio::task::AbortHandle,
 }
 
-/// Holds the receiver side of the notification channel until the runtime is started.
+/// Holds the receiver side of the notification channel until the actor is started.
 ///
 /// Mirrors the `SchedulerStarter` pattern: create the channel before the module is
-/// loaded (so the sender can be stored in `InstanceEnv`), then call [`IdcRuntimeStarter::start`]
+/// loaded (so the sender can be stored in `InstanceEnv`), then call [`IdcActorStarter::start`]
 /// once the DB is ready.
-pub struct IdcRuntimeStarter {
+pub struct IdcActorStarter {
     rx: mpsc::UnboundedReceiver<()>,
 }
 
-impl IdcRuntimeStarter {
-    /// Spawn the IDC runtime background task.
-    pub fn start(self, db: Arc<RelationalDB>, config: IdcRuntimeConfig, module_host: WeakModuleHost) -> IdcRuntime {
+impl IdcActorStarter {
+    /// Spawn the IDC actor background task.
+    pub fn start(self, db: Arc<RelationalDB>, config: IdcActorConfig, module_host: WeakModuleHost) -> IdcActor {
         let abort = tokio::spawn(run_idc_loop(db, config, module_host, self.rx)).abort_handle();
-        IdcRuntime { _abort: abort }
+        IdcActor { _abort: abort }
     }
 }
 
-impl IdcRuntime {
+impl IdcActor {
     /// Open the IDC channel, returning a starter and a sender.
     ///
     /// Store the sender in `ModuleCreationContext` so it reaches `InstanceEnv`.
-    /// After the module is ready, call [`IdcRuntimeStarter::start`] to spawn the loop.
-    pub fn open() -> (IdcRuntimeStarter, IdcSender) {
+    /// After the module is ready, call [`IdcActorStarter::start`] to spawn the loop.
+    pub fn open() -> (IdcActorStarter, IdcActorSender) {
         let (tx, rx) = mpsc::unbounded_channel();
-        (IdcRuntimeStarter { rx }, tx)
+        (IdcActorStarter { rx }, tx)
     }
 }
 
@@ -138,10 +138,10 @@ enum DeliveryOutcome {
     TransportError(String),
 }
 
-/// Main IDC loop: maintain per-target queues and deliver messages.
+/// Main IDC actor loop: maintain per-target queues and deliver messages.
 async fn run_idc_loop(
     db: Arc<RelationalDB>,
-    config: IdcRuntimeConfig,
+    config: IdcActorConfig,
     module_host: WeakModuleHost,
     mut notify_rx: mpsc::UnboundedReceiver<()>,
 ) {
@@ -169,7 +169,7 @@ async fn run_idc_loop(
                 match outcome {
                     DeliveryOutcome::TransportError(reason) => {
                         log::warn!(
-                            "idc_runtime: transport error delivering msg_id={} to {}: {reason}",
+                            "idc_actor: transport error delivering msg_id={} to {}: {reason}",
                             msg.msg_id,
                             hex::encode(msg.target_db_identity.to_byte_array()),
                         );
@@ -224,6 +224,9 @@ fn outcome_to_result(outcome: &DeliveryOutcome) -> (u8, String) {
 }
 
 /// Finalize a delivered message: call the on_result reducer (if any), then delete from ST_OUTBOUND_MSG.
+///
+/// On the happy path, `on_result_reducer` success and deletion of `st_outbound_msg`
+/// are committed atomically in the same reducer transaction.
 async fn finalize_message(
     db: &RelationalDB,
     module_host: &WeakModuleHost,
@@ -235,7 +238,7 @@ async fn finalize_message(
     if let Some(on_result_reducer) = &msg.on_result_reducer {
         let Some(host) = module_host.upgrade() else {
             log::warn!(
-                "idc_runtime: module host gone, cannot call on_result reducer '{}' for msg_id={}",
+                "idc_actor: module host gone, cannot call on_result reducer '{}' for msg_id={}",
                 on_result_reducer,
                 msg.msg_id,
             );
@@ -249,7 +252,7 @@ async fn finalize_message(
             Ok(b) => b,
             Err(e) => {
                 log::error!(
-                    "idc_runtime: failed to encode on_result args for msg_id={}: {e}",
+                    "idc_actor: failed to encode on_result args for msg_id={}: {e}",
                     msg.msg_id
                 );
                 delete_message(db, msg.msg_id);
@@ -259,28 +262,30 @@ async fn finalize_message(
 
         let caller_identity = Identity::ZERO; // system call
         let result = host
-            .call_reducer(
+            .call_reducer_delete_outbound_on_success(
                 caller_identity,
                 None, // no connection_id
                 None, // no client sender
                 None, // no request_id
                 None, // no timer
                 on_result_reducer,
                 FunctionArgs::Bsatn(bytes::Bytes::from(args_bytes)),
+                msg.msg_id,
             )
             .await;
 
         match result {
             Ok(_) => {
                 log::debug!(
-                    "idc_runtime: on_result reducer '{}' called for msg_id={}",
+                    "idc_actor: on_result reducer '{}' called for msg_id={}",
                     on_result_reducer,
                     msg.msg_id,
                 );
+                return;
             }
             Err(e) => {
                 log::error!(
-                    "idc_runtime: on_result reducer '{}' failed for msg_id={}: {e:?}",
+                    "idc_actor: on_result reducer '{}' failed for msg_id={}: {e:?}",
                     on_result_reducer,
                     msg.msg_id,
                 );
@@ -307,7 +312,7 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
                 .collect()
         })
         .unwrap_or_else(|e| {
-            log::error!("idc_runtime: failed to read st_outbound_msg: {e}");
+            log::error!("idc_actor: failed to read st_outbound_msg: {e}");
             Vec::new()
         });
 
@@ -321,7 +326,7 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
             Ok(s) => s,
             Err(e) => {
                 log::error!(
-                    "idc_runtime: cannot find schema for outbox table {:?} (msg_id={}): {e}",
+                    "idc_actor: cannot find schema for outbox table {:?} (msg_id={}): {e}",
                     outbox_table_id,
                     st_row.msg_id,
                 );
@@ -333,7 +338,7 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
             Some(o) => o,
             None => {
                 log::error!(
-                    "idc_runtime: table {:?} (msg_id={}) is not an outbox table",
+                    "idc_actor: table {:?} (msg_id={}) is not an outbox table",
                     schema.table_name,
                     st_row.msg_id,
                 );
@@ -351,7 +356,7 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
 
         let Some(outbox_row_ref) = outbox_row else {
             log::error!(
-                "idc_runtime: outbox row not found in table {:?} for row_id={} (msg_id={})",
+                "idc_actor: outbox row not found in table {:?} for row_id={} (msg_id={})",
                 outbox_table_id,
                 st_row.row_id,
                 st_row.msg_id,
@@ -366,7 +371,7 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
             Some(AlgebraicValue::U256(u)) => Identity::from_u256(**u),
             other => {
                 log::error!(
-                    "idc_runtime: outbox row col 1 expected U256 (Identity), got {other:?} (msg_id={})",
+                    "idc_actor: outbox row col 1 expected U256 (Identity), got {other:?} (msg_id={})",
                     st_row.msg_id,
                 );
                 continue;
@@ -409,7 +414,7 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
 /// Attempt a single HTTP delivery of a message.
 async fn attempt_delivery(
     client: &reqwest::Client,
-    config: &IdcRuntimeConfig,
+    config: &IdcActorConfig,
     msg: &PendingMessage,
 ) -> DeliveryOutcome {
     let target_db_hex = hex::encode(msg.target_db_identity.to_byte_array());
@@ -453,11 +458,11 @@ async fn attempt_delivery(
 fn delete_message(db: &RelationalDB, msg_id: u64) {
     let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal);
     if let Err(e) = tx.delete_outbound_msg(msg_id) {
-        log::error!("idc_runtime: failed to delete msg_id={msg_id}: {e}");
+        log::error!("idc_actor: failed to delete msg_id={msg_id}: {e}");
         let _ = db.rollback_mut_tx(tx);
         return;
     }
     if let Err(e) = db.commit_tx(tx) {
-        log::error!("idc_runtime: failed to commit delete for msg_id={msg_id}: {e}");
+        log::error!("idc_actor: failed to commit delete for msg_id={msg_id}: {e}");
     }
 }