Merge branch 'master' into centril/reuse-ws-msg-allocations

Author: gefjon

crates/cli/src/subcommands/dns.rs

@@ -1,10 +1,10 @@
 use crate::common_args;
 use crate::config::Config;
-use crate::util::{add_auth_header_opt, decode_identity, get_auth_header, get_login_token_or_log_in, ResponseExt};
+use crate::util::{add_auth_header_opt, get_auth_header, ResponseExt};
 use clap::ArgMatches;
 use clap::{Arg, Command};
 
-use spacetimedb_client_api_messages::name::{DomainName, InsertDomainResult};
+use spacetimedb_client_api_messages::name::{DomainName, SetDomainsResult};
 
 pub fn cli() -> Command {
     Command::new("rename")
@@ -30,70 +30,39 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     let database_identity = args.get_one::<String>("database-identity").unwrap();
     let server = args.get_one::<String>("server").map(|s| s.as_ref());
     let force = args.get_flag("force");
-    let token = get_login_token_or_log_in(&mut config, server, !force).await?;
-    let identity = decode_identity(&token)?;
     let auth_header = get_auth_header(&mut config, false, server, !force).await?;
 
     let domain: DomainName = domain.parse()?;
 
     let builder = reqwest::Client::new()
-        .post(format!(
+        .put(format!(
             "{}/v1/database/{database_identity}/names",
             config.get_host_url(server)?
         ))
-        .body(String::from(domain));
+        .header(reqwest::header::CONTENT_TYPE, "application/json")
+        .body(serde_json::to_string(&[&domain])?);
     let builder = add_auth_header_opt(builder, &auth_header);
 
-    let result = builder.send().await?.json_or_error().await?;
-    match result {
-        InsertDomainResult::Success {
-            domain,
-            database_identity,
-        } => {
-            println!("Domain set to {} for identity {}.", domain, database_identity);
-        }
-        InsertDomainResult::TldNotRegistered { domain } => {
-            return Err(anyhow::anyhow!(
-                "The top level domain that you provided is not registered.\n\
-            This tld is not yet registered to any identity. You can register this domain with the following command:\n\
-            \n\
-            \tspacetime dns register-tld {}\n",
-                domain.tld()
-            ));
-        }
-        InsertDomainResult::PermissionDenied { domain } => {
-            //TODO(jdetter): Have a nice name generator here, instead of using some abstract characters
-            // we should perhaps generate fun names like 'green-fire-dragon' instead
-            let suggested_tld: String = identity.chars().take(12).collect();
-            if let Some(sub_domain) = domain.sub_domain() {
-                return Err(anyhow::anyhow!(
-                    "The top level domain {} is not registered to the identity you provided.\n\
-                We suggest you register a new tld:\n\
-                \tspacetime dns register-tld {}\n\
-                \n\
-                And then push to the domain that uses that tld:\n\
-                \tspacetime publish {}/{}\n",
-                    domain.tld(),
-                    suggested_tld,
-                    suggested_tld,
-                    sub_domain
-                ));
-            } else {
-                return Err(anyhow::anyhow!(
-                    "The top level domain {} is not registered to the identity you provided.\n\
-                We suggest you register a new tld:\n\
-                \tspacetime dns register-tld {}\n\
-                \n\
-                And then push to the domain that uses that tld:\n\
-                \tspacetime publish {}\n",
-                    domain.tld(),
-                    suggested_tld,
-                    suggested_tld
-                ));
-            }
-        }
-        InsertDomainResult::OtherError(e) => return Err(anyhow::anyhow!(e)),
+    let response = builder.send().await?;
+    let status = &response.status();
+    let result: SetDomainsResult = response.json_or_error().await?;
+
+    if !status.is_success() {
+        anyhow::bail!(match result {
+            SetDomainsResult::Success => "".to_string(),
+            SetDomainsResult::PermissionDenied { domain } => format!("Permission denied for domain: {}", domain),
+            SetDomainsResult::PermissionDeniedOnAny { domains } =>
+                format!("Permission denied for domains: {:?}", domains),
+            SetDomainsResult::DatabaseNotFound => format!("Database {} not found", database_identity),
+            SetDomainsResult::NotYourDatabase { .. } => format!(
+                "You cannot rename {} because it is owned by another identity.",
+                database_identity
+            ),
+            SetDomainsResult::OtherError(err) => err,
+        });
     }
 
+    println!("Name set to {} for identity {}.", domain, database_identity);
+
     Ok(())
 }

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

@@ -29,7 +29,7 @@ use spacetimedb::messages::control_db::{Database, HostType};
 use spacetimedb_client_api_messages::name::{self, DatabaseName, DomainName, PublishOp, PublishResult};
 use spacetimedb_lib::db::raw_def::v9::RawModuleDefV9;
 use spacetimedb_lib::identity::AuthCtx;
-use spacetimedb_lib::sats;
+use spacetimedb_lib::{sats, Timestamp};
 
 use super::subscribe::handle_websocket;
 
@@ -371,7 +371,7 @@ fn mime_ndjson() -> mime::Mime {
     "application/x-ndjson".parse().unwrap()
 }
 
-async fn worker_ctx_find_database(
+pub(crate) async fn worker_ctx_find_database(
     worker_ctx: &(impl ControlStateDelegate + ?Sized),
     database_identity: &Identity,
 ) -> axum::response::Result<Option<Database>> {
@@ -704,6 +704,18 @@ pub async fn set_names<S: ControlStateDelegate>(
         ));
     }
 
+    for name in &validated_names {
+        if ctx.lookup_identity(name.as_str()).unwrap().is_some() {
+            return Ok((
+                StatusCode::BAD_REQUEST,
+                axum::Json(name::SetDomainsResult::OtherError(format!(
+                    "Cannot rename to {} because it already is in use.",
+                    name.as_str()
+                ))),
+            ));
+        }
+    }
+
     let response = ctx
         .replace_dns_records(&database_identity, &database.owner_identity, &validated_names)
         .await
@@ -720,6 +732,33 @@ pub async fn set_names<S: ControlStateDelegate>(
     Ok((status, axum::Json(response)))
 }
 
+#[derive(serde::Deserialize)]
+pub struct TimestampParams {
+    name_or_identity: NameOrIdentity,
+}
+
+/// Returns the database's view of the current time,
+/// as a SATS-JSON encoded [`Timestamp`].
+///
+/// Takes a particular database's [`NameOrIdentity`] as an argument
+/// because in a clusterized SpacetimeDB-cloud deployment,
+/// this request will be routed to the node running the requested database.
+async fn get_timestamp<S: ControlStateDelegate>(
+    State(worker_ctx): State<S>,
+    Path(TimestampParams { name_or_identity }): Path<TimestampParams>,
+) -> axum::response::Result<impl IntoResponse> {
+    let db_identity = name_or_identity.resolve(&worker_ctx).await?;
+
+    let _database = worker_ctx_find_database(&worker_ctx, &db_identity)
+        .await?
+        .ok_or_else(|| {
+            log::error!("Could not find database: {}", db_identity.to_hex());
+            NO_SUCH_DATABASE
+        })?;
+
+    Ok(axum::Json(sats::serde::SerdeWrapper(Timestamp::now())).into_response())
+}
+
 /// This struct allows the edition to customize `/database` routes more meticulously.
 pub struct DatabaseRoutes<S> {
     /// POST /database
@@ -748,6 +787,9 @@ pub struct DatabaseRoutes<S> {
     pub logs_get: MethodRouter<S>,
     /// POST: /database/:name_or_identity/sql
     pub sql_post: MethodRouter<S>,
+
+    /// GET: /database/: name_or_identity/unstable/timestamp
+    pub timestamp_get: MethodRouter<S>,
 }
 
 impl<S> Default for DatabaseRoutes<S>
@@ -770,6 +812,7 @@ where
             schema_get: get(schema::<S>),
             logs_get: get(logs::<S>),
             sql_post: post(sql::<S>),
+            timestamp_get: get(get_timestamp::<S>),
         }
     }
 }
@@ -791,7 +834,8 @@ where
             .route("/call/:reducer", self.call_reducer_post)
             .route("/schema", self.schema_get)
             .route("/logs", self.logs_get)
-            .route("/sql", self.sql_post);
+            .route("/sql", self.sql_post)
+            .route("/unstable/timestamp", self.timestamp_get);
 
         axum::Router::new()
             .route("/", self.root_post)

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

@@ -12,7 +12,6 @@ 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.
@@ -41,5 +40,4 @@ where
     axum::Router::new()
         .nest("/v1", router.layer(cors))
         .nest("/internal", internal::router())
-        .nest("/unstable", unstable::router())
 }

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

