Merge branch 'master' into jsdt/broadcast-one-offs

Author: jsdt

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

@@ -12,6 +12,7 @@ mod internal;
 pub mod metrics;
 pub mod prometheus;
 pub mod subscribe;
+pub mod unstable;
 
 /// This API call is just designed to allow clients to determine whether or not they can
 /// establish a connection to SpacetimeDB. This API call doesn't actually do anything.
@@ -40,4 +41,5 @@ where
     axum::Router::new()
         .nest("/v1", router.layer(cors))
         .nest("/internal", internal::router())
+        .nest("/unstable", unstable::router())
 }

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

@@ -296,7 +296,8 @@ async fn ws_client_actor_inner(
                     // TODO: this isn't great. when we receive a close request from the peer,
                     //       tungstenite doesn't let us send any new messages on the socket,
                     //       even though the websocket RFC allows it. should we fork tungstenite?
-                    log::info!("dropping messages due to ws already being closed: {:?}", &rx_buf[..n]);
+                    log::info!("dropping {n} messages due to ws already being closed");
+                    log::debug!("dropped messages: {:?}", &rx_buf[..n]);
                 } else {
                 let send_all = async {
                     for msg in rx_buf.drain(..n) {

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

@@ -0,0 +1,21 @@
+use axum::response::IntoResponse;
+use spacetimedb_lib::{sats, Timestamp};
+
+use crate::NodeDelegate;
+
+/// Returns the database's view of the current time,
+/// as a SATS-JSON encoded [`Timestamp`].
+async fn get_timestamp() -> impl IntoResponse {
+    axum::Json(sats::serde::SerdeWrapper(Timestamp::now())).into_response()
+}
+
+/// The internal router is for routes which are early in design,
+/// and may incompatibly change or be removed without a major version bump.
+pub fn router<S>() -> axum::Router<S>
+where
+    S: NodeDelegate + Clone + 'static,
+{
+    use axum::routing::get;
+
+    axum::Router::new().route("/timestamp", get(get_timestamp))
+}

crates/core/src/db/datastore/system_tables.rs

@@ -813,10 +813,18 @@ impl From<StRowLevelSecurityRow> for RowLevelSecuritySchema {
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub struct ModuleKind(u8);
 
-/// The [`ModuleKind`] of WASM-based modules.
-///
-/// This is currently the only known kind.
-pub const WASM_MODULE: ModuleKind = ModuleKind(0);
+impl ModuleKind {
+    /// The [`ModuleKind`] of WASM-based modules.
+    pub const WASM: ModuleKind = ModuleKind(0);
+}
+
+impl From<crate::messages::control_db::HostType> for ModuleKind {
+    fn from(host_type: crate::messages::control_db::HostType) -> Self {
+        match host_type {
+            crate::messages::control_db::HostType::Wasm => Self::WASM,
+        }
+    }
+}
 
 impl_serialize!([] ModuleKind, (self, ser) => self.0.serialize(ser));
 impl_deserialize!([] ModuleKind, de => u8::deserialize(de).map(Self));
@@ -852,7 +860,7 @@ impl From<Identity> for IdentityViaU256 {
 ///
 /// * `database_identity` is the [`Identity`] of the database.
 /// * `owner_identity` is the [`Identity`] of the owner of the database.
-/// * `program_kind` is the [`ModuleKind`] (currently always [`WASM_MODULE`]).
+/// * `program_kind` is the [`ModuleKind`] (currently always [`ModuleKind::WASM`]).
 /// * `program_hash` is the [`Hash`] of the raw bytes of the (compiled) module.
 /// * `program_bytes` are the raw bytes of the (compiled) module.
 /// * `module_version` is the version of the module.

crates/core/src/db/relational_db.rs

@@ -17,7 +17,7 @@ use super::datastore::{
     traits::TxData,
 };
 use super::db_metrics::DB_METRICS;
-use crate::db::datastore::system_tables::{StModuleRow, WASM_MODULE};
+use crate::db::datastore::system_tables::StModuleRow;
 use crate::error::{DBError, DatabaseError, RestoreSnapshotError};
 use crate::execution_context::{ReducerContext, Workload, WorkloadType};
 use crate::messages::control_db::HostType;
@@ -427,9 +427,7 @@ impl RelationalDB {
             database_identity: self.database_identity.into(),
             owner_identity: self.owner_identity.into(),
 
-            program_kind: match host_type {
-                HostType::Wasm => WASM_MODULE,
-            },
+            program_kind: host_type.into(),
             program_hash: program.hash,
             program_bytes: program.bytes,
             module_version: ONLY_MODULE_VERSION.into(),
@@ -473,10 +471,7 @@ impl RelationalDB {
     /// - the `__init__` reducer contained in the module has been executed
     ///   within the transactional context `tx`.
     pub fn update_program(&self, tx: &mut MutTx, host_type: HostType, program: Program) -> Result<(), DBError> {
-        let program_kind = match host_type {
-            HostType::Wasm => WASM_MODULE,
-        };
-        Ok(self.inner.update_program(tx, program_kind, program)?)
+        Ok(self.inner.update_program(tx, host_type.into(), program)?)
     }
 
     fn restore_from_snapshot_or_bootstrap(

crates/core/src/startup.rs

@@ -1,5 +1,6 @@
 use core_affinity::CoreId;
 use crossbeam_queue::ArrayQueue;
+use itertools::Itertools;
 use spacetimedb_paths::server::{ConfigToml, LogsDir};
 use std::path::PathBuf;
 use std::time::Duration;
@@ -194,7 +195,14 @@ pub struct Cores {
 impl Cores {
     fn get() -> Option<Self> {
         let cores = &mut core_affinity::get_core_ids()
-            .filter(|cores| cores.len() >= 8)?
+            .filter(|cores| cores.len() >= 10)?
+            .into_iter()
+            // We reserve the first two cores for the OS.
+            // This allows us to pin interrupt handlers (IRQs) to these cores,
+            // particularly those for incoming network traffic,
+            // preventing them from preempting the main reducer threads.
+            .filter(|core_id| core_id.id > 1)
+            .collect_vec()
             .into_iter();
 
         let total = cores.len() as f64;

smoketests/__init__.py

@@ -264,10 +264,11 @@ def api_call(self, method, path, body = None, headers = {}):
             log_cmd([method, path])
             conn.request(method, path, body, headers)
             resp = conn.getresponse()
-            logging.debug(f"{resp.status} {resp.read()}")
+            body = resp.read()
+            logging.debug(f"{resp.status} {body}")
             if resp.status != 200:
                 raise resp
-            resp
+            return body
 
 
     @classmethod

smoketests/tests/timestamp_route.py

@@ -0,0 +1,19 @@
+from .. import Smoketest, random_string
+import unittest
+import json
+import io
+
+TIMESTAMP_TAG = "__timestamp_micros_since_unix_epoch__"
+
+class TimestampRoute(Smoketest):
+    AUTOPUBLISH = False
+
+    def test_timestamp_route(self):
+        resp = self.api_call(
+            "GET",
+            "/unstable/timestamp",
+        )
+        timestamp = json.load(io.BytesIO(resp))
+        self.assertIsInstance(timestamp, dict)
+        self.assertIn(TIMESTAMP_TAG, timestamp)
+        self.assertIsInstance(timestamp[TIMESTAMP_TAG], int)