Merge branch 'master' into bfops/remove-docker-compose

Author: bfops

crates/core/src/db/relational_db.rs

@@ -16,7 +16,9 @@ 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::{ApplyHistoryCounters, IndexScanPointOrRange, MutTxId, TxId};
+use spacetimedb_datastore::locking_tx_datastore::{
+    ApplyHistoryCounters, IndexScanPointOrRange, MutTxId, TxId, ViewCallInfo,
+};
 use spacetimedb_datastore::system_tables::{
     system_tables, StModuleRow, ST_CLIENT_ID, ST_CONNECTION_CREDENTIALS_ID, ST_VIEW_SUB_ID,
 };
@@ -1601,6 +1603,26 @@ impl RelationalDB {
         Ok(())
     }
 
+    /// Materialize a view call and replace its committed read set on transaction commit.
+    ///
+    /// The read set is only marked for replacement after materialization succeeds. If the
+    /// transaction rolls back, the replacement marker rolls back with it.
+    pub fn materialize_view_call(
+        &self,
+        tx: &mut MutTxId,
+        table_id: TableId,
+        view_call: ViewCallInfo,
+        rows: Vec<ProductValue>,
+    ) -> Result<(), DBError> {
+        match view_call.sender {
+            Some(sender) => self.materialize_view(tx, table_id, sender, rows)?,
+            None => self.materialize_anonymous_view(tx, table_id, rows)?,
+        }
+        tx.replace_view_read_set(view_call);
+
+        Ok(())
+    }
+
     fn write_view_rows(
         &self,
         tx: &mut MutTxId,

crates/core/src/host/module_host.rs

@@ -1089,6 +1089,66 @@ pub struct CallViewParams {
     pub timestamp: Timestamp,
 }
 
+pub(crate) struct ResolvedViewForRefresh<'a> {
+    pub view_id: ViewId,
+    pub table_id: TableId,
+    pub view_def: &'a ViewDef,
+}
+
+/// Lookup a module's [`ViewDef`] and check for consistency among
+/// its readset, `st_view`, and the [`ModuleDef`].
+pub(crate) fn resolve_view_for_refresh<'a>(
+    tx: &MutTxId,
+    module_def: &'a ModuleDef,
+    view_call: &ViewCallInfo,
+) -> anyhow::Result<ResolvedViewForRefresh<'a>> {
+    let st_view = tx
+        .lookup_st_view(view_call.view_id)
+        .with_context(|| format!("failed to look up view {}", view_call.view_id))?;
+
+    let view_id = st_view.view_id;
+    let table_id = st_view
+        .table_id
+        .ok_or_else(|| anyhow::anyhow!("view {:?} does not have a backing table", view_id))?;
+
+    let view_name: Identifier = st_view.view_name.into();
+    let view_def = module_def.view(&view_name).ok_or_else(|| {
+        anyhow::anyhow!(
+            "view `{}` for view id `{}` not found in current module",
+            view_name,
+            view_id
+        )
+    })?;
+
+    let is_anonymous = view_call.sender.is_none();
+
+    if st_view.is_anonymous != is_anonymous {
+        return Err(anyhow::anyhow!(
+            "found is_anonymous={} in st_view, but {} in readset when updating view `{}`",
+            st_view.is_anonymous,
+            is_anonymous,
+            view_name,
+        ));
+    }
+
+    let is_anonymous = view_def.is_anonymous;
+
+    if st_view.is_anonymous != is_anonymous {
+        return Err(anyhow::anyhow!(
+            "found is_anonymous={} in st_view, but {} in module when updating view `{}`",
+            st_view.is_anonymous,
+            is_anonymous,
+            view_name,
+        ));
+    }
+
+    Ok(ResolvedViewForRefresh {
+        view_id,
+        table_id,
+        view_def,
+    })
+}
+
 pub struct CallProcedureParams {
     pub timestamp: Timestamp,
     pub caller_identity: Identity,
@@ -2897,17 +2957,21 @@ impl ModuleHost {
         let mut total_duration = Duration::ZERO;
         let mut abi_duration = Duration::ZERO;
         let mut trapped = false;
-        for ViewCallInfo {
-            view_id,
-            table_id,
-            fn_ptr,
-            sender,
-        } in tx.views_for_refresh().cloned().collect::<Vec<_>>()
-        {
-            let Some(view_def) = module_def.get_view_by_id(fn_ptr, sender.is_none()) else {
-                outcome = ViewOutcome::Failed(format!("view with fn_ptr `{fn_ptr}` not found"));
-                break;
+        for view_call in tx.views_for_refresh().cloned().collect::<Vec<_>>() {
+            let sender = view_call.sender;
+            let resolved = match resolve_view_for_refresh(&tx, module_def, &view_call) {
+                Ok(resolved) => resolved,
+                Err(err) => {
+                    outcome = ViewOutcome::Failed(format!("failed to resolve view: {err}"));
+                    break;
+                }
             };
+            let ResolvedViewForRefresh {
+                view_id,
+                table_id,
+                view_def,
+            } = resolved;
+            let view_name = &view_def.name;
             let args = match FunctionArgs::Nullary.into_tuple_for_def(module_def, view_def) {
                 Ok(args) => args,
                 Err(err) => {
@@ -2919,7 +2983,7 @@ impl ModuleHost {
             let (result, trap) = Self::call_view_inner(
                 instance,
                 tx,
-                &view_def.name,
+                view_name,
                 view_id,
                 table_id,
                 view_def.fn_ptr,

crates/core/src/host/v8/syscall/common.rs

@@ -25,7 +25,7 @@ use anyhow::Context;
 use bytes::Bytes;
 use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, MutTxId, ViewCallInfo};
 use spacetimedb_lib::{ConnectionId, Identity, RawModuleDef, Timestamp};
-use spacetimedb_primitives::{ColId, IndexId, ProcedureId, TableId};
+use spacetimedb_primitives::{ColId, IndexId, ProcedureId, TableId, ViewFnPtr};
 use spacetimedb_sats::bsatn;
 use spacetimedb_schema::def::ModuleDef;
 use spacetimedb_schema::identifier::Identifier;
@@ -752,19 +752,22 @@ fn refresh_views(
 
     for view_call in views_for_refresh {
         let res: SysCallResult<()> = (|| {
-            let view_def = module_def
-                .get_view_by_id(view_call.fn_ptr, view_call.sender.is_none())
-                .ok_or_else(|| {
-                    TypeError(format!(
-                        "view with fn_ptr `{}` not found while refreshing procedure transaction",
-                        view_call.fn_ptr
-                    ))
-                    .throw(scope)
-                })?;
+            let resolved = crate::host::module_host::resolve_view_for_refresh(
+                tx.as_ref().expect("procedure tx missing during view refresh"),
+                module_def,
+                &view_call,
+            )
+            .map_err(|err| TypeError(format!("view refresh failed after procedure call: {err}")).throw(scope))?;
+
+            let table_id = resolved.table_id;
+            let view_def = resolved.view_def;
+            let view_name = &view_def.name;
+            let fn_ptr = view_def.fn_ptr;
 
             let current_tx = tx.take().expect("procedure tx missing during view refresh");
-            let (next_tx, call_result) =
-                tx_slot.set(current_tx, || call_view(scope, hooks, &view_call, &view_def.name));
+            let (next_tx, call_result) = tx_slot.set(current_tx, || {
+                call_view(scope, hooks, &view_call, view_name, table_id, fn_ptr)
+            });
             tx = Some(next_tx);
             let return_data = call_result?;
 
@@ -814,25 +817,14 @@ fn refresh_views(
                 })?,
             };
 
-            match view_call.sender {
-                Some(sender) => stdb
-                    .materialize_view(
-                        tx.as_mut()
-                            .expect("procedure tx missing while materializing authenticated view"),
-                        view_call.table_id,
-                        sender,
-                        rows,
-                    )
-                    .map_err(NodesError::from)?,
-                None => stdb
-                    .materialize_anonymous_view(
-                        tx.as_mut()
-                            .expect("procedure tx missing while materializing anonymous view"),
-                        view_call.table_id,
-                        rows,
-                    )
-                    .map_err(NodesError::from)?,
-            }
+            stdb.materialize_view_call(
+                tx.as_mut()
+                    .expect("procedure tx missing while materializing refreshed view"),
+                table_id,
+                view_call.clone(),
+                rows,
+            )
+            .map_err(NodesError::from)?;
 
             Ok(())
         })();
@@ -857,6 +849,8 @@ fn call_view(
     hooks: &HookFunctions<'_>,
     view_call: &ViewCallInfo,
     view_name: &Identifier,
+    table_id: TableId,
+    fn_ptr: ViewFnPtr,
 ) -> SysCallResult<ViewReturnData> {
     let prev_func_type = get_env(scope)?
         .instance_env
@@ -871,8 +865,8 @@ fn call_view(
                 ViewOp {
                     name: view_name,
                     view_id: view_call.view_id,
-                    table_id: view_call.table_id,
-                    fn_ptr: view_call.fn_ptr,
+                    table_id,
+                    fn_ptr,
                     args: &args,
                     sender: &sender,
                     timestamp: Timestamp::now(),
@@ -884,8 +878,8 @@ fn call_view(
                 AnonymousViewOp {
                     name: view_name,
                     view_id: view_call.view_id,
-                    table_id: view_call.table_id,
-                    fn_ptr: view_call.fn_ptr,
+                    table_id,
+                    fn_ptr,
                     args: &args,
                     timestamp: Timestamp::now(),
                 },

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

@@ -1359,6 +1359,7 @@ impl InstanceCommon {
             (Ok(raw), sender) => {
                 // This is wrapped in a closure to simplify error handling.
                 let outcome: Result<ViewOutcome, anyhow::Error> = (|| {
+                    let view_call = ViewCallInfo { view_id, sender };
                     let result = ViewResult::from_return_data(raw).context("Error parsing view result")?;
                     let typespace = self.info.module_def.typespace();
                     let row_product_type = typespace
@@ -1371,28 +1372,14 @@ impl InstanceCommon {
                         ViewResult::Rows(bytes) => deserialize_view_rows(row_type, bytes, typespace)
                             .context("Error deserializing rows returned by view".to_string())?,
                         ViewResult::RawSql(query) => self
-                            .run_query_for_view(
-                                &mut tx,
-                                &query,
-                                &row_product_type,
-                                &ViewCallInfo {
-                                    view_id,
-                                    table_id,
-                                    fn_ptr,
-                                    sender,
-                                },
-                            )
+                            .run_query_for_view(&mut tx, &query, &row_product_type, &view_call)
                             .context("Error executing raw SQL returned by view".to_string())?,
                     };
 
                     let replica_ctx = inst.replica_ctx();
                     let stdb = replica_ctx.relational_db();
-                    let res = match sender {
-                        Some(sender) => stdb.materialize_view(&mut tx, table_id, sender, rows),
-                        None => stdb.materialize_anonymous_view(&mut tx, table_id, rows),
-                    };
-
-                    res.context("Error materializing view")?;
+                    stdb.materialize_view_call(&mut tx, table_id, view_call, rows)
+                        .context("Error materializing view")?;
 
                     Ok(ViewOutcome::Success)
                 })();
@@ -1798,8 +1785,6 @@ impl InstanceOp for ViewOp<'_> {
     fn call_type(&self) -> FuncCallType {
         FuncCallType::View(ViewCallInfo {
             view_id: self.view_id,
-            table_id: self.table_id,
-            fn_ptr: self.fn_ptr,
             sender: Some(*self.sender),
         })
     }
@@ -1828,8 +1813,6 @@ impl InstanceOp for AnonymousViewOp<'_> {
     fn call_type(&self) -> FuncCallType {
         FuncCallType::View(ViewCallInfo {
             view_id: self.view_id,
-            table_id: self.table_id,
-            fn_ptr: self.fn_ptr,
             sender: None,
         })
     }

crates/core/src/host/wasmtime/wasm_instance_env.rs

@@ -19,7 +19,7 @@ use spacetimedb_data_structures::map::IntMap;
 use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, MutTxId, ViewCallInfo};
 use spacetimedb_lib::{bsatn, ConnectionId, Timestamp};
 use spacetimedb_primitives::errno::HOST_CALL_FAILURE;
-use spacetimedb_primitives::{errno, ColId};
+use spacetimedb_primitives::{errno, ColId, ViewFnPtr};
 use spacetimedb_schema::def::ModuleDef;
 use spacetimedb_schema::identifier::Identifier;
 use std::future::Future;
@@ -1697,13 +1697,20 @@ impl WasmInstanceEnv {
 
         for view_call in views_for_refresh {
             let res: anyhow::Result<()> = (|| {
-                let view_def = module_def
-                    .get_view_by_id(view_call.fn_ptr, view_call.sender.is_none())
-                    .ok_or_else(|| anyhow!("view with fn_ptr `{}` not found", view_call.fn_ptr))?;
+                let resolved = crate::host::module_host::resolve_view_for_refresh(
+                    tx.as_ref().expect("procedure tx missing during view refresh"),
+                    &module_def,
+                    &view_call,
+                )?;
+
+                let table_id = resolved.table_id;
+                let view_def = resolved.view_def;
+                let view_name = &view_def.name;
+                let fn_ptr = view_def.fn_ptr;
 
                 let current_tx = tx.take().expect("procedure tx missing during view refresh");
                 let (next_tx, call_result) =
-                    tx_slot.set(current_tx, || Self::call_view(caller, &view_call, &view_def.name));
+                    tx_slot.set(current_tx, || Self::call_view(caller, &view_call, view_name, fn_ptr));
                 tx = Some(next_tx);
                 let return_data = call_result?;
 
@@ -1727,21 +1734,13 @@ impl WasmInstanceEnv {
                 };
 
                 let stdb = caller.data().instance_env.relational_db().clone();
-                match view_call.sender {
-                    Some(sender) => stdb.materialize_view(
-                        tx.as_mut()
-                            .expect("procedure tx missing while materializing authenticated view"),
-                        view_call.table_id,
-                        sender,
-                        rows,
-                    )?,
-                    None => stdb.materialize_anonymous_view(
-                        tx.as_mut()
-                            .expect("procedure tx missing while materializing anonymous view"),
-                        view_call.table_id,
-                        rows,
-                    )?,
-                }
+                stdb.materialize_view_call(
+                    tx.as_mut()
+                        .expect("procedure tx missing while materializing refreshed view"),
+                    table_id,
+                    view_call.clone(),
+                    rows,
+                )?;
 
                 Ok(())
             })();
@@ -1766,6 +1765,7 @@ impl WasmInstanceEnv {
         caller: &mut Caller<'a, Self>,
         view_call: &ViewCallInfo,
         view_name: &Identifier,
+        fn_ptr: ViewFnPtr,
     ) -> anyhow::Result<ViewReturnData> {
         let prev_func_type = caller
             .data_mut()
@@ -1792,7 +1792,7 @@ impl WasmInstanceEnv {
                 call_view,
                 call_view_anon,
                 view_name,
-                view_call.fn_ptr.0,
+                fn_ptr.0,
                 view_call.sender,
                 args_source.0,
                 result_sink,