Merge branch 'master' into kim/compression-stats

Author: kim

crates/core/src/config.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/db/durability.rs

@@ -15,7 +15,7 @@ use tokio::{
     runtime,
     sync::{
         futures::OwnedNotified,
-        mpsc::{channel, unbounded_channel, Receiver, Sender, UnboundedReceiver, UnboundedSender},
+        mpsc::{self, channel, Receiver, Sender},
         oneshot, Notify,
     },
     time::timeout,
@@ -69,7 +69,7 @@ type ShutdownReply = oneshot::Sender<OwnedNotified>;
 /// [RelationalDB]: crate::db::relational_db::RelationalDB
 pub struct DurabilityWorker {
     database: Identity,
-    request_tx: UnboundedSender<DurabilityRequest>,
+    request_tx: Sender<DurabilityRequest>,
     shutdown: Sender<ShutdownReply>,
     durability: Arc<Durability>,
     runtime: runtime::Handle,
@@ -86,7 +86,7 @@ impl DurabilityWorker {
         next_tx_offset: TxOffset,
         reorder_window_size: NonZeroUsize,
     ) -> Self {
-        let (request_tx, request_rx) = unbounded_channel();
+        let (request_tx, request_rx) = channel(4 * 4096);
         let (shutdown_tx, shutdown_rx) = channel(1);
 
         let actor = DurabilityWorkerActor {
@@ -123,8 +123,8 @@ impl DurabilityWorker {
     /// this method is responsible only for reading its decision out of the `tx_data`
     /// and calling `durability.append_tx`.
     ///
-    /// This method does not block,
-    /// and sends the work to an actor that collects data and calls `durability.append_tx`.
+    /// This method sends the work to an actor that collects data and calls `durability.append_tx`.
+    /// It blocks if the queue is at capacity.
     ///
     /// # Panics
     ///
@@ -135,12 +135,40 @@ impl DurabilityWorker {
     /// - [Self::shutdown] was called
     ///
     pub fn request_durability(&self, reducer_context: Option<ReducerContext>, tx_data: &Arc<TxData>) {
-        self.request_tx
-            .send(DurabilityRequest {
-                reducer_context,
-                tx_data: tx_data.clone(),
-            })
-            .unwrap_or_else(|_| panic!("durability actor vanished database={}", self.database));
+        // We first try to send it without blocking.
+        match self.request_tx.try_reserve() {
+            Ok(permit) => {
+                permit.send(DurabilityRequest {
+                    reducer_context,
+                    tx_data: tx_data.clone(),
+                });
+            }
+            Err(mpsc::error::TrySendError::Closed(_)) => {
+                panic!("durability actor vanished database={}", self.database);
+            }
+            Err(mpsc::error::TrySendError::Full(_)) => {
+                // If the channel was full, we use the blocking version.
+                let start = std::time::Instant::now();
+                let send = || {
+                    self.request_tx.blocking_send(DurabilityRequest {
+                        reducer_context,
+                        tx_data: tx_data.clone(),
+                    })
+                };
+                if tokio::runtime::Handle::try_current().is_ok() {
+                    tokio::task::block_in_place(send)
+                } else {
+                    send()
+                }
+                .unwrap_or_else(|_| panic!("durability actor vanished database={}", self.database));
+                // We could cache this metric, but if we are already in the blocking code path,
+                // the extra time of looking up the metric is probably negligible.
+                WORKER_METRICS
+                    .durability_blocking_send_duration
+                    .with_label_values(&self.database)
+                    .observe(start.elapsed().as_secs_f64());
+            }
+        }
     }
 
     /// Get the [`DurableOffset`] of this database.
@@ -281,8 +309,8 @@ impl<T> ReorderWindow<T> {
     }
 }
 
-struct DurabilityWorkerActor {
-    request_rx: UnboundedReceiver<DurabilityRequest>,
+pub struct DurabilityWorkerActor {
+    request_rx: mpsc::Receiver<DurabilityRequest>,
     shutdown: Receiver<ShutdownReply>,
     durability: Arc<Durability>,
     reorder_window: ReorderWindow<DurabilityRequest>,
@@ -483,7 +511,7 @@ mod tests {
         }
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn shutdown_waits_until_durable() {
         let durability = Arc::new(CountingDurability::default());
         let worker = DurabilityWorker::new(

crates/core/src/db/relational_db.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;
 
@@ -1394,32 +1395,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)?)
     }
 

crates/core/src/host/instance_env.rs

@@ -17,7 +17,7 @@ use spacetimedb_client_api_messages::energy::EnergyQuanta;
 use spacetimedb_datastore::db_metrics::DB_METRICS;
 use spacetimedb_datastore::execution_context::Workload;
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
-use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, MutTxId};
+use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, IndexScanPointOrRange, MutTxId};
 use spacetimedb_datastore::traits::IsolationLevel;
 use spacetimedb_lib::{http as st_http, ConnectionId, Identity, Timestamp};
 use spacetimedb_primitives::{ColId, ColList, IndexId, TableId};
@@ -489,9 +489,12 @@ impl InstanceEnv {
         let tx = &mut *self.get_tx()?;
 
         // Find all rows in the table to delete.
-        let (table_id, _, _, iter) = stdb.index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
+        let (table_id, iter) = stdb.index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
         // Re. `SmallVec`, `delete_by_field` only cares about 1 element, so optimize for that.
-        let rows_to_delete = iter.map(|row_ref| row_ref.pointer()).collect::<SmallVec<[_; 1]>>();
+        let rows_to_delete = match iter {
+            IndexScanPointOrRange::Point(_, iter) => iter.map(|row_ref| row_ref.pointer()).collect(),
+            IndexScanPointOrRange::Range(iter) => iter.map(|row_ref| row_ref.pointer()).collect(),
+        };
 
         Ok(Self::datastore_delete_by_index_scan(stdb, tx, table_id, rows_to_delete))
     }
@@ -653,19 +656,22 @@ impl InstanceEnv {
         let tx = &mut *self.get_tx()?;
 
         // Open index iterator
-        let (table_id, lower, upper, iter) =
+        let (table_id, iter) =
             self.relational_db()
                 .index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
 
         // Scan the index and serialize rows to BSATN.
-        let (chunks, rows_scanned, bytes_scanned) = ChunkedWriter::collect_iter(pool, iter);
+        let (point, (chunks, rows_scanned, bytes_scanned)) = match iter {
+            IndexScanPointOrRange::Point(point, iter) => (Some(point), ChunkedWriter::collect_iter(pool, iter)),
+            IndexScanPointOrRange::Range(iter) => (None, ChunkedWriter::collect_iter(pool, iter)),
+        };
 
         // Record the number of rows and the number of bytes scanned by the iterator.
         tx.metrics.index_seeks += 1;
         tx.metrics.bytes_scanned += bytes_scanned;
         tx.metrics.rows_scanned += rows_scanned;
 
-        tx.record_index_scan_range(&self.func_type, table_id, index_id, lower, upper);
+        tx.record_index_scan_range(&self.func_type, table_id, index_id, point);
 
         Ok(chunks)
     }

crates/core/src/subscription/module_subscription_actor.rs

@@ -4272,7 +4272,7 @@ mod tests {
         Ok(())
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn test_confirmed_reads() -> anyhow::Result<()> {
         let (db, durability) = relational_db_with_manual_durability(tokio::runtime::Handle::current())?;
 

crates/core/src/worker_metrics/mod.rs

@@ -481,6 +481,12 @@ metrics_group!(
         #[labels(db: Identity, scan_type: str, table: str, unindexed_columns: str)]
         pub subscription_queries_total: IntCounterVec,
 
+        #[name = spacetime_durability_blocking_send_duration_sec]
+        #[help = "Latency of blocking sends in request_durability (seconds); _count gives the number of times the channel was full"]
+        #[labels(database_identity: Identity)]
+        #[buckets(0.001, 0.01, 0.1, 1.0, 10.0)]
+        pub durability_blocking_send_duration: HistogramVec,
+
         #[name = spacetime_durability_worker_reorder_window_length]
         #[help = "The number of transactions currently being held in the reorder window"]
         #[labels(db: Identity)]