idc runtime

Author: Shubham8287

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

@@ -230,9 +230,9 @@ pub struct CallFromDatabaseParams {
 pub struct CallFromDatabaseQuery {
     /// Hex-encoded [`Identity`] of the sending database.
     sender_identity: String,
-    /// The commitlog offset of the message on the sender side.
-    /// Used for at-most-once delivery via `st_databases_tx_offset`.
-    tx_offset: u64,
+    /// The inter-database message ID from the sender's st_msg_id.
+    /// Used for at-most-once delivery via `st_inbound_msg_id`.
+    msg_id: u64,
 }
 
 /// Call a reducer on behalf of another database, with deduplication.
@@ -241,10 +241,10 @@ pub struct CallFromDatabaseQuery {
 ///
 /// Required query params:
 /// - `sender_identity` — hex-encoded identity of the sending database.
-/// - `tx_offset`       — the sender's commitlog offset for this message.
+/// - `msg_id`          — the inter-database message ID from the sender's st_msg_id.
 ///
-/// Before invoking the reducer, the receiver checks `st_databases_tx_offset`.
-/// If the incoming `tx_offset` is ≤ the last delivered offset for `sender_identity`,
+/// Before invoking the reducer, the receiver checks `st_inbound_msg_id`.
+/// If the incoming `msg_id` is ≤ the last delivered msg_id for `sender_identity`,
 /// the call is a duplicate and 200 OK is returned immediately without running the reducer.
 /// Otherwise the reducer is invoked, the dedup index is updated atomically in the same
 /// transaction, and an acknowledgment is returned on success.
@@ -257,19 +257,22 @@ pub async fn call_from_database<S: ControlStateDelegate + NodeDelegate>(
     }): Path<CallFromDatabaseParams>,
     Query(CallFromDatabaseQuery {
         sender_identity,
-        tx_offset,
+        msg_id,
     }): Query<CallFromDatabaseQuery>,
     TypedHeader(content_type): TypedHeader<headers::ContentType>,
-    ByteStringBody(body): ByteStringBody,
+    body: axum::body::Bytes,
 ) -> axum::response::Result<impl IntoResponse> {
-    assert_content_type_json(content_type)?;
+    // IDC callers send BSATN (application/octet-stream).
+    if content_type != headers::ContentType::octet_stream() {
+        return Err((StatusCode::UNSUPPORTED_MEDIA_TYPE, "Expected application/octet-stream").into());
+    }
 
     let caller_identity = auth.claims.identity;
 
     let sender_identity = Identity::from_hex(&sender_identity)
         .map_err(|_| (StatusCode::BAD_REQUEST, "Invalid sender_identity: expected hex-encoded identity"))?;
 
-    let args = FunctionArgs::Json(body);
+    let args = FunctionArgs::Bsatn(body);
     let connection_id = generate_random_connection_id();
 
     let (module, Database { owner_identity, .. }) = find_module_and_database(&worker_ctx, name_or_identity).await?;
@@ -290,7 +293,7 @@ pub async fn call_from_database<S: ControlStateDelegate + NodeDelegate>(
             &reducer,
             args,
             sender_identity,
-            tx_offset,
+            msg_id,
         )
         .await;
 
@@ -304,7 +307,9 @@ 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()),
-                ReducerOutcome::Failed(errmsg) => (StatusCode::from_u16(530).unwrap(), *errmsg),
+                // 422 = reducer ran but returned Err; IDC runtime uses this to distinguish
+                // reducer failures from transport errors (which it retries).
+                ReducerOutcome::Failed(errmsg) => (StatusCode::UNPROCESSABLE_ENTITY, *errmsg),
                 ReducerOutcome::BudgetExceeded => {
                     log::warn!(
                         "Node's energy budget exceeded for identity: {owner_identity} while executing {reducer}"
@@ -1300,7 +1305,7 @@ pub struct DatabaseRoutes<S> {
     pub subscribe_get: MethodRouter<S>,
     /// POST: /database/:name_or_identity/call/:reducer
     pub call_reducer_procedure_post: MethodRouter<S>,
-    /// POST: /database/:name_or_identity/call-from-database/:reducer?sender_identity=<hex>&tx_offset=<u64>
+    /// POST: /database/:name_or_identity/call-from-database/:reducer?sender_identity=<hex>&msg_id=<u64>
     pub call_from_database_post: MethodRouter<S>,
     /// GET: /database/:name_or_identity/schema
     pub schema_get: MethodRouter<S>,

crates/core/src/host/host_controller.rs

@@ -1,3 +1,4 @@
+use super::idc_runtime::{IdcRuntime, IdcRuntimeConfig, IdcRuntimeStarter, IdcSender};
 use super::module_host::{EventStatus, ModuleHost, ModuleInfo, NoSuchModule};
 use super::scheduler::SchedulerStarter;
 use super::wasmtime::WasmtimeRuntime;
@@ -709,6 +710,7 @@ async fn make_module_host(
     runtimes: Arc<HostRuntimes>,
     replica_ctx: Arc<ReplicaContext>,
     scheduler: Scheduler,
+    idc_sender: IdcSender,
     program: Program,
     energy_monitor: Arc<dyn EnergyMonitor>,
     unregister: impl Fn() + Send + Sync + 'static,
@@ -724,6 +726,7 @@ async fn make_module_host(
     let mcc = ModuleCreationContext {
         replica_ctx,
         scheduler,
+        idc_sender,
         program_hash: program.hash,
         energy_monitor,
     };
@@ -761,6 +764,7 @@ struct LaunchedModule {
     module_host: ModuleHost,
     scheduler: Scheduler,
     scheduler_starter: SchedulerStarter,
+    idc_starter: IdcRuntimeStarter,
 }
 
 struct ModuleLauncher<F> {
@@ -797,10 +801,12 @@ 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 (program, module_host) = make_module_host(
             self.runtimes.clone(),
             replica_ctx.clone(),
             scheduler.clone(),
+            idc_sender.clone(),
             self.program,
             self.energy_monitor,
             self.on_panic,
@@ -817,6 +823,7 @@ impl<F: Fn() + Send + Sync + 'static> ModuleLauncher<F> {
                 module_host,
                 scheduler,
                 scheduler_starter,
+                idc_starter,
             },
         ))
     }
@@ -882,6 +889,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_msg_id`.
+    /// Stopped when [`Host`] is dropped.
+    _idc_runtime: IdcRuntime,
 }
 
 impl Host {
@@ -1075,6 +1085,7 @@ impl Host {
             module_host,
             scheduler,
             scheduler_starter,
+            idc_starter,
         } = launched;
 
         // Disconnect dangling clients.
@@ -1101,6 +1112,14 @@ 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(
+            replica_ctx.relational_db().clone(),
+            IdcRuntimeConfig {
+                sender_identity: replica_ctx.database_identity,
+            },
+            module_host.downgrade(),
+        );
+
         let module = watch::Sender::new(module_host);
 
         Ok(Host {
@@ -1110,6 +1129,7 @@ impl Host {
             disk_metrics_recorder_task,
             tx_metrics_recorder_task,
             view_cleanup_task,
+            _idc_runtime: idc_runtime,
         })
     }
 
@@ -1182,11 +1202,13 @@ 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 (program, module) = make_module_host(
             runtimes,
             replica_ctx.clone(),
             scheduler.clone(),
+            idc_sender,
             program,
             energy_monitor,
             on_panic,