Merge branch 'master' into joshua/wasm-sync

Author: joshua-spacetime

crates/client-api-messages/src/energy.rs

@@ -16,17 +16,13 @@ pub struct EnergyQuanta {
 impl EnergyQuanta {
     pub const ZERO: Self = EnergyQuanta { quanta: 0 };
 
-    // per the comment on [`FunctionBudget::DEFAULT_BUDGET`]: 1 second of wasm runtime is roughtly 2 TeV
-    pub const PER_EXECUTION_SEC: Self = Self::new(2_000_000_000_000);
-    pub const PER_EXECUTION_NANOSEC: Self = Self::new(Self::PER_EXECUTION_SEC.get() / 1_000_000_000);
-
     #[inline]
-    pub const fn new(quanta: u128) -> Self {
+    pub fn new(quanta: u128) -> Self {
         Self { quanta }
     }
 
     #[inline]
-    pub const fn get(&self) -> u128 {
+    pub fn get(&self) -> u128 {
         self.quanta
     }
 

crates/commitlog/src/commit.rs

@@ -67,9 +67,9 @@ impl Header {
         match &mut hdr.as_slice() {
             [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] => Ok(None),
             buf => {
-                let min_tx_offset = buf.get_u64().map_err(decode_error)?;
-                let n = buf.get_u16().map_err(decode_error)?;
-                let len = buf.get_u32().map_err(decode_error)?;
+                let min_tx_offset = buf.get_u64().map_err(|e| decode_header_error(e, "min_tx_offset"))?;
+                let n = buf.get_u16().map_err(|e| decode_header_error(e, "n"))?;
+                let len = buf.get_u32().map_err(|e| decode_header_error(e, "len"))?;
 
                 Ok(Some(Self {
                     min_tx_offset,
@@ -88,15 +88,15 @@ impl Header {
                 return Ok(None);
             }
 
-            return Err(e);
+            return Err(io::Error::new(e.kind(), format!("error reading commit header: {e}")));
         }
         match &mut hdr.as_slice() {
             [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] => Ok(None),
             buf => {
-                let min_tx_offset = buf.get_u64().map_err(decode_error)?;
-                let epoch = buf.get_u64().map_err(decode_error)?;
-                let n = buf.get_u16().map_err(decode_error)?;
-                let len = buf.get_u32().map_err(decode_error)?;
+                let min_tx_offset = buf.get_u64().map_err(|e| decode_header_error(e, "min_tx_offset"))?;
+                let epoch = buf.get_u64().map_err(|e| decode_header_error(e, "epoch"))?;
+                let n = buf.get_u16().map_err(|e| decode_header_error(e, "n"))?;
+                let len = buf.get_u32().map_err(|e| decode_header_error(e, "len"))?;
 
                 Ok(Some(Self {
                     min_tx_offset,
@@ -310,10 +310,16 @@ impl StoredCommit {
             return Ok(None);
         };
         let mut records = vec![0; hdr.len as usize];
-        reader.read_exact(&mut records)?;
+        reader.read_exact(&mut records).map_err(|e| {
+            io::Error::new(
+                e.kind(),
+                format!("failed to read {} bytes of commit payload: {}", hdr.len, e),
+            )
+        })?;
 
         let chk = reader.crc32c();
-        let crc = decode_u32(reader.into_inner())?;
+        let crc = decode_u32(reader.into_inner())
+            .map_err(|e| io::Error::new(e.kind(), format!("failed to read checksum: {e}")))?;
 
         if chk != crc {
             return Err(invalid_data(ChecksumMismatch));
@@ -385,8 +391,8 @@ fn decode_u32<R: Read>(mut read: R) -> io::Result<u32> {
     Ok(u32::from_le_bytes(buf))
 }
 
-fn decode_error(e: DecodeError) -> io::Error {
-    invalid_data(e)
+fn decode_header_error(e: DecodeError, field: &str) -> io::Error {
+    invalid_data(format!("failed to decode commit header field '{field}': {e}"))
 }
 
 fn invalid_data<E>(e: E) -> io::Error

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

@@ -13,7 +13,7 @@ use core::ptr;
 use core::sync::atomic::Ordering;
 use core::time::Duration;
 use core::{ffi::c_void, sync::atomic::AtomicBool};
-use spacetimedb_client_api_messages::energy::{EnergyQuanta, FunctionBudget};
+use spacetimedb_client_api_messages::energy::FunctionBudget;
 use std::sync::Arc;
 use v8::{Isolate, IsolateHandle};
 
@@ -118,12 +118,7 @@ fn budget_to_duration(_budget: FunctionBudget) -> Duration {
 /// Returns [`EnergyStats`] for a reducer given its `budget`
 /// and the `duration` it took to execute.
 pub(super) fn energy_from_elapsed(budget: FunctionBudget, duration: Duration) -> EnergyStats {
-    let used = duration.as_nanos() * EnergyQuanta::PER_EXECUTION_NANOSEC.get();
-    // in order for duration_nanos * ev_per_ns >= u64::MAX:
-    //              duration_nanos >= u64::MAX / ev_per_ns
-    //              duration_nanos >= (9223372036854775 ns = 106.75 days)
-    // so it's unlikely we'll have to worry about it
-    let used = FunctionBudget::new(u64::try_from(used).unwrap_or(u64::MAX));
+    let used = duration_to_budget(duration);
     let remaining = budget - used;
     EnergyStats { budget, remaining }
 }