call_from_database endpoint

Author: Shubham8287

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

@@ -216,6 +216,117 @@ pub async fn call<S: ControlStateDelegate + NodeDelegate>(
     }
 }
 
+/// Path parameters for the `call_from_database` route.
+#[derive(Deserialize)]
+pub struct CallFromDatabaseParams {
+    name_or_identity: NameOrIdentity,
+    reducer: String,
+}
+
+/// Query parameters for the `call_from_database` route.
+///
+/// Both fields are mandatory; a missing field results in a 400 Bad Request.
+#[derive(Deserialize)]
+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,
+}
+
+/// Call a reducer on behalf of another database, with deduplication.
+///
+/// Endpoint: `POST /database/:name_or_identity/call-from-database/:reducer`
+///
+/// Required query params:
+/// - `sender_identity` — hex-encoded identity of the sending database.
+/// - `tx_offset`       — the sender's commitlog offset for this message.
+///
+/// Before invoking the reducer, the receiver checks `st_databases_tx_offset`.
+/// If the incoming `tx_offset` is ≤ the last delivered offset 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.
+pub async fn call_from_database<S: ControlStateDelegate + NodeDelegate>(
+    State(worker_ctx): State<S>,
+    Extension(auth): Extension<SpacetimeAuth>,
+    Path(CallFromDatabaseParams {
+        name_or_identity,
+        reducer,
+    }): Path<CallFromDatabaseParams>,
+    Query(CallFromDatabaseQuery {
+        sender_identity,
+        tx_offset,
+    }): Query<CallFromDatabaseQuery>,
+    TypedHeader(content_type): TypedHeader<headers::ContentType>,
+    ByteStringBody(body): ByteStringBody,
+) -> axum::response::Result<impl IntoResponse> {
+    assert_content_type_json(content_type)?;
+
+    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 connection_id = generate_random_connection_id();
+
+    let (module, Database { owner_identity, .. }) = find_module_and_database(&worker_ctx, name_or_identity).await?;
+
+    // Call client_connected, if defined.
+    module
+        .call_identity_connected(auth.into(), connection_id)
+        .await
+        .map_err(client_connected_error_to_response)?;
+
+    let result = module
+        .call_reducer_from_database(
+            caller_identity,
+            Some(connection_id),
+            None,
+            None,
+            None,
+            &reducer,
+            args,
+            sender_identity,
+            tx_offset,
+        )
+        .await;
+
+    module
+        .call_identity_disconnected(caller_identity, connection_id)
+        .await
+        .map_err(client_disconnected_error_to_response)?;
+
+    match result {
+        Ok(rcr) => {
+            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),
+                ReducerOutcome::BudgetExceeded => {
+                    log::warn!(
+                        "Node's energy budget exceeded for identity: {owner_identity} while executing {reducer}"
+                    );
+                    (StatusCode::PAYMENT_REQUIRED, "Module energy budget exhausted.".into())
+                }
+            };
+            Ok((
+                status,
+                TypedHeader(SpacetimeEnergyUsed(rcr.energy_used)),
+                TypedHeader(SpacetimeExecutionDurationMicros(rcr.execution_duration)),
+                body,
+            )
+                .into_response())
+        }
+        Err(e) => {
+            let (status, msg) = map_reducer_error(e, &reducer);
+            Err((status, msg).into())
+        }
+    }
+}
+
 fn assert_content_type_json(content_type: headers::ContentType) -> axum::response::Result<()> {
     if content_type != headers::ContentType::json() {
         Err(axum::extract::rejection::MissingJsonContentType::default().into())
@@ -230,7 +341,7 @@ fn reducer_outcome_response(
     outcome: ReducerOutcome,
 ) -> (StatusCode, Box<str>) {
     match outcome {
-        ReducerOutcome::Committed => (StatusCode::OK, "".into()),
+        ReducerOutcome::Committed | ReducerOutcome::Deduplicated => (StatusCode::OK, "".into()),
         ReducerOutcome::Failed(errmsg) => {
             // TODO: different status code? this is what cloudflare uses, sorta
             (StatusCode::from_u16(530).unwrap(), *errmsg)
@@ -1189,6 +1300,8 @@ 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>
+    pub call_from_database_post: MethodRouter<S>,
     /// GET: /database/:name_or_identity/schema
     pub schema_get: MethodRouter<S>,
     /// GET: /database/:name_or_identity/logs
@@ -1220,6 +1333,7 @@ where
             identity_get: get(get_identity::<S>),
             subscribe_get: get(handle_websocket::<S>),
             call_reducer_procedure_post: post(call::<S>),
+            call_from_database_post: post(call_from_database::<S>),
             schema_get: get(schema::<S>),
             logs_get: get(logs::<S>),
             sql_post: post(sql::<S>),
@@ -1245,6 +1359,7 @@ where
             .route("/identity", self.identity_get)
             .route("/subscribe", self.subscribe_get)
             .route("/call/:reducer", self.call_reducer_procedure_post)
+            .route("/call-from-database/:reducer", self.call_from_database_post)
             .route("/schema", self.schema_get)
             .route("/logs", self.logs_get)
             .route("/sql", self.sql_post)

crates/core/src/host/host_controller.rs

@@ -160,19 +160,22 @@ pub enum ReducerOutcome {
     Committed,
     Failed(Box<Box<str>>),
     BudgetExceeded,
+    /// The call was identified as a duplicate via the `st_databases_tx_offset` dedup index
+    /// and was discarded without running the reducer.
+    Deduplicated,
 }
 
 impl ReducerOutcome {
     pub fn into_result(self) -> anyhow::Result<()> {
         match self {
-            Self::Committed => Ok(()),
+            Self::Committed | Self::Deduplicated => Ok(()),
             Self::Failed(e) => Err(anyhow::anyhow!(e)),
             Self::BudgetExceeded => Err(anyhow::anyhow!("reducer ran out of energy")),
         }
     }
 
     pub fn is_err(&self) -> bool {
-        !matches!(self, Self::Committed)
+        !matches!(self, Self::Committed | Self::Deduplicated)
     }
 }
 

crates/core/src/host/module_host.rs

@@ -589,7 +589,7 @@ pub fn call_identity_connected(
             // then insert into `st_client`,
             // but if we crashed in between, we'd be left in an inconsistent state
             // where the reducer had run but `st_client` was not yet updated.
-            ReducerOutcome::Committed => Ok(()),
+            ReducerOutcome::Committed | ReducerOutcome::Deduplicated => Ok(()),
 
             // If the reducer returned an error or couldn't run due to insufficient energy,
             // abort the connection: the module code has decided it doesn't want this client.
@@ -624,6 +624,15 @@ pub struct CallReducerParams {
     pub timer: Option<Instant>,
     pub reducer_id: ReducerId,
     pub args: ArgsTuple,
+    /// If set, enables at-most-once delivery semantics for database-to-database calls.
+    ///
+    /// The tuple is `(sender_database_identity, sender_tx_offset)`.
+    /// Before running the reducer, `st_databases_tx_offset` is consulted:
+    /// if `sender_tx_offset` ≤ the stored last-delivered offset for the sender,
+    /// the call is a duplicate and returns [`ReducerCallError::Deduplicated`].
+    /// Otherwise the reducer runs, and the stored offset is updated atomically
+    /// within the same transaction.
+    pub dedup_sender: Option<(Identity, u64)>,
 }
 
 impl CallReducerParams {
@@ -644,6 +653,7 @@ impl CallReducerParams {
             timer: None,
             reducer_id,
             args,
+            dedup_sender: None,
         }
     }
 }
@@ -1490,9 +1500,9 @@ impl ModuleHost {
                     ..
                 }) => fallback(),
 
-                // If it succeeded, as mentioned above, `st_client` is already updated.
+                // If it succeeded (or was deduplicated), as mentioned above, `st_client` is already updated.
                 Ok(ReducerCallResult {
-                    outcome: ReducerOutcome::Committed,
+                    outcome: ReducerOutcome::Committed | ReducerOutcome::Deduplicated,
                     ..
                 }) => Ok(()),
             }
@@ -1567,6 +1577,7 @@ impl ModuleHost {
             timer,
             reducer_id,
             args,
+            dedup_sender: None,
         })
     }
 
@@ -1594,6 +1605,7 @@ impl ModuleHost {
             timer,
             reducer_id,
             args,
+            dedup_sender: None,
         };
 
         self.call(
@@ -1655,6 +1667,80 @@ impl ModuleHost {
         res
     }
 
+    /// Variant of [`Self::call_reducer`] for database-to-database calls.
+    ///
+    /// Behaves identically to `call_reducer`, except that it enforces at-most-once
+    /// delivery using the `st_databases_tx_offset` dedup index.
+    /// Before invoking the reducer, the receiver checks whether
+    /// `sender_tx_offset` ≤ the last delivered offset for `sender_database_identity`.
+    /// If so, the call is a duplicate and [`ReducerOutcome::Deduplicated`] is returned
+    /// without running the reducer.
+    /// Otherwise the reducer runs, and the dedup index is updated atomically
+    /// within the same transaction.
+    pub async fn call_reducer_from_database(
+        &self,
+        caller_identity: Identity,
+        caller_connection_id: Option<ConnectionId>,
+        client: Option<Arc<ClientConnectionSender>>,
+        request_id: Option<RequestId>,
+        timer: Option<Instant>,
+        reducer_name: &str,
+        args: FunctionArgs,
+        sender_database_identity: Identity,
+        sender_tx_offset: u64,
+    ) -> Result<ReducerCallResult, ReducerCallError> {
+        let res = async {
+            let (reducer_id, reducer_def) = self
+                .info
+                .module_def
+                .reducer_full(reducer_name)
+                .ok_or(ReducerCallError::NoSuchReducer)?;
+            if let Some(lifecycle) = reducer_def.lifecycle {
+                return Err(ReducerCallError::LifecycleReducer(lifecycle));
+            }
+
+            if reducer_def.visibility.is_private() && !self.is_database_owner(caller_identity) {
+                return Err(ReducerCallError::NoSuchReducer);
+            }
+
+            let args = args
+                .into_tuple_for_def(&self.info.module_def, reducer_def)
+                .map_err(InvalidReducerArguments)?;
+            let caller_connection_id = caller_connection_id.unwrap_or(ConnectionId::ZERO);
+            let call_reducer_params = CallReducerParams {
+                timestamp: Timestamp::now(),
+                caller_identity,
+                caller_connection_id,
+                client,
+                request_id,
+                timer,
+                reducer_id,
+                args,
+                dedup_sender: Some((sender_database_identity, sender_tx_offset)),
+            };
+
+            self.call(
+                &reducer_def.name,
+                call_reducer_params,
+                async |p, inst| Ok(inst.call_reducer(p)),
+                async |p, inst| inst.call_reducer(p).await,
+            )
+            .await?
+        }
+        .await;
+
+        let log_message = match &res {
+            Err(ReducerCallError::NoSuchReducer) => Some(no_such_function_log_message("reducer", reducer_name)),
+            Err(ReducerCallError::Args(_)) => Some(args_error_log_message("reducer", reducer_name)),
+            _ => None,
+        };
+        if let Some(log_message) = log_message {
+            self.inject_logs(LogLevel::Error, reducer_name, &log_message)
+        }
+
+        res
+    }
+
     pub async fn call_view_add_single_subscription(
         &self,
         sender: Arc<ClientConnectionSender>,

crates/core/src/host/v8/mod.rs

@@ -613,7 +613,7 @@ enum JsWorkerRequest {
     },
 }
 
-static_assert_size!(CallReducerParams, 192);
+static_assert_size!(CallReducerParams, 256);
 
 fn send_worker_reply<T>(ctx: &str, reply_tx: JsReplyTx<T>, value: T, trapped: bool) {
     if reply_tx.send(JsWorkerReply { value, trapped }).is_err() {

crates/core/src/host/wasm_common/module_host_actor.rs

@@ -2,7 +2,7 @@ use super::instrumentation::CallTimes;
 use super::*;
 use crate::client::ClientActorId;
 use crate::database_logger;
-use crate::energy::{EnergyMonitor, FunctionBudget, FunctionFingerprint};
+use crate::energy::{EnergyMonitor, EnergyQuanta, FunctionBudget, FunctionFingerprint};
 use crate::error::DBError;
 use crate::host::host_controller::CallProcedureReturn;
 use crate::host::instance_env::{InstanceEnv, TxSlot};
@@ -820,6 +820,7 @@ impl InstanceCommon {
             reducer_id,
             args,
             timer,
+            dedup_sender,
         } = params;
         let caller_connection_id_opt = (caller_connection_id != ConnectionId::ZERO).then_some(caller_connection_id);
 
@@ -844,6 +845,25 @@ impl InstanceCommon {
 
         let workload = Workload::Reducer(ReducerContext::from(op.clone()));
         let tx = tx.unwrap_or_else(|| stdb.begin_mut_tx(IsolationLevel::Serializable, workload));
+
+        // Check the dedup index atomically within this transaction.
+        // If the incoming offset is ≤ the last delivered offset for this sender,
+        // the message is a duplicate; discard it and roll back.
+        if let Some((sender_identity, sender_tx_offset)) = dedup_sender {
+            let last_delivered = tx.get_databases_tx_offset(sender_identity);
+            if last_delivered.is_some_and(|last| sender_tx_offset <= last) {
+                let _ = stdb.rollback_mut_tx(tx);
+                return (
+                    ReducerCallResult {
+                        outcome: ReducerOutcome::Deduplicated,
+                        energy_used: EnergyQuanta::ZERO,
+                        execution_duration: Duration::ZERO,
+                    },
+                    false,
+                );
+            }
+        }
+
         let mut tx_slot = inst.tx_slot();
 
         let vm_metrics = self.vm_metrics.get_for_reducer_id(reducer_id);
@@ -885,12 +905,21 @@ impl InstanceCommon {
             Ok(return_value) => {
                 // If this is an OnDisconnect lifecycle event, remove the client from st_clients.
                 // We handle OnConnect events before running the reducer.
-                let res = match reducer_def.lifecycle {
+                let lifecycle_res = match reducer_def.lifecycle {
                     Some(Lifecycle::OnDisconnect) => {
                         tx.delete_st_client(caller_identity, caller_connection_id, info.database_identity)
                     }
                     _ => Ok(()),
                 };
+                // Atomically update the dedup index so this sender's tx_offset is recorded
+                // as delivered. This prevents re-processing if the message is retried.
+                let dedup_res = match dedup_sender {
+                    Some((sender_identity, sender_tx_offset)) => {
+                        tx.upsert_databases_tx_offset(sender_identity, sender_tx_offset)
+                    }
+                    None => Ok(()),
+                };
+                let res = lifecycle_res.and(dedup_res);
                 match res {
                     Ok(()) => (EventStatus::Committed(DatabaseUpdate::default()), return_value),
                     Err(err) => {