Add ProcedureContext::with_tx (#3638)

# Description of Changes

Adds `ProcedureContext::{with_tx, try_with_tx}`.

Fixes #3515.

# API and ABI breaking changes

None

# Expected complexity level and risk

2

# Testing

An integration test `test_calling_with_tx` is added.

Author: Centril

crates/bindings-sys/src/lib.rs

@@ -645,6 +645,7 @@ pub mod raw {
         pub fn get_jwt(connection_id_ptr: *const u8, bytes_source_id: *mut BytesSource) -> u16;
     }
 
+    #[cfg(feature = "unstable")]
     #[link(wasm_import_module = "spacetime_10.3")]
     extern "C" {
         /// Suspends execution of this WASM instance until approximately `wake_at_micros_since_unix_epoch`.
@@ -661,8 +662,76 @@ pub mod raw {
         /// - The calling WASM instance is holding open a transaction.
         /// - The calling WASM instance is not executing a procedure.
         // TODO(procedure-sleep-until): remove this
-        #[cfg(feature = "unstable")]
         pub fn procedure_sleep_until(wake_at_micros_since_unix_epoch: i64) -> i64;
+
+        /// Starts a mutable transaction,
+        /// suspending execution of this WASM instance until
+        /// a mutable transaction lock is aquired.
+        ///
+        /// Upon resuming, returns `0` on success,
+        /// enabling further calls that require a pending transaction,
+        /// or an error code otherwise.
+        ///
+        /// # Traps
+        ///
+        /// Traps if:
+        /// - `out` is NULL or `out[..size_of::<i64>()]` is not in bounds of WASM memory.
+        ///
+        /// # Errors
+        ///
+        /// Returns an error:
+        ///
+        /// - `WOULD_BLOCK_TRANSACTION`, if there's already an ongoing transaction.
+        pub fn procedure_start_mut_tx(out: *mut i64) -> u16;
+
+        /// Commits a mutable transaction,
+        /// suspending execution of this WASM instance until
+        /// the transaction has been committed
+        /// and subscription queries have been run and broadcast.
+        ///
+        /// Upon resuming, returns `0` on success, or an error code otherwise.
+        ///
+        /// # Traps
+        ///
+        /// This function does not trap.
+        ///
+        /// # Errors
+        ///
+        /// Returns an error:
+        ///
+        /// - `TRANSACTION_NOT_ANONYMOUS`,
+        ///   if the transaction was not started in [`procedure_start_mut_tx`].
+        ///   This can happen if this syscall is erroneously called by a reducer.
+        ///   The code `NOT_IN_TRANSACTION` does not happen,
+        ///   as it is subsumed by `TRANSACTION_NOT_ANONYMOUS`.
+        /// - `TRANSACTION_IS_READ_ONLY`, if the pending transaction is read-only.
+        ///   This currently does not happen as anonymous read transactions
+        ///   are not exposed to modules.
+        pub fn procedure_commit_mut_tx() -> u16;
+
+        /// Aborts a mutable transaction,
+        /// suspending execution of this WASM instance until
+        /// the transaction has been rolled back.
+        ///
+        /// Upon resuming, returns `0` on success, or an error code otherwise.
+        ///
+        /// # Traps
+        ///
+        /// This function does not trap.
+        ///
+        /// # Errors
+        ///
+        /// Returns an error:
+        ///
+        /// - `TRANSACTION_NOT_ANONYMOUS`,
+        ///   if the transaction was not started in [`procedure_start_mut_tx`].
+        ///   This can happen if this syscall is erroneously called by a reducer.
+        ///   The code `NOT_IN_TRANSACTION` does not happen,
+        ///   as it is subsumed by `TRANSACTION_NOT_ANONYMOUS`.
+        /// - `TRANSACTION_IS_READ_ONLY`, if the pending transaction is read-only.
+        ///   This currently does not happen as anonymous read transactions
+        ///   are not exposed to modules.
+        pub fn procedure_abort_mut_tx() -> u16;
     }
 
     /// What strategy does the database index use?
@@ -819,7 +888,7 @@ impl fmt::Display for Errno {
 
 /// Convert the status value `x` into a result.
 /// When `x = 0`, we have a success status.
-fn cvt(x: u16) -> Result<(), Errno> {
+fn cvt(x: u16) -> Result<()> {
     match Errno::from_code(x) {
         None => Ok(()),
         Some(err) => Err(err),
@@ -838,13 +907,25 @@ fn cvt(x: u16) -> Result<(), Errno> {
 /// - The function `f` never reads a safe and valid `T` from the `out` pointer
 ///   before writing a safe and valid `T` to it.
 #[inline]
-unsafe fn call<T: Copy>(f: impl FnOnce(*mut T) -> u16) -> Result<T, Errno> {
+unsafe fn call<T: Copy>(f: impl FnOnce(*mut T) -> u16) -> Result<T> {
     let mut out = MaybeUninit::uninit();
     let f_code = f(out.as_mut_ptr());
     cvt(f_code)?;
     Ok(out.assume_init())
 }
 
+/// Runs the given function `f`.
+///
+/// Assuming the call to `f` returns 0, `Ok(())` is returned,
+/// and otherwise `Err(err)` is returned.
+#[inline]
+#[cfg(feature = "unstable")]
+fn call_no_ret(f: impl FnOnce() -> u16) -> Result<()> {
+    let f_code = f();
+    cvt(f_code)?;
+    Ok(())
+}
+
 /// Queries the `table_id` associated with the given (table) `name`.
 ///
 /// The table id is returned.
@@ -856,7 +937,7 @@ unsafe fn call<T: Copy>(f: impl FnOnce(*mut T) -> u16) -> Result<T, Errno> {
 /// - `NOT_IN_TRANSACTION`, when called outside of a transaction.
 /// - `NO_SUCH_TABLE`, when `name` is not the name of a table.
 #[inline]
-pub fn table_id_from_name(name: &str) -> Result<TableId, Errno> {
+pub fn table_id_from_name(name: &str) -> Result<TableId> {
     unsafe { call(|out| raw::table_id_from_name(name.as_ptr(), name.len(), out)) }
 }
 
@@ -871,7 +952,7 @@ pub fn table_id_from_name(name: &str) -> Result<TableId, Errno> {
 /// - `NOT_IN_TRANSACTION`, when called outside of a transaction.
 /// - `NO_SUCH_INDEX`, when `name` is not the name of an index.
 #[inline]
-pub fn index_id_from_name(name: &str) -> Result<IndexId, Errno> {
+pub fn index_id_from_name(name: &str) -> Result<IndexId> {
     unsafe { call(|out| raw::index_id_from_name(name.as_ptr(), name.len(), out)) }
 }
 
@@ -884,7 +965,7 @@ pub fn index_id_from_name(name: &str) -> Result<IndexId, Errno> {
 /// - `NOT_IN_TRANSACTION`, when called outside of a transaction.
 /// - `NO_SUCH_TABLE`, when `table_id` is not a known ID of a table.
 #[inline]
-pub fn datastore_table_row_count(table_id: TableId) -> Result<u64, Errno> {
+pub fn datastore_table_row_count(table_id: TableId) -> Result<u64> {
     unsafe { call(|out| raw::datastore_table_row_count(table_id, out)) }
 }
 
@@ -901,7 +982,7 @@ pub fn datastore_table_row_count(table_id: TableId) -> Result<u64, Errno> {
 /// - `row` doesn't decode from BSATN to a `ProductValue`
 ///   according to the `ProductType` that the table's schema specifies.
 #[inline]
-pub fn datastore_insert_bsatn(table_id: TableId, row: &mut [u8]) -> Result<&[u8], Errno> {
+pub fn datastore_insert_bsatn(table_id: TableId, row: &mut [u8]) -> Result<&[u8]> {
     let row_ptr = row.as_mut_ptr();
     let row_len = &mut row.len();
     cvt(unsafe { raw::datastore_insert_bsatn(table_id, row_ptr, row_len) }).map(|()| &row[..*row_len])
@@ -927,7 +1008,7 @@ pub fn datastore_insert_bsatn(table_id: TableId, row: &mut [u8]) -> Result<&[u8]
 ///   or if `row` cannot project to the index's type.
 /// - the row was not found
 #[inline]
-pub fn datastore_update_bsatn(table_id: TableId, index_id: IndexId, row: &mut [u8]) -> Result<&[u8], Errno> {
+pub fn datastore_update_bsatn(table_id: TableId, index_id: IndexId, row: &mut [u8]) -> Result<&[u8]> {
     let row_ptr = row.as_mut_ptr();
     let row_len = &mut row.len();
     cvt(unsafe { raw::datastore_update_bsatn(table_id, index_id, row_ptr, row_len) }).map(|()| &row[..*row_len])
@@ -954,7 +1035,7 @@ pub fn datastore_update_bsatn(table_id: TableId, index_id: IndexId, row: &mut [u
 /// - `BSATN_DECODE_ERROR`, when `rel` cannot be decoded to `Vec<ProductValue>`
 ///   where each `ProductValue` is typed at the `ProductType` the table's schema specifies.
 #[inline]
-pub fn datastore_delete_all_by_eq_bsatn(table_id: TableId, relation: &[u8]) -> Result<u32, Errno> {
+pub fn datastore_delete_all_by_eq_bsatn(table_id: TableId, relation: &[u8]) -> Result<u32> {
     unsafe { call(|out| raw::datastore_delete_all_by_eq_bsatn(table_id, relation.as_ptr(), relation.len(), out)) }
 }
 
@@ -968,7 +1049,7 @@ pub fn datastore_delete_all_by_eq_bsatn(table_id: TableId, relation: &[u8]) -> R
 ///
 /// - `NOT_IN_TRANSACTION`, when called outside of a transaction.
 /// - `NO_SUCH_TABLE`, when `table_id` is not a known ID of a table.
-pub fn datastore_table_scan_bsatn(table_id: TableId) -> Result<RowIter, Errno> {
+pub fn datastore_table_scan_bsatn(table_id: TableId) -> Result<RowIter> {
     let raw = unsafe { call(|out| raw::datastore_table_scan_bsatn(table_id, out))? };
     Ok(RowIter { raw })
 }
@@ -1028,7 +1109,7 @@ pub fn datastore_index_scan_range_bsatn(
     prefix_elems: ColId,
     rstart: &[u8],
     rend: &[u8],
-) -> Result<RowIter, Errno> {
+) -> Result<RowIter> {
     let raw = unsafe {
         call(|out| {
             raw::datastore_index_scan_range_bsatn(
@@ -1076,7 +1157,7 @@ pub fn datastore_delete_by_index_scan_range_bsatn(
     prefix_elems: ColId,
     rstart: &[u8],
     rend: &[u8],
-) -> Result<u32, Errno> {
+) -> Result<u32> {
     unsafe {
         call(|out| {
             raw::datastore_delete_by_index_scan_range_bsatn(
@@ -1239,13 +1320,81 @@ impl Drop for RowIter {
     }
 }
 
+#[cfg(feature = "unstable")]
 pub mod procedure {
     //! Side-effecting or asynchronous operations which only procedures are allowed to perform.
+
+    use super::{call, call_no_ret, raw, Result};
+
     #[inline]
-    #[cfg(feature = "unstable")]
     pub fn sleep_until(wake_at_timestamp: i64) -> i64 {
         // Safety: Just calling an `extern "C"` function.
         // Nothing weird happening here.
-        unsafe { super::raw::procedure_sleep_until(wake_at_timestamp) }
+        unsafe { raw::procedure_sleep_until(wake_at_timestamp) }
+    }
+
+    /// Starts a mutable transaction,
+    /// suspending execution of this WASM instance until
+    /// a mutable transaction lock is aquired.
+    ///
+    /// Upon resuming, returns `Ok(timestamp)` on success,
+    /// enabling further calls that require a pending transaction,
+    /// or [`Errno`] otherwise.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error:
+    ///
+    /// - `WOULD_BLOCK_TRANSACTION`, if there's already an ongoing transaction.
+    #[inline]
+    pub fn procedure_start_mut_tx() -> Result<i64> {
+        unsafe { call(|out| raw::procedure_start_mut_tx(out)) }
+    }
+
+    /// Commits a mutable transaction,
+    /// suspending execution of this WASM instance until
+    /// the transaction has been committed
+    /// and subscription queries have been run and broadcast.
+    ///
+    /// Upon resuming, returns `Ok(()` on success, or an [`Errno`] otherwise.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error:
+    ///
+    /// - `TRANSACTION_NOT_ANONYMOUS`,
+    ///   if the transaction was not started in [`procedure_start_mut_tx`].
+    ///   This can happen if this syscall is erroneously called by a reducer.
+    ///   The code `NOT_IN_TRANSACTION` does not happen,
+    ///   as it is subsumed by `TRANSACTION_NOT_ANONYMOUS`.
+    /// - `TRANSACTION_IS_READ_ONLY`, if the pending transaction is read-only.
+    ///   This currently does not happen as anonymous read transactions
+    ///   are not exposed to modules.
+    #[inline]
+    pub fn procedure_commit_mut_tx() -> Result<()> {
+        call_no_ret(|| unsafe { raw::procedure_commit_mut_tx() })
+    }
+
+    /// Aborts a mutable transaction,
+    /// suspending execution of this WASM instance until
+    /// the transaction has been rolled back.
+    ///
+    /// Upon resuming, returns `Ok(())` on success, or an [`Errno`] otherwise.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error:
+    ///
+    /// - `TRANSACTION_NOT_ANONYMOUS`,
+    ///   if the transaction was not started in [`procedure_start_mut_tx`].
+    ///   This can happen if this syscall is erroneously called by a reducer.
+    ///   The code `NOT_IN_TRANSACTION` does not happen,
+    ///   as it is subsumed by `TRANSACTION_NOT_ANONYMOUS`.
+    /// - `TRANSACTION_IS_READ_ONLY`, if the pending transaction is read-only.
+    ///   This currently does not happen as anonymous read transactions
+    ///   are not exposed to modules.
+    #[inline]
+    pub fn procedure_abort_mut_tx() -> Result<()> {
+        call_no_ret(|| unsafe { raw::procedure_abort_mut_tx() })
     }
 }