@@ -19,7 +19,7 @@ use futures::prelude::*;
 use prometheus::{Histogram, IntCounter, IntGauge};
 use spacetimedb_client_api_messages::websocket::{
     BsatnFormat, CallReducerFlags, Compression, FormatSwitch, JsonFormat, SubscribeMulti, SubscribeSingle, Unsubscribe,
-    UnsubscribeMulti, WebsocketFormat,
+    UnsubscribeMulti,
 };
 use spacetimedb_lib::identity::RequestId;
 use spacetimedb_lib::metrics::ExecutionMetrics;
@@ -450,41 +450,40 @@ impl ClientConnection {
         .await
     }
 
-    pub fn one_off_query_json(&self, query: &str, message_id: &[u8], timer: Instant) -> Result<(), anyhow::Error> {
-        let response = self.one_off_query::<JsonFormat>(query, message_id, timer);
-        self.send_message(response)?;
-        Ok(())
-    }
-
-    pub fn one_off_query_bsatn(&self, query: &str, message_id: &[u8], timer: Instant) -> Result<(), anyhow::Error> {
-        let response = self.one_off_query::<BsatnFormat>(query, message_id, timer);
-        self.send_message(response)?;
-        Ok(())
+    pub async fn one_off_query_json(
+        &self,
+        query: &str,
+        message_id: &[u8],
+        timer: Instant,
+    ) -> Result<(), anyhow::Error> {
+        self.module
+            .one_off_query::<JsonFormat>(
+                self.id.identity,
+                query.to_owned(),
+                self.sender.clone(),
+                message_id.to_owned(),
+                timer,
+                |msg: OneOffQueryResponseMessage<JsonFormat>| msg.into(),
+            )
+            .await
     }
 
-    fn one_off_query<F: WebsocketFormat>(
+    pub async fn one_off_query_bsatn(
         &self,
         query: &str,
         message_id: &[u8],
         timer: Instant,
-    ) -> OneOffQueryResponseMessage<F> {
-        let result = self.module.one_off_query::<F>(self.id.identity, query.to_owned());
-        let message_id = message_id.to_owned();
-        let total_host_execution_duration = timer.elapsed().into();
-        match result {
-            Ok(results) => OneOffQueryResponseMessage {
-                message_id,
-                error: None,
-                results: vec![results],
-                total_host_execution_duration,
-            },
-            Err(err) => OneOffQueryResponseMessage {
-                message_id,
-                error: Some(format!("{}", err)),
-                results: vec![],
-                total_host_execution_duration,
-            },
-        }
+    ) -> Result<(), anyhow::Error> {
+        self.module
+            .one_off_query::<BsatnFormat>(
+                self.id.identity,
+                query.to_owned(),
+                self.sender.clone(),
+                message_id.to_owned(),
+                timer,
+                |msg: OneOffQueryResponseMessage<BsatnFormat>| msg.into(),
+            )
+            .await
     }
 
     pub async fn disconnect(self) {

crates/core/src/client/client_connection.rs

@@ -120,8 +120,8 @@ pub async fn handle(client: &ClientConnection, message: DataMessage, timer: Inst
             message_id,
         }) => {
             let res = match client.config.protocol {
-                Protocol::Binary => client.one_off_query_bsatn(&query, &message_id, timer),
-                Protocol::Text => client.one_off_query_json(&query, &message_id, timer),
+                Protocol::Binary => client.one_off_query_bsatn(&query, &message_id, timer).await,
+                Protocol::Text => client.one_off_query_json(&query, &message_id, timer).await,
             };
             mod_metrics
                 .request_round_trip_sql

crates/core/src/client/message_handlers.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/datastore/system_tables.rs

@@ -155,6 +155,36 @@ metrics_group!(
         #[help = "Number of subscriptions via the legacy api"]
         #[labels(database_identity: Identity)]
         pub num_legacy_subscriptions: IntGaugeVec,
+
+        #[name = spacetime_subscription_compile_time_sec]
+        #[help = "How much time (in seconds) do we spend compiling subscriptions"]
+        #[labels(db: Identity, workload: WorkloadType)]
+        pub subscription_compile_time: HistogramVec,
+
+        #[name = spacetime_subscription_lock_num_waiters]
+        #[help = "The number of clients waiting to acquire the subscription lock"]
+        #[labels(db: Identity, workload: WorkloadType)]
+        pub subscription_lock_waiters: IntGaugeVec,
+
+        #[name = spacetime_subscription_lock_wait_time_sec]
+        #[help = "How much time (in seconds) do we spend waiting to acquire the subscription lock"]
+        #[labels(db: Identity, workload: WorkloadType)]
+        pub subscription_lock_wait_time: HistogramVec,
+
+        #[name = spacetime_num_queries_subscribed]
+        #[help = "How many total queries make up each subscribe call"]
+        #[labels(db: Identity)]
+        pub num_queries_subscribed: IntCounterVec,
+
+        #[name = spacetime_num_new_queries_subscribed]
+        #[help = "How many new (uncached) queries are make up each subscribe call"]
+        #[labels(db: Identity)]
+        pub num_new_queries_subscribed: IntCounterVec,
+
+        #[name = spacetime_num_queries_evaluated]
+        #[help = "How many queries are evaluated in each subscribe and unsubscribe"]
+        #[labels(db: Identity, workload: WorkloadType)]
+        pub num_queries_evaluated: IntCounterVec,
     }
 );