Merge remote-tracking branch 'origin/jdetter/tpcc' into shub/2pc-regular

Author: Shubham8287

Cargo.lock

@@ -52,6 +52,7 @@ members = [
   "modules/sdk-test-view-pk",
   "modules/sdk-test-event-table",
   "modules/tpcc",
+  "modules/tpcc-metrics",
   "sdks/rust/tests/test-client",
   "sdks/rust/tests/test-counter",
   "sdks/rust/tests/connect_disconnect_client",

Cargo.toml

@@ -52,16 +52,22 @@ impl core::fmt::Display for RemoteCallError {
 /// - `reducer_name`: the name of the reducer to invoke (must be valid UTF-8).
 /// - `args`: BSATN-encoded reducer arguments.
 ///
-/// Returns `Ok(())` when the remote reducer ran and succeeded.
+/// Returns `Ok(bytes)` when the remote reducer ran and succeeded, with `bytes` being the reducer's output.
 /// Returns `Err(RemoteCallError::Failed(msg))` when the reducer ran but returned an error.
 /// Returns `Err(RemoteCallError::NotFound(msg))` when the database or reducer does not exist.
 /// Returns `Err(RemoteCallError::Unreachable(msg))` on transport failure (connection refused, timeout, …).
-pub fn call_reducer_on_db(database_identity: Identity, reducer_name: &str, args: &[u8]) -> Result<(), RemoteCallError> {
+pub fn call_reducer_on_db(
+    database_identity: Identity,
+    reducer_name: &str,
+    args: &[u8],
+) -> Result<Vec<u8>, RemoteCallError> {
     let identity_bytes = database_identity.to_byte_array();
     match spacetimedb_bindings_sys::call_reducer_on_db(identity_bytes, reducer_name, args) {
         Ok((status, body_source)) => {
             if status < 300 {
-                return Ok(());
+                let mut out = Vec::new();
+                read_bytes_source_into(body_source, &mut out);
+                return Ok(out);
             }
             // Decode the response body as the error message.
             let msg = if body_source == spacetimedb_bindings_sys::raw::BytesSource::INVALID {

crates/bindings/src/remote_reducer.rs

@@ -142,7 +142,7 @@ pub async fn call<S: ControlStateDelegate + NodeDelegate>(
 ) -> axum::response::Result<impl IntoResponse> {
     let caller_identity = auth.claims.identity;
 
-    let args = parse_call_args(content_type, body)?;
+    let (args, want_bsatn) = parse_call_args(content_type, body)?;
 
     // HTTP callers always need a connection ID to provide to connect/disconnect,
     // so generate one.
@@ -195,8 +195,14 @@ pub async fn call<S: ControlStateDelegate + NodeDelegate>(
 
     match result {
         Ok(CallResult::Reducer(result)) => {
-            let (status, body) =
-                reducer_outcome_response(&module, &owner_identity, &reducer, result.outcome, reducer_return_value)?;
+            let (status, body) = reducer_outcome_response(
+                &module,
+                &owner_identity,
+                &reducer,
+                result.outcome,
+                reducer_return_value,
+                want_bsatn,
+            )?;
             Ok((
                 status,
                 TypedHeader(SpacetimeEnergyUsed(result.energy_used)),
@@ -209,7 +215,7 @@ pub async fn call<S: ControlStateDelegate + NodeDelegate>(
             // Procedures don't assign a special meaning to error returns, unlike reducers,
             // as there's no transaction for them to automatically abort.
             // Instead, we just pass on their return value with the OK status so long as we successfully invoked the procedure.
-            let (status, body) = procedure_outcome_response(result.return_val);
+            let (status, body) = procedure_outcome_response(result.return_val, want_bsatn);
             Ok((
                 status,
                 TypedHeader(SpacetimeExecutionDurationMicros(result.execution_duration)),
@@ -225,13 +231,15 @@ pub async fn call<S: ControlStateDelegate + NodeDelegate>(
 ///
 /// - `application/json` → [`FunctionArgs::Json`] (UTF-8 required).
 /// - `application/octet-stream` → [`FunctionArgs::Bsatn`] (raw BSATN bytes).
-fn parse_call_args(content_type: headers::ContentType, body: Bytes) -> axum::response::Result<FunctionArgs> {
+///
+/// Also returns `want_bsatn`, a bool, which is true if and only if the response should be BSATN-encoded.
+fn parse_call_args(content_type: headers::ContentType, body: Bytes) -> axum::response::Result<(FunctionArgs, bool)> {
     if content_type == headers::ContentType::json() {
         let s = bytestring::ByteString::try_from(body)
             .map_err(|_| (StatusCode::BAD_REQUEST, "request body is not valid UTF-8").into_response())?;
-        Ok(FunctionArgs::Json(s))
+        Ok((FunctionArgs::Json(s), false))
     } else if content_type == headers::ContentType::octet_stream() {
-        Ok(FunctionArgs::Bsatn(body))
+        Ok((FunctionArgs::Bsatn(body), true))
     } else {
         Err((
             StatusCode::BAD_REQUEST,
@@ -276,13 +284,13 @@ pub async fn prepare<S: ControlStateDelegate + NodeDelegate>(
     // call_identity_connected/disconnected submit jobs to the module's executor, which
     // will be blocked holding the 2PC write lock after prepare_reducer returns — deadlock.
     let result = module
-        .prepare_reducer(caller_identity, None, &reducer, args, coordinator_identity)
+        .prepare_reducer(caller_identity, None, &reducer, args.0, coordinator_identity)
         .await;
 
     match result {
         Ok((prepare_id, rcr, return_value)) => {
             let (status, body) =
-                reducer_outcome_response(&module, &owner_identity, &reducer, rcr.outcome, return_value)?;
+                reducer_outcome_response(&module, &owner_identity, &reducer, rcr.outcome, return_value, args.1)?;
             let mut response = (
                 status,
                 TypedHeader(SpacetimeEnergyUsed(rcr.energy_used)),
@@ -398,12 +406,18 @@ pub async fn ack_commit_2pc<S: ControlStateDelegate + NodeDelegate>(
     Ok(StatusCode::OK)
 }
 
+/// Encode a reducer return value as an HTTP response.
+///
+/// If the outcome is an error, return a raw string with `application/text`. Ignore `want_bsatn` in this case.
+/// If the outcome is successful, and `want_bsatn`, send BSATN with `application/octet-stream`.
+/// If the outcome is successful, and not `want_bsatn`, send JSON with `application/json`.
 fn reducer_outcome_response(
     module: &ModuleHost,
     owner_identity: &Identity,
     reducer: &str,
     outcome: ReducerOutcome,
     reducer_return_value: Option<Bytes>,
+    want_bsatn: bool,
 ) -> axum::response::Result<(StatusCode, axum::response::Response)> {
     match outcome {
         ReducerOutcome::Committed => {
@@ -415,20 +429,29 @@ fn reducer_outcome_response(
             };
 
             if let Some(bytes) = reducer_return_value.filter(|value| !value.is_empty()) {
-                let seed = sats::WithTypespace::new(module.info.module_def.typespace(), &return_value);
-                let mut reader = &bytes[..];
-                let value: AlgebraicValue = seed.deserialize(bsatn::Deserializer::new(&mut reader)).map_err(|err| {
-                    (
-                        StatusCode::INTERNAL_SERVER_ERROR,
-                        format!("Failed to decode reducer return value: {err}"),
-                    )
-                })?;
-                Ok((
-                    StatusCode::OK,
-                    axum::Json(sats::serde::SerdeWrapper(value)).into_response(),
-                ))
+                if want_bsatn {
+                    Ok((StatusCode::OK, bytes.into_response()))
+                } else {
+                    let seed = sats::WithTypespace::new(module.info.module_def.typespace(), &return_value);
+                    let mut reader = &bytes[..];
+                    let value: AlgebraicValue =
+                        seed.deserialize(bsatn::Deserializer::new(&mut reader)).map_err(|err| {
+                            (
+                                StatusCode::INTERNAL_SERVER_ERROR,
+                                format!("Failed to decode reducer return value: {err}"),
+                            )
+                        })?;
+                    Ok((
+                        StatusCode::OK,
+                        axum::Json(sats::serde::SerdeWrapper(value)).into_response(),
+                    ))
+                }
             } else {
-                Ok((StatusCode::OK, "".into_response()))
+                if want_bsatn {
+                    Ok((StatusCode::OK, Vec::<u8>::new().into_response()))
+                } else {
+                    Ok((StatusCode::OK, "".into_response()))
+                }
             }
         }
         ReducerOutcome::Failed(errmsg) => {
@@ -511,10 +534,20 @@ pub enum DBCallErr {
     InstanceNotScheduled,
 }
 
-fn procedure_outcome_response(return_val: AlgebraicValue) -> (StatusCode, axum::response::Response) {
+/// Encode a procedure's result as an HTTP response.
+///
+/// If `want_bsatn`, send BSATN bytes as `application/octet-stream`.
+/// If not `want_bsatn`, send JSON as `application/json`.
+fn procedure_outcome_response(return_val: AlgebraicValue, want_bsatn: bool) -> (StatusCode, axum::response::Response) {
     (
         StatusCode::OK,
-        axum::Json(sats::serde::SerdeWrapper(return_val)).into_response(),
+        if want_bsatn {
+            bsatn::to_vec(&return_val)
+                .expect("BSATN serializing an AlgebraicValue should be infallible")
+                .into_response()
+        } else {
+            axum::Json(sats::serde::SerdeWrapper(return_val)).into_response()
+        },
     )
 }
 

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

@@ -56,11 +56,12 @@ impl MetadataFile {
         path.write(self.to_string())
     }
 
-    fn check_compatibility(previous: &Self, current: &Self) -> anyhow::Result<()> {
+    fn check_compatibility(previous: &Self, current: &Self, metafile: &Path) -> anyhow::Result<()> {
         anyhow::ensure!(
             previous.edition == current.edition,
-            "metadata.toml indicates that this database is from a different \
+            "metadata.toml at {} indicates that this database is from a different \
             edition of SpacetimeDB (running {:?}, but this database is {:?})",
+            metafile.display(),
             current.edition,
             previous.edition,
         );
@@ -110,8 +111,8 @@ impl MetadataFile {
     /// `self` is the metadata file read from a database, and current is
     /// the default metadata file that the active database version would
     /// right to a new database.
-    pub fn check_compatibility_and_update(mut self, current: Self) -> anyhow::Result<Self> {
-        Self::check_compatibility(&self, &current)?;
+    pub fn check_compatibility_and_update(mut self, current: Self, metafile: &Path) -> anyhow::Result<Self> {
+        Self::check_compatibility(&self, &current, metafile)?;
         // bump the version in the file only if it's being run in a newer database.
         self.version = std::cmp::max(self.version, current.version);
         Ok(self)
@@ -344,67 +345,67 @@ mod tests {
     fn check_metadata_compatibility_checking() {
         assert_eq!(
             mkmeta(1, 0, 0)
-                .check_compatibility_and_update(mkmeta(1, 0, 1))
+                .check_compatibility_and_update(mkmeta(1, 0, 1), Path::new("metadata.toml"))
                 .unwrap()
                 .version,
             mkver(1, 0, 1)
         );
         assert_eq!(
             mkmeta(1, 0, 1)
-                .check_compatibility_and_update(mkmeta(1, 0, 0))
+                .check_compatibility_and_update(mkmeta(1, 0, 0), Path::new("metadata.toml"))
                 .unwrap()
                 .version,
             mkver(1, 0, 1)
         );
 
         mkmeta(1, 1, 0)
-            .check_compatibility_and_update(mkmeta(1, 0, 5))
+            .check_compatibility_and_update(mkmeta(1, 0, 5), Path::new("metadata.toml"))
             .unwrap_err();
         mkmeta(2, 0, 0)
-            .check_compatibility_and_update(mkmeta(1, 3, 5))
+            .check_compatibility_and_update(mkmeta(1, 3, 5), Path::new("metadata.toml"))
             .unwrap_err();
         assert_eq!(
             mkmeta(1, 12, 0)
-                .check_compatibility_and_update(mkmeta(2, 0, 0))
+                .check_compatibility_and_update(mkmeta(2, 0, 0), Path::new("metadata.toml"))
                 .unwrap()
                 .version,
             mkver(2, 0, 0)
         );
         mkmeta(2, 0, 0)
-            .check_compatibility_and_update(mkmeta(3, 0, 0))
+            .check_compatibility_and_update(mkmeta(3, 0, 0), Path::new("metadata.toml"))
             .unwrap_err();
     }
 
     #[test]
     fn check_metadata_compatibility_prerelease() {
         mkmeta(1, 9, 0)
-            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"))
+            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"), Path::new("metadata.toml"))
             .unwrap();
 
         mkmeta_pre(2, 0, 0, "rc1")
-            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"))
+            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"), Path::new("metadata.toml"))
             .unwrap();
 
         mkmeta_pre(2, 0, 0, "rc1")
-            .check_compatibility_and_update(mkmeta(2, 0, 1))
+            .check_compatibility_and_update(mkmeta(2, 0, 1), Path::new("metadata.toml"))
             .unwrap();
 
         mkmeta_pre(2, 0, 0, "rc1")
-            .check_compatibility_and_update(mkmeta(2, 0, 0))
+            .check_compatibility_and_update(mkmeta(2, 0, 0), Path::new("metadata.toml"))
             .unwrap();
 
         // Now check some failures..
 
         mkmeta_pre(2, 0, 0, "rc1")
-            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc2"))
+            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc2"), Path::new("metadata.toml"))
             .unwrap_err();
 
         mkmeta_pre(2, 0, 0, "rc2")
-            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"))
+            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"), Path::new("metadata.toml"))
             .unwrap_err();
 
         mkmeta(2, 0, 0)
-            .check_compatibility_and_update(mkmeta_pre(2, 1, 0, "rc1"))
+            .check_compatibility_and_update(mkmeta_pre(2, 1, 0, "rc1"), Path::new("metadata.toml"))
             .unwrap_err();
     }
 

crates/core/src/config.rs

@@ -17,7 +17,7 @@ use spacetimedb_datastore::locking_tx_datastore::datastore::TxMetrics;
 use spacetimedb_datastore::locking_tx_datastore::state_view::{
     IterByColEqMutTx, IterByColRangeMutTx, IterMutTx, StateView,
 };
-use spacetimedb_datastore::locking_tx_datastore::{MutTxId, TxId};
+use spacetimedb_datastore::locking_tx_datastore::{IndexScanPointOrRange, MutTxId, TxId};
 use spacetimedb_datastore::system_tables::{
     system_tables, StModuleRow, ST_CLIENT_ID, ST_CONNECTION_CREDENTIALS_ID, ST_VIEW_SUB_ID,
 };
@@ -55,10 +55,11 @@ use spacetimedb_snapshot::{ReconstructedSnapshot, SnapshotError, SnapshotReposit
 use spacetimedb_table::indexes::RowPointer;
 use spacetimedb_table::page_pool::PagePool;
 use spacetimedb_table::table::{RowRef, TableScanIter};
+use spacetimedb_table::table_index::IndexKey;
 use std::borrow::Cow;
 use std::io;
 use std::num::NonZeroUsize;
-use std::ops::{Bound, RangeBounds};
+use std::ops::RangeBounds;
 use std::sync::Arc;
 use tokio::sync::watch;
 
@@ -1424,32 +1425,24 @@ impl RelationalDB {
         Ok(self.inner.iter_by_col_range_tx(tx, table_id.into(), cols, range)?)
     }
 
-    pub fn index_scan_range<'a>(
+    pub fn index_scan_range<'de, 'a>(
         &'a self,
         tx: &'a MutTx,
         index_id: IndexId,
-        prefix: &[u8],
+        prefix: &'de [u8],
         prefix_elems: ColId,
-        rstart: &[u8],
-        rend: &[u8],
-    ) -> Result<
-        (
-            TableId,
-            Bound<AlgebraicValue>,
-            Bound<AlgebraicValue>,
-            impl Iterator<Item = RowRef<'a>> + use<'a>,
-        ),
-        DBError,
-    > {
+        rstart: &'de [u8],
+        rend: &'de [u8],
+    ) -> Result<(TableId, IndexScanPointOrRange<'de, 'a>), DBError> {
         Ok(tx.index_scan_range(index_id, prefix, prefix_elems, rstart, rend)?)
     }
 
-    pub fn index_scan_point<'a>(
+    pub fn index_scan_point<'a, 'p>(
         &'a self,
         tx: &'a MutTx,
         index_id: IndexId,
-        point: &[u8],
-    ) -> Result<(TableId, AlgebraicValue, impl Iterator<Item = RowRef<'a>> + use<'a>), DBError> {
+        point: &'p [u8],
+    ) -> Result<(TableId, IndexKey<'p>, impl Iterator<Item = RowRef<'a>> + use<'a>), DBError> {
         Ok(tx.index_scan_point(index_id, point)?)
     }