diff --git a/Cargo.lock b/Cargo.lock index 72a4c4b21b0..24f385f39c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3646,6 +3646,13 @@ version = "2.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "25a2bc672d1148e28034f176e01fffebb08b35768468cc954630da77a1449005" +[[package]] +name = "idc-auth-test" +version = "0.0.0" +dependencies = [ + "spacetimedb", +] + [[package]] name = "ident_case" version = "1.0.1" @@ -9146,6 +9153,7 @@ name = "spacetimedb-testing" version = "2.13.0" dependencies = [ "anyhow", + "axum", "bytes", "clap 4.5.50", "duct", @@ -9169,6 +9177,7 @@ dependencies = [ "spacetimedb-standalone", "tempfile", "tokio", + "tower", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index ee78bb6bf57..50e137f0631 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,6 +51,7 @@ members = [ "modules/perf-test", "modules/module-test", "modules/environment-test", + "modules/idc-auth-test", "templates/basic-rs/spacetimedb", "templates/chat-console-rs/spacetimedb", "modules/sdk-test", diff --git a/crates/auth/src/idc.rs b/crates/auth/src/idc.rs new file mode 100644 index 00000000000..ba0755f0a5a --- /dev/null +++ b/crates/auth/src/idc.rs @@ -0,0 +1,70 @@ +//! Host-signed transport credentials. These are never exposed as module JWTs. + +use crate::token_signing::TokenSigner; +use jsonwebtoken::{Algorithm, DecodingKey, Validation}; +use serde::{Deserialize, Serialize}; +use spacetimedb_lib::Identity; +use std::time::{SystemTime, UNIX_EPOCH}; + +const ISSUER: &str = "spacetimedb-idc-v1"; +// Keep leaked credentials short-lived. The actor signs a fresh token for each +// delivery attempt, including retries, so queued messages do not need long-lived tokens. +// TODO(idc): Consider caching tokens per sender/receiver pair and extending their lifetime +// to reduce signing overhead, refreshing them before expiry. +const TOKEN_LIFETIME_SECS: u64 = 60; + +/// Signs transport credentials for calls between databases. +pub trait IdcTokenSigner: Send + Sync { + /// Attest the sending database's identity to the receiver. + fn sign(&self, sender: Identity, receiver: Identity) -> anyhow::Result; +} + +impl IdcTokenSigner for T { + fn sign(&self, sender: Identity, receiver: Identity) -> anyhow::Result { + sign(self, sender, receiver) + } +} + +/// We can't use [`crate::identity::IncomingClaims`] here, client auth derives +/// identity from `(iss, sub)`, and we don't store that pair for databases. +/// Using the database identity as a new subject would give us a different identity. +/// Instead, a trusted host signs the database's existing identity directly. +/// The token is only used between hosts, not exposed through `sender_auth().jwt()`. +#[derive(Serialize, Deserialize)] +struct Claims { + iss: String, + aud: String, + sender: Identity, + iat: u64, + exp: u64, +} + +/// Mint credentials from trusted host state, not a caller-supplied identity. +pub fn sign(signer: &impl TokenSigner, sender: Identity, receiver: Identity) -> anyhow::Result { + let iat = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs(); + let claims = Claims { + iss: ISSUER.into(), + aud: receiver.to_hex().to_string(), + sender, + iat, + exp: iat + TOKEN_LIFETIME_SECS, + }; + Ok(signer.sign(&claims)?) +} + +/// Authenticate only with an explicitly trusted host key, never OIDC discovery. +pub fn verify(key: &DecodingKey, token: &str, receiver: Identity) -> anyhow::Result { + let mut validation = Validation::new(Algorithm::ES256); + validation.set_required_spec_claims(&["iss", "aud", "exp", "iat"]); + validation.set_issuer(&[ISSUER]); + validation.set_audience(&[receiver.to_hex().to_string()]); + validation.leeway = 0; + let claims = jsonwebtoken::decode::(token, key, &validation)?.claims; + let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs(); + anyhow::ensure!(claims.iat <= now && claims.exp > now, "invalid IDC token timestamps"); + anyhow::ensure!( + claims.exp > claims.iat && claims.exp - claims.iat <= TOKEN_LIFETIME_SECS, + "invalid IDC token lifetime" + ); + Ok(claims.sender) +} diff --git a/crates/auth/src/identity.rs b/crates/auth/src/identity.rs index 41b38d00cb7..4cf47e16c42 100644 --- a/crates/auth/src/identity.rs +++ b/crates/auth/src/identity.rs @@ -9,8 +9,9 @@ use std::time::SystemTime; #[derive(Debug, Clone)] pub struct ConnectionAuthCtx { - pub claims: SpacetimeIdentityClaims, - pub jwt_payload: Box, + pub identity: Identity, + /// Client JWT claims, absent for host-authenticated database calls. + pub jwt_payload: Option>, } impl TryFrom for ConnectionAuthCtx { @@ -18,8 +19,8 @@ impl TryFrom for ConnectionAuthCtx { fn try_from(claims: SpacetimeIdentityClaims) -> Result { let payload = serde_json::to_string(&claims).map_err(|e| anyhow::anyhow!("Failed to serialize claims: {e}"))?; Ok(ConnectionAuthCtx { - claims, - jwt_payload: payload.into(), + identity: claims.identity, + jwt_payload: Some(payload.into()), }) } } diff --git a/crates/auth/src/lib.rs b/crates/auth/src/lib.rs index db53a0c9064..08ef6de877e 100644 --- a/crates/auth/src/lib.rs +++ b/crates/auth/src/lib.rs @@ -1 +1,3 @@ +pub mod idc; pub mod identity; +pub mod token_signing; diff --git a/crates/auth/src/token_signing.rs b/crates/auth/src/token_signing.rs new file mode 100644 index 00000000000..24ce6383be2 --- /dev/null +++ b/crates/auth/src/token_signing.rs @@ -0,0 +1,14 @@ +use jsonwebtoken::{errors::Error as JwtError, Algorithm, EncodingKey, Header}; +use serde::Serialize; + +/// Signs tokens without validating their claims. +pub trait TokenSigner: Sync + Send { + /// Serialize the claims and sign a JWT with them as the payload. + fn sign(&self, claims: &T) -> Result; +} + +impl TokenSigner for EncodingKey { + fn sign(&self, claims: &T) -> Result { + jsonwebtoken::encode(&Header::new(Algorithm::ES256), claims, self) + } +} diff --git a/crates/client-api-messages/src/idc.rs b/crates/client-api-messages/src/idc.rs new file mode 100644 index 00000000000..98e0a8d53bf --- /dev/null +++ b/crates/client-api-messages/src/idc.rs @@ -0,0 +1,124 @@ +//! HTTP responses shared by the IDC receiver and sender. +//! +//! Only recorded outcomes finish a delivery. Authentication, routing, and other +//! failures leave it pending; their response bodies are diagnostics, not outcomes. + +use bytes::Bytes; +use http::StatusCode; +use std::fmt; + +/// A receiver result or an HTTP error that leaves delivery pending. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum IdcResponse { + /// Empty for a newly committed call; replays may contain BSATN return bytes. + Success(Bytes), + /// A reducer or validation failure stored by the receiver. Retrying the same + /// message returns the same error. Hence, should be considered terminal. + /// The payload is UTF-8 error text, not BSATN. + TerminalError(Bytes), + /// A recorded energy failure, with a UTF-8 description, not BSATN. + BudgetExceeded(Bytes), + /// The sender already acknowledged this sequence. There is no result to replay. + AlreadyAcked, + + /// A delivery failure the sender should retry. It may resolve on a later attempt. + DeliveryError(IdcDeliveryError), +} + +impl IdcResponse { + pub fn into_http_parts(self) -> (StatusCode, Bytes) { + match self { + Self::Success(body) => (StatusCode::OK, body), + Self::TerminalError(body) => (StatusCode::UNPROCESSABLE_ENTITY, body), + Self::BudgetExceeded(body) => (StatusCode::PAYMENT_REQUIRED, body), + Self::AlreadyAcked => (StatusCode::ALREADY_REPORTED, Bytes::new()), + Self::DeliveryError(error) => error.into_http_parts(), + } + } + + /// Decode a complete response body. Anything outside the IDC outcome contract + /// is a delivery failure, including authentication failures and redirects. + pub fn from_http_parts(status: StatusCode, body: Bytes) -> Self { + match status { + StatusCode::OK => Self::Success(body), + StatusCode::UNPROCESSABLE_ENTITY => Self::TerminalError(body), + StatusCode::PAYMENT_REQUIRED => Self::BudgetExceeded(body), + StatusCode::ALREADY_REPORTED => Self::AlreadyAcked, + _ => Self::DeliveryError(IdcDeliveryError { status, body }), + } + } +} + +/// An HTTP response without a usable IDC outcome. Retry without acknowledging +/// the sequence or calling `on_result`; the body must not control retry policy. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub struct IdcDeliveryError { + pub status: StatusCode, + /// Uninterpreted HTTP body, displayed as lossy UTF-8 for diagnostics only. + pub body: Bytes, +} + +impl IdcDeliveryError { + pub fn into_http_parts(self) -> (StatusCode, Bytes) { + (self.status, self.body) + } +} + +impl fmt::Display for IdcDeliveryError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "HTTP {}: {}", self.status, String::from_utf8_lossy(&self.body)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn outcomes_round_trip_without_changing_payloads() { + for (response, status) in [ + (IdcResponse::Success(Bytes::new()), StatusCode::OK), + (IdcResponse::Success(Bytes::from_static(b"\x00\xff")), StatusCode::OK), + ( + IdcResponse::TerminalError(Bytes::from_static(b"unauthorized sender")), + StatusCode::UNPROCESSABLE_ENTITY, + ), + ( + IdcResponse::BudgetExceeded(Bytes::from_static(b"out of energy")), + StatusCode::PAYMENT_REQUIRED, + ), + (IdcResponse::AlreadyAcked, StatusCode::ALREADY_REPORTED), + ] { + let (actual_status, body) = response.clone().into_http_parts(); + assert_eq!(actual_status, status); + assert_eq!(IdcResponse::from_http_parts(actual_status, body), response); + } + } + + #[test] + fn other_responses_leave_delivery_pending_regardless_of_body() { + for status in [ + 201, 204, 301, 307, 400, 401, 403, 404, 408, 415, 425, 429, 500, 502, 503, 504, 530, + ] { + for body in [ + Bytes::from_static(b"Ok"), + Bytes::from_static(b"permanent error"), + Bytes::new(), + ] { + let status = StatusCode::from_u16(status).unwrap(); + let response = IdcResponse::from_http_parts(status, body.clone()); + assert_eq!( + response, + IdcResponse::DeliveryError(IdcDeliveryError { + status, + body: body.clone() + }) + ); + let (encoded_status, encoded_body) = response.clone().into_http_parts(); + assert_eq!(encoded_status, status); + assert_eq!(encoded_body, body); + assert_eq!(IdcResponse::from_http_parts(encoded_status, encoded_body), response); + } + } + } +} diff --git a/crates/client-api-messages/src/lib.rs b/crates/client-api-messages/src/lib.rs index c174c310170..7e6fea824eb 100644 --- a/crates/client-api-messages/src/lib.rs +++ b/crates/client-api-messages/src/lib.rs @@ -2,6 +2,7 @@ pub mod energy; pub mod http; +pub mod idc; pub mod name; pub mod publish; pub mod websocket; diff --git a/crates/client-api/src/auth.rs b/crates/client-api/src/auth.rs index 729d001e2c3..3205b06b2ee 100644 --- a/crates/client-api/src/auth.rs +++ b/crates/client-api/src/auth.rs @@ -107,8 +107,8 @@ impl SpacetimeAuth { impl From for ConnectionAuthCtx { fn from(auth: SpacetimeAuth) -> Self { ConnectionAuthCtx { - claims: auth.claims, - jwt_payload: auth.jwt_payload.clone(), + identity: auth.claims.identity, + jwt_payload: Some(auth.jwt_payload), } } } @@ -259,8 +259,7 @@ impl JwtKeyAuthProvider { impl TokenSigner for JwtKeyAuthProvider { fn sign(&self, claims: &T) -> Result { - let header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256); - jsonwebtoken::encode(&header, &claims, &self.keys.private) + self.keys.sign(claims) } } @@ -562,3 +561,33 @@ pub async fn anon_auth_middleware( let resp = next.run(req).await; Ok((auth.into_headers(), resp)) } + +/// The sending database identity verified by the IDC authentication middleware. +#[derive(Clone, Copy)] +pub struct IdcCallerIdentity(pub Identity); + +/// IDC authenticates a database, but must not expose its transport token to module code. +pub async fn idc_auth_middleware( + State(worker_ctx): State, + mut req: Request, + next: Next, +) -> axum::response::Result { + use crate::routes::database::ResolvedDatabase; + let receiver = req + .extensions() + .get::() + .ok_or(http::StatusCode::INTERNAL_SERVER_ERROR)? + .0 + .database_identity; + let bearer = req + .headers() + .typed_try_get::>() + .map_err(|_| http::StatusCode::UNAUTHORIZED)? + .ok_or(http::StatusCode::UNAUTHORIZED)?; + let key = jsonwebtoken::DecodingKey::from_ec_pem(worker_ctx.jwt_auth_provider().public_key_bytes()) + .map_err(log_and_500)?; + let identity = + spacetimedb::auth::idc::verify(&key, bearer.token(), receiver).map_err(|_| http::StatusCode::UNAUTHORIZED)?; + req.extensions_mut().insert(IdcCallerIdentity(identity)); + Ok(next.run(req).await) +} diff --git a/crates/client-api/src/routes/database.rs b/crates/client-api/src/routes/database.rs index fe677648829..cdba2d7df8b 100644 --- a/crates/client-api/src/routes/database.rs +++ b/crates/client-api/src/routes/database.rs @@ -9,8 +9,8 @@ use std::time::Duration; use std::{env, io}; use crate::auth::{ - anon_auth_middleware, SpacetimeAuth, SpacetimeEnergyUsed, SpacetimeExecutionDurationMicros, SpacetimeIdentity, - SpacetimeIdentityToken, + anon_auth_middleware, IdcCallerIdentity, SpacetimeAuth, SpacetimeEnergyUsed, SpacetimeExecutionDurationMicros, + SpacetimeIdentity, SpacetimeIdentityToken, }; use crate::routes::subscribe::generate_random_connection_id; use crate::util::serde::humantime_duration; @@ -39,17 +39,19 @@ use spacetimedb::host::module_host::{ }; use spacetimedb::host::{CallResult, UpdateDatabaseResult}; use spacetimedb::host::{FunctionArgs, MigratePlanResult}; +use spacetimedb::host::{IdcDeliveryFailure, IdcReducerCallOutcome, ProcedureCallError, ReducerCallError}; use spacetimedb::host::{ModuleHost, ReducerCallResult, ReducerCallResultWithTxOffset, ReducerOutcome}; -use spacetimedb::host::{ProcedureCallError, ReducerCallError}; use spacetimedb::identity::Identity; use spacetimedb::messages::control_db::{Database, HostType}; use spacetimedb_client_api_messages::http::SqlStmtResult; +use spacetimedb_client_api_messages::idc::{IdcDeliveryError, IdcResponse}; use spacetimedb_client_api_messages::name::{ self, DatabaseName, DomainName, EnvironmentPublishError, EnvironmentVersionConflict, MigrationPolicy, PrePublishAutoMigrateResult, PrePublishManualMigrateResult, PrePublishResult, PrettyPrintStyle, PublishOp, PublishResult, }; use spacetimedb_datastore::db_metrics::DB_METRICS; +use spacetimedb_datastore::system_tables::StInboundMsgResultStatus; use spacetimedb_lib::db::raw_def::v10::RawModuleDefV10; use spacetimedb_lib::db::raw_def::v9::RawModuleDefV9; use spacetimedb_lib::{http as st_http, ConnectionId}; @@ -100,6 +102,28 @@ pub struct CallParams { reducer: String, } +/// Query parameters for an async IDC request to another database. +#[derive(Deserialize)] +pub struct CallFromDatabaseQuery { + /// Sender-assigned identifier for the (sender, receiver, outbox table) stream. + /// Reused for every delivery on that stream. + stream_id: u64, + /// One-based sequence number within this stream. + seq: u64, + /// Signature hash computed by [`spacetimedb_schema::def::reducer_signature_hash`], + /// encoded as [`Hash::to_hex()`] by the sender. + #[serde(deserialize_with = "deserialize_signature_hash")] + signature_hash: Hash, + /// Highest contiguous sequence whose result the sender has already acknowledged. + #[serde(default)] + ack_prefix: u64, +} + +fn deserialize_signature_hash<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result { + // Hash's default serde representation is a U256, not digest-byte-order hex. + Hash::from_hex(String::deserialize(deserializer)?).map_err(serde::de::Error::custom) +} + pub const NO_SUCH_DATABASE: (StatusCode, &str) = (StatusCode::NOT_FOUND, "No such database."); const MISDIRECTED: (StatusCode, &str) = (StatusCode::NOT_FOUND, "Database is not scheduled on this host"); @@ -277,6 +301,109 @@ async fn confirm_call_offset( } Ok(()) } + +/// Call a reducer on behalf of another database. +/// +/// This is the host-side receiver transport for async IDC. It accepts delivery +/// metadata and waits for reducer durability before acknowledging the sender. +pub async fn call_from_database( + State(worker_ctx): State, + Extension(IdcCallerIdentity(caller_identity)): Extension, + Extension(ResolvedDatabase(database)): Extension, + Path(CallParams { reducer }): Path, + Query(query): Query, + TypedHeader(content_type): TypedHeader, + body: Bytes, +) -> axum::response::Result { + if content_type != headers::ContentType::octet_stream() { + return Err((StatusCode::UNSUPPORTED_MEDIA_TYPE, "Expected application/octet-stream").into()); + } + if query.seq == 0 { + return Err((StatusCode::BAD_REQUEST, "IDC seq must start at 1").into()); + } + + let owner_identity = database.owner_identity; + let module = find_database_module(&worker_ctx, &database).await?; + + // Run the normal connection lifecycle with the authenticated database identity, + // but do not expose the host's transport credentials as a module JWT. + let caller_auth = ConnectionAuthCtx { + identity: caller_identity, + jwt_payload: None, + }; + let fut = async move |module: ModuleHost, caller_identity: Identity, connection_id: ConnectionId| { + let durable_offset = module.durable_tx_offset(); + let result = module + .call_idc_reducer( + caller_identity, + connection_id, + &reducer, + FunctionArgs::Bsatn(body), + query.stream_id, + query.seq, + query.ack_prefix, + query.signature_hash, + ) + .await + .map_err(|e| map_idc_call_error(e, &reducer).into_http_parts())?; + + debug!( + "IDC delivery accepted: sender={}, receiver={}, stream_id={}, seq={}, ack_prefix={}, signature_hash={:?}", + caller_identity, owner_identity, query.stream_id, query.seq, query.ack_prefix, query.signature_hash, + ); + + idc_call_response(&owner_identity, &reducer, result, durable_offset).await + }; + + with_connection(module, caller_auth, caller_identity, fut).await +} + +// Return new and replayed results only after their transactions are durable. +async fn idc_call_response( + owner_identity: &Identity, + reducer: &str, + result: IdcReducerCallOutcome, + durable_offset: Option, +) -> axum::response::Result { + let (tx_offset, response) = match result { + IdcReducerCallOutcome::Recorded(result) => { + let tx_offset = if durable_offset.is_some() { + Some(result.tx_offset.await.map_err(|_| log_and_500("transaction aborted"))?) + } else { + None + }; + let (status, body) = + idc_reducer_outcome_response(owner_identity, reducer, result.result.outcome).into_http_parts(); + let response = ( + status, + TypedHeader(SpacetimeEnergyUsed(result.result.execution_budget_used)), + TypedHeader(SpacetimeExecutionDurationMicros(result.result.execution_duration)), + body, + ) + .into_response(); + (tx_offset, response) + } + IdcReducerCallOutcome::Replayed(outcome) => { + let (status, body) = + idc_stored_outcome_response(outcome.result_status, outcome.result_payload).into_http_parts(); + (Some(outcome.tx_offset), (status, body).into_response()) + } + IdcReducerCallOutcome::AlreadyAcked => { + // A delayed delivery or retry can arrive after the sender has acknowledged + // the outcome and we have removed it. The sender must ignore this response, + // not treat it as a new reducer result or invoke on_result again. + (None, IdcResponse::AlreadyAcked.into_http_parts().into_response()) + } + }; + + // A retry can observe a committed outcome before the original request's + // durability wait finishes. New and replayed outcomes need the same barrier. + if let (Some(mut durable_offset), Some(tx_offset)) = (durable_offset, tx_offset) { + durable_offset.wait_for(tx_offset).await.map_err(log_and_500)?; + } + Ok(response) +} + #[derive(Deserialize)] pub struct HttpRouteParams { path: String, @@ -486,6 +613,49 @@ fn reducer_outcome_response( } } +// call_idc_reducer has already recorded these results. The route waits for +// durability before sending them back to the sender. +fn idc_reducer_outcome_response(owner_identity: &Identity, reducer: &str, outcome: ReducerOutcome) -> IdcResponse { + match outcome { + ReducerOutcome::Committed => IdcResponse::Success(Bytes::new()), + ReducerOutcome::Failed(errmsg) => IdcResponse::TerminalError(Bytes::from((*errmsg).into_string())), + ReducerOutcome::BudgetExceeded => { + log::info!("Node's energy budget exceeded for identity: {owner_identity} while executing {reducer}"); + IdcResponse::BudgetExceeded(Bytes::from_static(b"Module energy budget exhausted.")) + } + } +} + +fn idc_stored_outcome_response(result_status: StInboundMsgResultStatus, result_payload: Bytes) -> IdcResponse { + match result_status { + StInboundMsgResultStatus::Ok => IdcResponse::Success(result_payload), + StInboundMsgResultStatus::Err => IdcResponse::TerminalError(result_payload), + StInboundMsgResultStatus::BudgetExceeded => IdcResponse::BudgetExceeded(result_payload), + } +} + +/// Builds a retryable HTTP response. Recorded terminal failures arrive as reducer outcomes. +fn map_idc_call_error(e: IdcDeliveryFailure, reducer: &str) -> IdcDeliveryError { + let status = match &e { + IdcDeliveryFailure::OutOfOrder { .. } => { + // The receiver needs an earlier sequence before it can process this one. + // TODO(idc): Lower the log level when pipelining makes this expected. + log::error!("Retryable IDC delivery error for {reducer}: {e:#}"); + StatusCode::TOO_EARLY + } + IdcDeliveryFailure::NoSuchModule(_) | IdcDeliveryFailure::Datastore(_) | IdcDeliveryFailure::Internal(_) => { + // No confirmed outcome is available. Leave the message pending for retry. + log::debug!("Retryable IDC delivery error for {reducer}: {e:#}"); + StatusCode::INTERNAL_SERVER_ERROR + } + }; + + IdcDeliveryError { + status, + body: Bytes::from(format!("{:#}", anyhow::anyhow!(e))), + } +} + pub(crate) fn client_connected_error_to_response(err: ClientConnectedError) -> ErrorResponse { match err { // If `call_identity_connected` returns `Err(Rejected)`, then the `client_connected` reducer errored, @@ -1713,6 +1883,8 @@ pub struct DatabaseRoutes { pub subscribe_get: MethodRouter, /// POST: /database/:name_or_identity/call/:reducer pub call_reducer_procedure_post: MethodRouter, + /// POST: /database/:name_or_identity/call-from-database/:reducer + pub call_from_database_post: MethodRouter, /// GET: /database/:name_or_identity/schema pub schema_get: MethodRouter, /// GET: /database/:name_or_identity/environment @@ -1762,6 +1934,7 @@ where identity_get: get(get_identity::), subscribe_get: get(handle_websocket::), call_reducer_procedure_post: post(call::), + call_from_database_post: post(call_from_database::), schema_get: get(schema::), environment_get: get(environment_metadata::), environment_put: put(environment_set::), @@ -1836,6 +2009,18 @@ where )), ); + // Authenticate IDC separately: there is no anonymous fallback and no module-visible JWT. + let idc_router = axum::Router::::new() + .route( + "/:name_or_identity/call-from-database/:reducer", + self.call_from_database_post, + ) + .route_layer(axum::middleware::from_fn_with_state( + ctx.clone(), + crate::auth::idc_auth_middleware::, + )) + .route_layer(resolving_egress_metrics_middleware.clone()); + let authed_named_router = axum::Router::new() .nest("/:name_or_identity", db_router) .route_layer(axum::middleware::from_fn_with_state(ctx, anon_auth_middleware::)); @@ -1854,6 +2039,7 @@ where axum::Router::new() .merge(authed_root_router) .merge(authed_named_router) + .merge(idc_router) .merge(http_route_router) } } @@ -1958,6 +2144,167 @@ mod tests { use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use tower::util::ServiceExt; + #[test] + fn idc_signature_hash_query_uses_digest_hex() { + let hash = spacetimedb_lib::hash_bytes(b"signature query test"); + let uri = format!("/call_from_database?stream_id=1&seq=1&signature_hash={hash}") + .parse() + .unwrap(); + let Query(query) = Query::::try_from_uri(&uri).unwrap(); + assert_eq!(query.signature_hash, hash); + assert_eq!(query.ack_prefix, 0); + + for suffix in [ + String::new(), + "&signature_hash=invalid".into(), + "&signature_hash=00".into(), + format!("&signature_hash={hash}00"), + ] { + let uri = format!("/call_from_database?stream_id=1&seq=1{suffix}") + .parse() + .unwrap(); + assert!(Query::::try_from_uri(&uri).is_err()); + } + } + + #[test] + fn idc_outcome_response_uses_terminal_status_codes() { + let receiver = Identity::ZERO; + for (outcome, stored_status, expected) in [ + ( + ReducerOutcome::Committed, + StInboundMsgResultStatus::Ok, + IdcResponse::Success(Bytes::new()), + ), + ( + ReducerOutcome::Failed(Box::new("rejected".into())), + StInboundMsgResultStatus::Err, + IdcResponse::TerminalError(Bytes::from_static(b"rejected")), + ), + ( + ReducerOutcome::BudgetExceeded, + StInboundMsgResultStatus::BudgetExceeded, + IdcResponse::BudgetExceeded(Bytes::from_static(b"Module energy budget exhausted.")), + ), + ] { + // Decode this the same way the sender does. + let response = idc_reducer_outcome_response(&receiver, "receive", outcome); + let (status, body) = response.into_http_parts(); + assert_eq!(IdcResponse::from_http_parts(status, body.clone()), expected); + let replay = idc_stored_outcome_response(stored_status, body); + assert_eq!(replay, expected); + } + } + + #[test] + fn idc_call_errors_leave_delivery_pending() { + use spacetimedb::host::module_host::NoSuchModule; + + for (error, expected_status) in [ + ( + IdcDeliveryFailure::NoSuchModule(NoSuchModule), + StatusCode::INTERNAL_SERVER_ERROR, + ), + ( + IdcDeliveryFailure::Datastore(spacetimedb::error::DBError::Other(anyhow::anyhow!("write failed"))), + StatusCode::INTERNAL_SERVER_ERROR, + ), + ( + IdcDeliveryFailure::OutOfOrder { expected: 1, actual: 2 }, + StatusCode::TOO_EARLY, + ), + ( + IdcDeliveryFailure::Internal("commit failed".into()), + StatusCode::INTERNAL_SERVER_ERROR, + ), + ] { + let expected_body = error.to_string(); + let response = map_idc_call_error(error, "receive"); + let (status, body) = response.clone().into_http_parts(); + assert_eq!(status, expected_status); + assert_eq!(body.as_ref(), expected_body.as_bytes()); + assert_eq!( + IdcResponse::from_http_parts(status, body), + IdcResponse::DeliveryError(response) + ); + } + } + + #[tokio::test] + async fn idc_responses_wait_until_the_result_is_durable() { + use spacetimedb::energy::FunctionBudget; + use spacetimedb::host::module_host::IdcStoredReducerOutcome; + use std::pin::pin; + use std::task::Poll; + use tokio::sync::watch; + + for (outcome, stored_status, expected) in [ + ( + ReducerOutcome::Committed, + StInboundMsgResultStatus::Ok, + IdcResponse::Success(Bytes::new()), + ), + ( + ReducerOutcome::Failed(Box::new("rejected".into())), + StInboundMsgResultStatus::Err, + IdcResponse::TerminalError(Bytes::from_static(b"rejected")), + ), + ( + ReducerOutcome::BudgetExceeded, + StInboundMsgResultStatus::BudgetExceeded, + IdcResponse::BudgetExceeded(Bytes::from_static(b"Module energy budget exhausted.")), + ), + ] { + let (durable, rx) = watch::channel(None); + let (committed, tx_offset) = oneshot::channel(); + let receiver = Identity::ZERO; + let mut recorded = pin!(idc_call_response( + &receiver, + "receive", + IdcReducerCallOutcome::Recorded(ReducerCallResultWithTxOffset { + result: ReducerCallResult { + outcome, + execution_budget_used: FunctionBudget::ZERO, + execution_duration: Duration::ZERO, + }, + tx_offset, + }), + Some(rx.clone().into()), + )); + assert!(futures::poll!(&mut recorded).is_pending()); + + // Commit makes the result readable by a retry, but neither request may reply yet. + committed.send(7).unwrap(); + let (_, payload) = expected.clone().into_http_parts(); + let mut replayed = pin!(idc_call_response( + &receiver, + "receive", + IdcReducerCallOutcome::Replayed(IdcStoredReducerOutcome { + result_status: stored_status, + result_payload: payload, + tx_offset: 7, + }), + Some(rx.into()), + )); + assert!(futures::poll!(&mut recorded).is_pending()); + assert!(futures::poll!(&mut replayed).is_pending()); + + durable.send_replace(Some(6)); + assert!(futures::poll!(&mut recorded).is_pending()); + assert!(futures::poll!(&mut replayed).is_pending()); + + durable.send_replace(Some(7)); + for response in [futures::poll!(recorded), futures::poll!(replayed)] { + let Poll::Ready(Ok(response)) = response else { + panic!("IDC response must be ready once its transaction is durable"); + }; + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), 1024).await.unwrap(); + assert_eq!(IdcResponse::from_http_parts(status, body), expected); + } + } + } + #[tokio::test] async fn environment_update_reports_success_only_after_confirmation() { let performed = |tx_offset| { @@ -2113,6 +2460,7 @@ mod tests { #[derive(Clone)] struct DummyJwtProvider { validator: DummyValidator, + keys: Option>, } impl TokenSigner for DummyJwtProvider { @@ -2139,7 +2487,7 @@ mod tests { } fn public_key_bytes(&self) -> &[u8] { - b"" + self.keys.as_ref().map_or(b"", |keys| keys.public_pem.as_ref()) } } @@ -2158,6 +2506,7 @@ mod tests { Self { jwt: DummyJwtProvider { validator: DummyValidator, + keys: None, }, client_actor_index: std::sync::Arc::new(ClientActorIndex::new()), module_logs_dir: ModuleLogsDir::from_path_unchecked(std::env::temp_dir()), @@ -2442,6 +2791,67 @@ mod tests { } } + #[tokio::test] + async fn idc_route_requires_host_credentials_and_exposes_no_jwt() -> anyhow::Result<()> { + let receiver = test_identity(201); + let sender = test_identity(202); + let keys = std::sync::Arc::new(spacetimedb::auth::JwtKeys::generate()?); + let mut state = DummyState::new().with_database(receiver); + state.jwt.keys = Some(keys.clone()); + let app = DatabaseRoutes { + call_from_database_post: axum::routing::post( + |Extension(IdcCallerIdentity(sender)): Extension, + auth: Option>| async move { + assert!(auth.is_none()); + sender.to_hex().to_string() + }, + ), + ..DatabaseRoutes::::default() + } + .into_router(state.clone()) + .with_state(state); + let valid = spacetimedb::auth::idc::sign(&keys.private, sender, receiver)?; + let wrong_receiver = spacetimedb::auth::idc::sign(&keys.private, sender, sender)?; + let wrong_key = + spacetimedb::auth::idc::sign(&spacetimedb::auth::JwtKeys::generate()?.private, sender, receiver)?; + let (_, client_token) = + crate::auth::TokenClaims::new("localhost".into(), "client".into()).encode_and_sign(&keys.private)?; + let call_idc_route = async |token: Option<&str>| { + let app = app.clone(); + let mut request = Request::builder() + .method("POST") + .uri(format!("/{}/call-from-database/receive", receiver.to_hex())); + if let Some(token) = token { + request = request.header(http::header::AUTHORIZATION, format!("Bearer {token}")); + } + Ok::<_, anyhow::Error>(app.oneshot(request.body(Body::empty())?).await?) + }; + let assert_unauthorized = async |token: Option<&str>| { + let response = call_idc_route(token).await?; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + Ok::<_, anyhow::Error>(()) + }; + + assert_unauthorized(None).await?; + assert_unauthorized(Some("malformed")).await?; + assert_unauthorized(Some(wrong_receiver.as_str())).await?; + assert_unauthorized(Some(wrong_key.as_str())).await?; + assert_unauthorized(Some(client_token.as_str())).await?; + + { + let token = valid.as_str(); + let mut request = Request::builder() + .method("POST") + .uri(format!("/{}/call-from-database/receive", receiver.to_hex())); + request = request.header(http::header::AUTHORIZATION, format!("Bearer {token}")); + let response = app.clone().oneshot(request.body(Body::empty())?).await?; + assert_eq!(response.status(), StatusCode::OK); + let body = response.into_body().collect().await?.to_bytes(); + assert_eq!(body.as_ref(), sender.to_hex().as_bytes()); + } + Ok(()) + } + /// Tests that requests to user-defined routes under `/database/:name-or-identity/routes` /// bypass the usual SpacetimeDB auth middleware, /// and accept requests with `Authorization` headers that SpacetimeDB would treat as malformed. diff --git a/crates/core/src/auth/mod.rs b/crates/core/src/auth/mod.rs index e1e38a667c4..ceedb779a6b 100644 --- a/crates/core/src/auth/mod.rs +++ b/crates/core/src/auth/mod.rs @@ -6,9 +6,83 @@ use spacetimedb_paths::cli::{PrivKeyPath, PubKeyPath}; use crate::config::CertificateAuthority; -pub use spacetimedb_auth::identity; +pub use spacetimedb_auth::{idc, identity}; pub mod token_validation; +#[cfg(test)] +mod idc_tests { + use super::*; + use serde_json::json; + use spacetimedb_lib::Identity; + use std::time::{SystemTime, UNIX_EPOCH}; + + #[test] + fn idc_credentials_require_trusted_key_recipient_and_expiry() -> anyhow::Result<()> { + let keys = JwtKeys::generate()?; + let sender = Identity::ZERO; + let receiver = Identity::ONE; + let signer: &dyn idc::IdcTokenSigner = &keys.private; + let token = signer.sign(sender, receiver)?; + assert_eq!(idc::verify(&keys.public, &token, receiver)?, sender); + assert!(idc::verify(&keys.public, &token, sender).is_err()); + assert!(idc::verify(&JwtKeys::generate()?.public, &token, receiver).is_err()); + + let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs(); + let claims = json!({ + "iss": "spacetimedb-idc-v1", "aud": receiver.to_hex().to_string(), + "sender": sender, "iat": now, "exp": now + 60, + }); + for (field, value) in [ + ("iss", json!("client-issuer")), + ("exp", json!(now - 1)), + ("exp", json!(now + 3600)), + ("iat", json!(now + 30)), + ("sender", json!("invalid")), + ] { + let mut invalid = claims.clone(); + invalid[field] = value; + let token = jsonwebtoken::encode( + &jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256), + &invalid, + &keys.private, + )?; + assert!( + idc::verify(&keys.public, &token, receiver).is_err(), + "accepted invalid {field}" + ); + } + for field in ["iss", "aud", "sender", "iat", "exp"] { + let mut invalid = claims.clone(); + invalid.as_object_mut().unwrap().remove(field); + let token = jsonwebtoken::encode( + &jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256), + &invalid, + &keys.private, + )?; + assert!( + idc::verify(&keys.public, &token, receiver).is_err(), + "accepted missing {field}" + ); + } + let forged = jsonwebtoken::encode( + &jsonwebtoken::Header::new(jsonwebtoken::Algorithm::HS256), + &claims, + &EncodingKey::from_secret(&keys.public_pem), + )?; + assert!(idc::verify(&keys.public, &forged, receiver).is_err()); + Ok(()) + } + + #[tokio::test] + async fn idc_credentials_are_not_client_tokens() -> anyhow::Result<()> { + use token_validation::TokenValidator; + let keys = JwtKeys::generate()?; + let token = idc::sign(&keys.private, Identity::ZERO, Identity::ONE)?; + assert!(keys.public.validate_token(&token).await.is_err()); + Ok(()) + } +} + /// JWT verification and signing keys. #[derive(Clone)] pub struct JwtKeys { diff --git a/crates/core/src/auth/token_validation.rs b/crates/core/src/auth/token_validation.rs index 1d12d3146b4..e52bf660403 100644 --- a/crates/core/src/auth/token_validation.rs +++ b/crates/core/src/auth/token_validation.rs @@ -11,6 +11,7 @@ use jsonwebtoken::{decode, AlgorithmFamily, Validation}; pub use jsonwebtoken::{DecodingKey, EncodingKey}; use lazy_static::lazy_static; use serde::{Deserialize, Serialize}; +pub use spacetimedb_auth::token_signing::TokenSigner; use std::sync::Arc; use std::time::{Duration, SystemTime}; use thiserror; @@ -37,19 +38,6 @@ pub enum TokenValidationError { Other(#[from] anyhow::Error), } -// A token signer is responsible for signing tokens without doing any validation. -pub trait TokenSigner: Sync + Send { - // Serialize the given claims and sign a JWT token with them as the payload. - fn sign(&self, claims: &T) -> Result; -} - -impl TokenSigner for EncodingKey { - fn sign(&self, claims: &Token) -> Result { - let header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256); - jsonwebtoken::encode(&header, claims, self) - } -} - impl TokenSigner for JwtKeys { fn sign(&self, claims: &Token) -> Result { self.private.sign(claims) diff --git a/crates/core/src/host/mod.rs b/crates/core/src/host/mod.rs index d0a1f23c560..c3eb7936d47 100644 --- a/crates/core/src/host/mod.rs +++ b/crates/core/src/host/mod.rs @@ -30,7 +30,8 @@ pub use host_controller::{ ProcedureCallResult, ProgramStorage, ReducerCallResult, ReducerCallResultWithTxOffset, ReducerOutcome, }; pub use module_host::{ - InitDatabaseResult, ModuleHost, NoSuchModule, ProcedureCallError, ReducerCallError, UpdateDatabaseResult, + IdcDeliveryFailure, IdcReducerCallOutcome, InitDatabaseResult, ModuleHost, NoSuchModule, ProcedureCallError, + ReducerCallError, UpdateDatabaseResult, }; pub use scheduler::Scheduler; diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs index f59fedeb6e5..e39e0de5e26 100644 --- a/crates/core/src/host/module_host.rs +++ b/crates/core/src/host/module_host.rs @@ -13,6 +13,7 @@ use crate::hash::Hash; use crate::host::host_controller::{CallProcedureReturn, ProcedureCallResult}; use crate::host::scheduler::{CallScheduledFunctionError, CallScheduledFunctionResult, ScheduledFunctionParams}; use crate::host::v8::{JsFatalHook, JsMainInstance, JsProcedureCallCompletion, JsProcedureInstance}; +use crate::host::wasm_common::module_host_actor::ReducerSuccessAction; pub use crate::host::wasm_common::module_host_actor::{InstanceCommon, WasmInstance}; use crate::host::wasmtime::ModuleInstance; use crate::host::{InvalidFunctionArguments, InvalidViewArguments}; @@ -49,6 +50,7 @@ use spacetimedb_data_structures::map::{HashCollectionExt as _, HashSet}; use spacetimedb_datastore::error::DatastoreError; use spacetimedb_datastore::execution_context::{Workload, WorkloadType}; use spacetimedb_datastore::locking_tx_datastore::{MutTxId, ViewCallInfo, ViewInstanceArgs}; +use spacetimedb_datastore::system_tables::{StInboundMsgResultStatus, StInboundMsgRow}; use spacetimedb_datastore::traits::{IsolationLevel, Program, TxData}; pub use spacetimedb_durability::{DurabilityExited, DurableOffset}; use spacetimedb_engine::sql::rls::RowLevelExpr; @@ -66,15 +68,16 @@ use spacetimedb_query::compile_subscription; use spacetimedb_sats::raw_identifier::RawIdentifier; use spacetimedb_sats::{AlgebraicType, AlgebraicTypeRef, ProductValue, Typespace}; use spacetimedb_schema::auto_migrate::{AutoMigrateError, MigrationPolicy}; -use spacetimedb_schema::def::{ModuleDef, ProcedureDef, ReducerDef, ViewDef}; +use spacetimedb_schema::def::{reducer_signature_hash, ModuleDef, ProcedureDef, ReducerDef, ViewDef}; use spacetimedb_schema::identifier::{Identifier, NamespacedIdentifier}; use spacetimedb_schema::reducer_name::ReducerName; use spacetimedb_schema::table_name::TableName; use std::collections::VecDeque; use std::fmt; use std::num::NonZeroUsize; +use std::ops::ControlFlow; use std::sync::atomic::AtomicBool; -use std::sync::{Arc, Weak}; +use std::sync::{Arc, OnceLock, Weak}; use std::time::{Duration, Instant}; use tokio::sync::{oneshot, OwnedSemaphorePermit, Semaphore}; @@ -227,6 +230,9 @@ pub struct ModuleInfo { /// Loaded by loading the module's program from the system tables, extracting its definition, /// and validating. pub module_def: Arc, + /// Signature hashes computed on first IDC use, keyed by fully qualified reducer name. + /// A new module gets a fresh cache on update or restart. + reducer_signature_hashes: FxHashMap>, /// The identity of the module. pub owner_identity: Identity, /// The identity of the database. @@ -343,8 +349,14 @@ impl ModuleInfo { subscriptions: ModuleSubscriptions, ) -> Arc { let metrics = ModuleMetrics::new(&database_identity); + let reducer_signature_hashes = module_def + .reducer_ids_and_defs() + .into_iter() + .map(|(_, reducer)| (reducer.name.clone(), OnceLock::new())) + .collect(); Arc::new(ModuleInfo { module_def: Arc::new(module_def), + reducer_signature_hashes, owner_identity, database_identity, module_hash, @@ -356,6 +368,22 @@ impl ModuleInfo { pub fn relational_db(&self) -> &Arc { self.subscriptions.relational_db() } + + /// Returns the reducer's signature hash, computing it on first use. + pub fn reducer_signature_hash(&self, reducer_name: &ReducerName) -> Hash { + *self.reducer_signature_hashes[reducer_name].get_or_init(|| { + let (_, reducer, owning_def) = self + .module_def + .reducer_by_name_with_module(reducer_name) + .expect("reducer belongs to this module"); + reducer_signature_hash( + &reducer.params(), + &reducer.ok_return_type, + &reducer.err_return_type, + owning_def.typespace(), + ) + }) + } } /// A bidirectional map between `Identifiers` (reducer names) and `ReducerId`s. @@ -741,7 +769,7 @@ pub fn call_identity_connected( let stdb = module.relational_db(); let workload = Workload::reducer_no_args( ReducerName::new(Identifier::new_unsafe_assume_valid("call_identity_connected".into())), - caller_auth.claims.identity, + caller_auth.identity, caller_connection_id, ); let mut_tx = stdb.begin_mut_tx(IsolationLevel::Serializable, workload); @@ -755,9 +783,9 @@ pub fn call_identity_connected( mut_tx .insert_st_client( - caller_auth.claims.identity, + caller_auth.identity, caller_connection_id, - &caller_auth.jwt_payload, + caller_auth.jwt_payload.as_deref(), ) .map_err(DBError::from) .map_err(Box::new)?; @@ -770,7 +798,7 @@ pub fn call_identity_connected( let tx = Some(ScopeGuard::into_inner(mut_tx)); let params = ModuleHost::call_reducer_params( &module.module_def, - caller_auth.claims.identity, + caller_auth.identity, Some(caller_connection_id), None, None, @@ -1644,6 +1672,74 @@ impl From for UpdateDatabaseResult { #[error("no such module")] pub struct NoSuchModule; +/// Result of processing an IDC delivery, including retransmissions. +/// New and replayed results must be durable before the transport replies. +pub enum IdcReducerCallOutcome { + /// Recorded a new terminal result, either from reducer execution or validation failure. + /// The transaction offset identifies the commit that must become durable. + Recorded(ReducerCallResultWithTxOffset), + /// Returned the stored result of an earlier delivery without invoking the reducer again. + Replayed(IdcStoredReducerOutcome), + /// The sender already acknowledged this delivery, so its stored result was removed. + /// The reducer is not invoked again. + AlreadyAcked, +} + +#[derive(Debug)] +pub struct IdcStoredReducerOutcome { + pub result_status: StInboundMsgResultStatus, + pub result_payload: Bytes, + /// Committed prefix containing the stored outcome; the transport waits for durability. + pub tx_offset: u64, +} + +/// The receiver's state for an ordered delivery. +#[derive(Debug)] +pub(super) enum IdcSequenceState { + /// This is the next sequence. Execute or record its failure in the checking transaction. + Ready, + /// This message was already processed. Return its stored outcome without running it again. + Replayed(StInboundMsgRow), + /// The sender acknowledged this message, so its stored outcome has been removed. + AlreadyAcked, +} + +/// The sender, stream, and sequence of an IDC request. +#[derive(Clone, Copy)] +pub(super) struct IdcMessageParams { + pub sender_identity: Identity, + pub stream_id: u64, + pub seq: u64, +} + +/// A validation failure that must be recorded before replying to the sender. +#[derive(Debug, thiserror::Error)] +enum IdcTerminalError { + #[error(transparent)] + InvalidArgs(#[from] InvalidReducerArguments), + #[error("IDC reducer does not exist: {0}")] + NoSuchReducer(String), + #[error("IDC reducer signature mismatch for {reducer}: expected {expected}, got {actual}")] + SignatureMismatch { + reducer: String, + expected: Hash, + actual: Hash, + }, +} + +/// A receiver failure that leaves the message pending for retry. +#[derive(Debug, thiserror::Error)] +pub enum IdcDeliveryFailure { + #[error(transparent)] + NoSuchModule(#[from] NoSuchModule), + #[error("IDC delivery is out of order: expected seq {expected}, got {actual}")] + OutOfOrder { expected: u64, actual: u64 }, + #[error(transparent)] + Datastore(#[from] DBError), + #[error("internal IDC reducer call error: {0}")] + Internal(String), +} + #[derive(thiserror::Error, Debug)] pub enum ReducerCallError { #[error(transparent)] @@ -2426,6 +2522,50 @@ impl ModuleHost { )) } + /// Validate the requested reducer against module def, + /// prepare its arguments by calling `call_reducer_params`. + /// Failures must be considered as terminal outcomes. + fn call_idc_reducer_param( + &self, + caller_identity: Identity, + caller_connection_id: ConnectionId, + reducer_name: &str, + args: FunctionArgs, + expected_signature_hash: Hash, + ) -> Result<(&ReducerDef, CallReducerParams), IdcTerminalError> { + let Some((reducer_id, reducer_def, owning_def)) = + self.info.module_def.reducer_by_name_with_module(reducer_name) + else { + return Err(IdcTerminalError::NoSuchReducer(reducer_name.to_string())); + }; + + if reducer_def.lifecycle.is_some() || reducer_def.visibility.is_private() { + return Err(IdcTerminalError::NoSuchReducer(reducer_name.to_string())); + } + + let actual_signature_hash = self.info.reducer_signature_hash(&reducer_def.name); + if actual_signature_hash != expected_signature_hash { + return Err(IdcTerminalError::SignatureMismatch { + reducer: reducer_name.to_string(), + expected: expected_signature_hash, + actual: actual_signature_hash, + }); + } + + let params = Self::call_reducer_params( + owning_def, + caller_identity, + Some(caller_connection_id), + None, + None, + None, + reducer_id, + reducer_def, + args, + )?; + Ok((reducer_def, params)) + } + async fn call_reducer_with_params( &self, reducer_name: &ReducerName, @@ -2452,6 +2592,198 @@ impl ModuleHost { .map_err(Into::into) } + /// Check an inbound IDC delivery for the ordered configuration + /// against the receiver's stream state without mutating it. + /// + /// Algorithm: + /// + /// 1. Read `applied_prefix`, the highest contiguous `seq` whose terminal + /// outcome has already been decided for `(sender_identity, stream_id)`. + /// 2. If `seq <= applied_prefix`, this is a duplicate or delayed delivery. + /// Return the retained outcome when it still exists; otherwise the sender + /// has already acknowledged it and the delivery is `AlreadyAcked`. + /// 3. If `seq == applied_prefix + 1`, the delivery is the next message in + /// the stream and may run the reducer. + /// 4. Otherwise, an earlier message is still missing, so report the expected + /// sequence and make the sender retry later. + pub(super) fn check_idc_stream_seq( + tx: &MutTxId, + delivery: IdcMessageParams, + ) -> Result { + let IdcMessageParams { + sender_identity, + stream_id, + seq, + } = delivery; + let applied_prefix = tx + .inbound_idc_applied_prefix(sender_identity, stream_id) + .map_err(DBError::from)?; + + // This sequence was already processed. Replay its result if retained; + // otherwise it was acknowledged. Neither case executes the reducer again. + if seq <= applied_prefix { + let state = match tx + .inbound_idc_outcome(sender_identity, stream_id, seq) + .map_err(DBError::from)? + { + Some(row) => IdcSequenceState::Replayed(row), + None => IdcSequenceState::AlreadyAcked, + }; + return Ok(state); + } + + // Only the next sequence may execute; later deliveries must wait for + // the missing messages to be processed first. + let expected = applied_prefix + 1; + if seq == expected { + return Ok(IdcSequenceState::Ready); + } + Err(IdcDeliveryFailure::OutOfOrder { expected, actual: seq }) + } + + /// Keep the checking transaction open for a new delivery, or roll it back and + /// return the existing outcome. Delivery failures also roll back the transaction. + pub(super) fn check_idc_delivery( + db: &RelationalDB, + tx: MutTxId, + delivery: IdcMessageParams, + ) -> Result, IdcDeliveryFailure> { + let (tx, state) = db.with_auto_rollback(tx, |tx| Self::check_idc_stream_seq(tx, delivery))?; + let outcome = match state { + IdcSequenceState::Ready => return Ok(ControlFlow::Continue(tx)), + IdcSequenceState::Replayed(row) => { + let tx_offset = ModuleSubscriptions::rollback_mut_tx(db, tx); + IdcReducerCallOutcome::Replayed(IdcStoredReducerOutcome { + result_status: row.result_status, + result_payload: row.result_payload, + tx_offset, + }) + } + IdcSequenceState::AlreadyAcked => { + ModuleSubscriptions::rollback_mut_tx(db, tx); + IdcReducerCallOutcome::AlreadyAcked + } + }; + Ok(ControlFlow::Break(outcome)) + } + + async fn call_idc_reducer_with_params( + &self, + msg: IdcMessageParams, + reducer_name: &ReducerName, + params: CallReducerParams, + ) -> Result { + let IdcMessageParams { + sender_identity, + stream_id, + seq, + .. + } = msg; + + let on_success: ReducerSuccessAction = Box::new(move |tx, reducer_return_value| { + tx.record_inbound_idc_outcome( + sender_identity, + stream_id, + seq, + StInboundMsgResultStatus::Ok, + reducer_return_value.clone().unwrap_or_default(), + )?; + Ok(()) + }); + + self.guard_closed()?; + let timer_guard = self.start_call_timer(reducer_name); + match &*self.inner { + ModuleHostInner::Wasm(host) => { + let executor = host.executor.clone(); + executor + .run_sync_job(move |state| { + state.with_instance(move |inst| { + drop(timer_guard); + inst.call_idc_reducer(params, msg, on_success) + }) + }) + .await + } + ModuleHostInner::Js(host) => { + drop(timer_guard); + host.main_instance + .with_instance(|inst| async move { inst.call_idc_reducer(params, msg, on_success).await }) + .await + } + } + } + + /// Record a validation failure or a reducer failure after its transaction rolled back. + /// + /// Opens a new transaction and checks the stream before writing. Only the next + /// unapplied sequence can be recorded. The check and write share a transaction, + /// so another request cannot record an outcome between them. + /// + /// If another request already processed this message, returns its stored outcome + /// as [`IdcReducerCallOutcome::Replayed`], or [`IdcReducerCallOutcome::AlreadyAcked`] + /// if the sender acknowledged it and the outcome was removed. Neither case writes + /// a new failure. An out-of-order sequence returns [`IdcDeliveryFailure::OutOfOrder`] + /// without recording anything. + /// + /// For both newly recorded and replayed outcomes, the caller must wait for the + /// returned transaction offset to become durable before replying. + pub(super) fn record_idc_terminal_error( + stdb: &RelationalDB, + delivery: IdcMessageParams, + status: StInboundMsgResultStatus, + payload: Bytes, + ) -> Result { + let IdcMessageParams { + sender_identity, + stream_id, + seq, + .. + } = delivery; + + let tx = stdb.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + + let tx = match Self::check_idc_delivery(stdb, tx, delivery)? { + ControlFlow::Continue(tx) => tx, + ControlFlow::Break(outcome) => return Ok(outcome), + }; + + // The checking transaction stays open, so no other request can record this sequence. + let (tx, ()) = stdb + .with_auto_rollback(tx, |tx| { + tx.record_inbound_idc_outcome(sender_identity, stream_id, seq, status, payload.clone()) + }) + .inspect_err(|error| { + log::error!( + "Unexpected failure recording IDC outcome after sequence check: database_identity={} sender_identity={} stream_id={stream_id} sequence={seq}: {error:#}", + stdb.database_identity().to_hex(), + sender_identity.to_hex(), + ); + }) + .map_err(DBError::from)?; + + let (tx_offset, tx_data, tx_metrics, reducer) = stdb + .commit_tx(tx)? + .ok_or_else(|| IdcDeliveryFailure::Internal("IDC transaction was not committed".into()))?; + stdb.report_mut_tx_metrics(reducer, tx_metrics, Some(tx_data)); + + let outcome = if status == StInboundMsgResultStatus::BudgetExceeded { + ReducerOutcome::BudgetExceeded + } else { + let message = String::from_utf8_lossy(&payload).into_owned().into_boxed_str(); + ReducerOutcome::Failed(Box::new(message)) + }; + let result = ReducerCallResult { + outcome, + execution_budget_used: FunctionBudget::ZERO, + execution_duration: Duration::ZERO, + }; + Ok(IdcReducerCallOutcome::Recorded(ReducerCallResultWithTxOffset { + result, + tx_offset: from_tx_offset(tx_offset), + })) + } + fn log_reducer_submit_error(&self, reducer_name: &str, err: &ReducerCallError) { let log_message = match err { ReducerCallError::NoSuchReducer => Some(no_such_function_log_message("reducer", reducer_name)), @@ -2548,6 +2880,95 @@ impl ModuleHost { .await } + /// Remove acknowledged outcomes in a separate transaction so a reducer rollback + /// does not undo cleanup. + fn trim_idc_outcomes( + &self, + sender_identity: Identity, + stream_id: u64, + ack_prefix: u64, + ) -> Result<(), IdcDeliveryFailure> { + if ack_prefix == 0 { + return Ok(()); + } + let db = self.relational_db(); + let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let (tx, ()) = db + .with_auto_rollback(tx, |tx| { + tx.trim_inbound_idc_outcomes(sender_identity, stream_id, ack_prefix) + }) + .map_err(DBError::from)?; + let (_, tx_data, tx_metrics, reducer) = db + .commit_tx(tx)? + .ok_or_else(|| IdcDeliveryFailure::Internal("IDC transaction was not committed".into()))?; + db.report_mut_tx_metrics(reducer, tx_metrics, Some(tx_data)); + Ok(()) + } + + /// Processes an ordered IDC message and returns its recorded outcome. + /// + /// New messages are checked against the current module before running the reducer. + /// Successful changes and their outcome commit together. If validation or the + /// reducer fails, the error is recorded separately, with any reducer changes rolled back. + /// + /// Both success and recorded failure return [`IdcReducerCallOutcome::Recorded`]. + /// Retries return [`IdcReducerCallOutcome::Replayed`] without running the reducer + /// again. Once the sender acknowledges the outcome and it is removed, later + /// retries return [`IdcReducerCallOutcome::AlreadyAcked`]. + /// + /// # Errors + /// + /// Returns [`IdcDeliveryFailure`] if delivery cannot finish, including when an + /// outcome cannot be recorded. The sender should retry the same message. + /// + /// # Caller Requirements + /// + /// The caller handles authentication and the connection lifecycle. Send messages + /// in sequence order. Overlapping retries are safe, but out-of-order deliveries + /// are rejected rather than buffered for pipelining. Before replying, the caller + /// must wait for the returned transaction offset to become durable, including for replays. + pub async fn call_idc_reducer( + &self, + caller_identity: Identity, + caller_connection_id: ConnectionId, + reducer_name: &str, + args: FunctionArgs, + stream_id: u64, + seq: u64, + ack_prefix: u64, + expected_signature_hash: Hash, + ) -> Result { + self.trim_idc_outcomes(caller_identity, stream_id, ack_prefix)?; + let msg = IdcMessageParams { + sender_identity: caller_identity, + stream_id, + seq, + }; + + let reducer_params = self.call_idc_reducer_param( + caller_identity, + caller_connection_id, + reducer_name, + args, + expected_signature_hash, + ); + + let (reducer_def, params) = match reducer_params { + Ok(params) => params, + Err(err) => { + // Only a recorded failure becomes a terminal response. If this + // write fails, return a delivery failure so the sender retries. + return Self::record_idc_terminal_error( + self.relational_db(), + msg, + StInboundMsgResultStatus::Err, + Bytes::from(err.to_string()), + ); + } + }; + self.call_idc_reducer_with_params(msg, &reducer_def.name, params).await + } + pub async fn enqueue_reducer( &self, caller_identity: Identity, @@ -3830,7 +4251,7 @@ fn args_error_log_message(function_kind: &str, function_name: &str) -> String { #[cfg(test)] mod tests { - use super::ModuleHost; + use super::{ModuleHost, ModuleInfo}; use crate::client::{ ClientActorId, ClientConfig, ClientConnectionReceiver, ClientConnectionSender, OutboundMessage, Protocol, WsVersion, @@ -3843,6 +4264,210 @@ mod tests { use spacetimedb_sats::product; use std::sync::Arc; + #[tokio::test] + async fn idc_failure_recording_preserves_a_concurrent_outcome() -> anyhow::Result<()> { + use super::{IdcMessageParams, IdcReducerCallOutcome, IdcSequenceState}; + use bytes::Bytes; + use spacetimedb_datastore::execution_context::Workload; + use spacetimedb_datastore::system_tables::StInboundMsgResultStatus; + use spacetimedb_datastore::traits::IsolationLevel; + + for status in [ + StInboundMsgResultStatus::Ok, + StInboundMsgResultStatus::Err, + StInboundMsgResultStatus::BudgetExceeded, + ] { + let db = TestDB::in_memory()?; + let delivery = IdcMessageParams { + sender_identity: Identity::ONE, + stream_id: 1, + seq: 1, + }; + + // The first attempt checks the stream, then rolls back its reducer. + let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::ForTests); + assert!(matches!( + ModuleHost::check_idc_stream_seq(&tx, delivery)?, + IdcSequenceState::Ready + )); + let _ = db.rollback_mut_tx(tx); + + // A retry commits in the gap before the first attempt records its error. + let payload = Bytes::from_static(b"winning outcome"); + with_auto_commit(&db, |tx| { + tx.record_inbound_idc_outcome( + delivery.sender_identity, + delivery.stream_id, + delivery.seq, + status, + payload.clone(), + ) + .map_err(crate::error::DBError::from) + })?; + let result = ModuleHost::record_idc_terminal_error( + &db, + delivery, + StInboundMsgResultStatus::Err, + Bytes::from_static(b"late error"), + )?; + let IdcReducerCallOutcome::Replayed(result) = result else { + panic!("failure recording must return the winning outcome"); + }; + assert_eq!(result.result_status, status); + assert_eq!(result.result_payload, payload); + } + Ok(()) + } + + #[tokio::test] + async fn idc_failure_recording_checks_the_applied_prefix() -> anyhow::Result<()> { + use super::{IdcDeliveryFailure, IdcMessageParams, IdcReducerCallOutcome, ReducerOutcome}; + use bytes::Bytes; + use spacetimedb_datastore::system_tables::StInboundMsgResultStatus; + + for status in [StInboundMsgResultStatus::Err, StInboundMsgResultStatus::BudgetExceeded] { + let db = TestDB::in_memory()?; + let record = |seq| { + ModuleHost::record_idc_terminal_error( + &db, + IdcMessageParams { + sender_identity: Identity::ONE, + stream_id: 1, + seq, + }, + status, + Bytes::from_static(b"rejected"), + ) + }; + let IdcReducerCallOutcome::Recorded(recorded) = record(1)? else { + panic!("first delivery must record a failure"); + }; + // The first response and retries must agree on the failure's classification. + assert!(matches!( + (status, recorded.result.outcome), + (StInboundMsgResultStatus::Err, ReducerOutcome::Failed(_)) + | (StInboundMsgResultStatus::BudgetExceeded, ReducerOutcome::BudgetExceeded) + )); + let offset = recorded.tx_offset.await?; + let IdcReducerCallOutcome::Replayed(replayed) = record(1)? else { + panic!("duplicate delivery must replay the failure"); + }; + assert_eq!(replayed.result_status, status); + assert_eq!(replayed.result_payload, Bytes::from_static(b"rejected")); + assert!(replayed.tx_offset >= offset); + assert!(matches!(record(2)?, IdcReducerCallOutcome::Recorded(_))); + with_auto_commit(&db, |tx| { + tx.trim_inbound_idc_outcomes(Identity::ONE, 1, 1) + .map_err(crate::error::DBError::from) + })?; + + // Sequence 1 has been acknowledged and trimmed. A late failure must not + // recreate it or move the prefix backward, even without an outcome row. + assert!(matches!(record(1)?, IdcReducerCallOutcome::AlreadyAcked)); + assert!(matches!( + record(4), + Err(IdcDeliveryFailure::OutOfOrder { expected: 3, actual: 4 }) + )); + assert!(matches!(record(3)?, IdcReducerCallOutcome::Recorded(_))); + } + Ok(()) + } + + #[tokio::test] + async fn reducer_signature_hash_cache_is_lazy_and_scoped_to_module() -> anyhow::Result<()> { + use spacetimedb_lib::db::raw_def::v10::RawModuleDefV10Builder; + use spacetimedb_lib::{hash_bytes, ProductType}; + use spacetimedb_schema::def::reducer_signature_hash; + use spacetimedb_schema::reducer_name::ReducerName; + + let TestDB { db, .. } = TestDB::in_memory()?; + let subscriptions = ModuleSubscriptions::for_test_enclosing_runtime(db); + let make_info = |child_type| { + let mut child = RawModuleDefV10Builder::new(); + let payload = + child.add_algebraic_type([], "Payload", AlgebraicType::product([("value", child_type)]), false); + child.add_reducer("receive", ProductType::from([("payload", payload.into())])); + + let mut nested = RawModuleDefV10Builder::new(); + nested.add_reducer("receive", ProductType::from([("payload", AlgebraicType::U8)])); + child.add_submodule("nested", nested.finish()); + + let mut sibling = RawModuleDefV10Builder::new(); + sibling.add_reducer("receive", ProductType::from([("payload", AlgebraicType::U32)])); + + let mut parent = RawModuleDefV10Builder::new(); + let payload = parent.add_algebraic_type( + [], + "Payload", + AlgebraicType::product([("value", AlgebraicType::U64)]), + false, + ); + parent.add_reducer("receive", ProductType::from([("payload", payload.into())])); + parent.add_submodule("child", child.finish()); + parent.add_submodule("sibling", sibling.finish()); + ModuleInfo::new( + parent.finish().try_into().unwrap(), + Identity::ZERO, + Identity::ONE, + hash_bytes(b"test module"), + subscriptions.clone(), + ) + }; + let hash_for = |info: &ModuleInfo, name| { + let (_, reducer, owning) = info.module_def.reducer_by_name_with_module(name).unwrap(); + let hash = info.reducer_signature_hash(&reducer.name); + assert_eq!( + hash, + reducer_signature_hash( + &reducer.params(), + &reducer.ok_return_type, + &reducer.err_return_type, + owning.typespace(), + ) + ); + assert_eq!(info.reducer_signature_hashes[&reducer.name].get(), Some(&hash)); + hash + }; + + let info = make_info(AlgebraicType::String); + assert_eq!(info.reducer_signature_hashes.len(), 4); + assert!(info.reducer_signature_hashes.values().all(|slot| slot.get().is_none())); + let child_hash = hash_for(&info, "child.receive"); + for name in ["receive", "sibling.receive", "child.nested.receive"] { + assert!(info.reducer_signature_hashes[&ReducerName::for_test(name)] + .get() + .is_none()); + } + assert_eq!(hash_for(&info.clone(), "child.receive"), child_hash); + let parent_hash = hash_for(&info, "receive"); + assert_ne!(parent_hash, child_hash); + let sibling_hash = hash_for(&info, "sibling.receive"); + let nested_hash = hash_for(&info, "child.nested.receive"); + assert_ne!(sibling_hash, child_hash); + assert_ne!(nested_hash, child_hash); + + // Reloading the same definition starts empty and produces the same hash. + let reloaded = make_info(AlgebraicType::String); + assert!(reloaded + .reducer_signature_hashes + .values() + .all(|slot| slot.get().is_none())); + assert_eq!(hash_for(&reloaded, "child.receive"), child_hash); + + // A changed module must not reuse a hash cached for the same reducer name. + let updated = make_info(AlgebraicType::Bool); + assert!(updated + .reducer_signature_hashes + .values() + .all(|slot| slot.get().is_none())); + assert_ne!(hash_for(&updated, "child.receive"), child_hash); + assert_eq!(hash_for(&updated, "receive"), parent_hash); + assert_eq!(hash_for(&updated, "sibling.receive"), sibling_hash); + assert_eq!(hash_for(&updated, "child.nested.receive"), nested_hash); + assert_eq!(hash_for(&info, "child.receive"), child_hash); + Ok(()) + } + fn v2_client_config() -> ClientConfig { ClientConfig { protocol: Protocol::Binary, diff --git a/crates/core/src/host/v8/mod.rs b/crates/core/src/host/v8/mod.rs index 3f5b4ddccd5..8f3673e0c77 100644 --- a/crates/core/src/host/v8/mod.rs +++ b/crates/core/src/host/v8/mod.rs @@ -65,8 +65,8 @@ use self::syscall::{ }; use super::module_common::{build_common_module_from_raw, run_describer, ModuleCommon}; use super::module_host::{ - CallHttpHandlerParams, CallProcedureParams, CallReducerParams, InstanceManagerMetrics, ModuleInfo, - ModuleWithInstance, UpdateEnvironmentResult, + CallHttpHandlerParams, CallProcedureParams, CallReducerParams, IdcDeliveryFailure, IdcMessageParams, + IdcReducerCallOutcome, InstanceManagerMetrics, ModuleInfo, ModuleWithInstance, UpdateEnvironmentResult, }; use super::UpdateDatabaseResult; use crate::client::{ClientActorId, MeteredUnboundedReceiver, MeteredUnboundedSender}; @@ -82,7 +82,7 @@ use crate::host::wasm_common::instrumentation::CallTimes; use crate::host::wasm_common::module_host_actor::{ AnonymousViewOp, DescribeError, EnergyStats, ExecutionError, ExecutionResult, ExecutionStats, ExecutionTimings, HttpHandlerExecuteResult, HttpHandlerOp, InstanceCommon, InstanceOp, ProcedureExecuteResult, ProcedureOp, - ReducerExecuteResult, ReducerOp, ViewExecuteResult, ViewOp, WasmInstance, + ReducerExecuteResult, ReducerOp, ReducerSuccessAction, ViewExecuteResult, ViewOp, WasmInstance, }; use crate::host::wasm_common::{RowIters, TimingSpanSet}; use crate::host::{ @@ -507,6 +507,20 @@ impl JsMainInstance { self.request(CallReducerRequest { params }).await } + pub(in crate::host) async fn call_idc_reducer( + &self, + params: CallReducerParams, + delivery: IdcMessageParams, + on_success: ReducerSuccessAction, + ) -> Result { + self.request(CallIdcReducerRequest { + params, + delivery, + on_success, + }) + .await + } + pub(in crate::host) async fn call_scheduled_reducer( &self, params: ScheduledFunctionParams, @@ -665,6 +679,14 @@ js_main_request! { } => "scheduled_reducer", CallScheduledFunctionResult, ScheduledReducer } +js_main_request! { + CallIdcReducerRequest { + params: CallReducerParams, + delivery: IdcMessageParams, + on_success: ReducerSuccessAction, + } => "call_idc_reducer", Result, CallIdcReducer +} + js_main_request! { ClearAllClientsRequest => "clear_all_clients", anyhow::Result<()>, ClearAllClients } @@ -847,6 +869,13 @@ enum JsMainWorkerRequest { reply_tx: JsReplyTx, params: CallReducerParams, }, + /// See [`JsMainInstance::call_idc_reducer`]. + CallIdcReducer { + reply_tx: JsReplyTx>, + params: CallReducerParams, + delivery: IdcMessageParams, + on_success: ReducerSuccessAction, + }, /// See [`JsMainInstance::enqueue_reducer`]. CallReducerDetached { params: CallReducerParams, @@ -1453,6 +1482,14 @@ fn handle_main_worker_request( call_reducer(None, params) }) } + JsMainWorkerRequest::CallIdcReducer { + reply_tx, + params, + delivery, + on_success, + } => handle_worker_request("call_idc_reducer", reply_tx, || { + instance_common.call_idc_reducer(params, delivery, inst, on_success) + }), JsMainWorkerRequest::CallReducerDetached { params, on_panic } => { handle_detached_worker_request("call_reducer", on_panic, || { let mut call_reducer = |tx, params| instance_common.call_reducer_with_tx(tx, params, inst); diff --git a/crates/core/src/host/wasm_common/module_host_actor.rs b/crates/core/src/host/wasm_common/module_host_actor.rs index fc8bf93bde4..2e52679ad1d 100644 --- a/crates/core/src/host/wasm_common/module_host_actor.rs +++ b/crates/core/src/host/wasm_common/module_host_actor.rs @@ -11,9 +11,10 @@ use crate::host::instance_env::{InstanceEnv, TxSlot}; use crate::host::module_common::{build_common_module_from_raw, ModuleCommon}; use crate::host::module_host::{ call_identity_connected, init_database, CallHttpHandlerParams, CallProcedureParams, CallReducerParams, - CallViewParams, ClientConnectedError, DatabaseUpdate, EventStatus, HttpHandlerCallError, InitDatabaseResult, - ModuleEvent, ModuleFunctionCall, ModuleInfo, RefInstance, SqlCommand, SqlCommandResult, UpdateEnvironmentResult, - ViewCallResult, ViewCommand, ViewCommandResult, ViewOutcome, + CallViewParams, ClientConnectedError, DatabaseUpdate, EventStatus, HttpHandlerCallError, IdcDeliveryFailure, + IdcMessageParams, IdcReducerCallOutcome, InitDatabaseResult, ModuleEvent, ModuleFunctionCall, ModuleInfo, + RefInstance, SqlCommand, SqlCommandResult, UpdateEnvironmentResult, ViewCallResult, ViewCommand, ViewCommandResult, + ViewOutcome, }; use crate::host::scheduler::{CallScheduledFunctionResult, ScheduledFunctionParams}; use crate::host::{ @@ -39,6 +40,7 @@ use spacetimedb_datastore::error::{DatastoreError, ViewError}; use spacetimedb_datastore::execution_context::{self, ReducerContext, Workload}; use spacetimedb_datastore::locking_tx_datastore::state_view::StateView; use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, MutTxId, ViewCallInfo, ViewInstanceArgs}; +use spacetimedb_datastore::system_tables::StInboundMsgResultStatus; use spacetimedb_datastore::traits::{IsolationLevel, Program}; use spacetimedb_execution::ExecutionParams; use spacetimedb_lib::buffer::DecodeError; @@ -58,9 +60,13 @@ use spacetimedb_schema::identifier::NamespacedIdentifier; use spacetimedb_schema::reducer_name::ReducerName; use spacetimedb_subscription::SubscriptionPlan; use std::collections::HashMap; +use std::ops::ControlFlow; use std::sync::Arc; use tracing::span::EnteredSpan; +/// Runs after a successful reducer, before views refresh. Errors roll back the transaction. +pub(crate) type ReducerSuccessAction = Box) -> anyhow::Result<()> + Send>; + pub trait WasmModule: Send + 'static { type Instance: WasmInstance; type InstancePre: WasmInstancePre; @@ -525,6 +531,20 @@ impl WasmModuleInstance { res } + pub(in crate::host) fn call_idc_reducer( + &mut self, + params: CallReducerParams, + delivery: IdcMessageParams, + on_success: ReducerSuccessAction, + ) -> Result { + let (result, trapped) = crate::callgrind_flag::invoke_allowing_callgrind(|| { + self.common + .call_idc_reducer(params, delivery, &mut self.instance, on_success) + }); + self.trapped = trapped; + result + } + pub fn clear_all_clients(&self) -> anyhow::Result<()> { self.common.clear_all_clients() } @@ -620,9 +640,7 @@ impl WasmModuleInstance { impl WasmModuleInstance { #[tracing::instrument(level = "trace", skip_all)] fn call_reducer_with_tx(&mut self, tx: Option, params: CallReducerParams) -> (ReducerCallResult, bool) { - let (res, trapped) = crate::callgrind_flag::invoke_allowing_callgrind(|| { - self.common.call_reducer_with_tx(tx, params, &mut self.instance) - }); + let (res, trapped) = self.call_reducer_with_tx_offset(tx, params); (res.result, trapped) } @@ -1072,6 +1090,80 @@ impl InstanceCommon { tx: Option, params: CallReducerParams, inst: &mut I, + ) -> (ReducerCallResultWithTxOffset, bool) { + self.call_reducer_with_tx_inner(tx, params, inst, None) + } + + pub(crate) fn call_reducer_with_success_action( + &mut self, + tx: Option, + params: CallReducerParams, + inst: &mut I, + on_success: ReducerSuccessAction, + ) -> (ReducerCallResultWithTxOffset, bool) { + self.call_reducer_with_tx_inner(tx, params, inst, Some(on_success)) + } + + /// Execute the next IDC delivery or return its previously recorded outcome. + /// Failed execution is rolled back before recording its outcome in a new transaction. + /// The boolean reports an execution trap, even if recording the failure also fails. + pub(in crate::host) fn call_idc_reducer( + &mut self, + params: CallReducerParams, + delivery: IdcMessageParams, + inst: &mut I, + on_success: ReducerSuccessAction, + ) -> (Result, bool) { + let db = inst.replica_ctx().relational_db(); + let workload = Workload::Reducer(ReducerContext { + name: self.info.module_def.reducer_by_id(params.reducer_id).name.clone(), + caller_identity: params.caller_identity, + caller_connection_id: params.caller_connection_id, + timestamp: params.timestamp, + arg_bsatn: params.args.get_bsatn().clone(), + }); + let tx = db.begin_mut_tx(IsolationLevel::Serializable, workload); + let tx = match ModuleHost::check_idc_delivery(db, tx, delivery) { + Ok(ControlFlow::Continue(tx)) => tx, + Ok(ControlFlow::Break(outcome)) => return (Ok(outcome), false), + Err(error) => return (Err(error), false), + }; + + // Keep the sequence check, reducer writes, and success hook in one transaction. + let (attempt, trapped) = self.call_reducer_with_success_action(Some(tx), params, inst, on_success); + let (status, payload) = match &attempt.result.outcome { + ReducerOutcome::Committed => return (Ok(IdcReducerCallOutcome::Recorded(attempt)), trapped), + ReducerOutcome::Failed(error) => (StInboundMsgResultStatus::Err, Bytes::copy_from_slice(error.as_bytes())), + ReducerOutcome::BudgetExceeded => ( + StInboundMsgResultStatus::BudgetExceeded, + Bytes::from_static(b"Module energy budget exhausted."), + ), + }; + + // Execution rolled back. Record the failure in a new transaction, checking + // for a competing outcome before writing. Recorded always means a persisted outcome. + let db = inst.replica_ctx().relational_db(); + let recorded = ModuleHost::record_idc_terminal_error(db, delivery, status, payload); + let outcome = recorded.map(|outcome| match outcome { + IdcReducerCallOutcome::Recorded(recorded) => { + // Use this attempt's result and metrics with the failure commit's offset. + IdcReducerCallOutcome::Recorded(ReducerCallResultWithTxOffset { + result: attempt.result, + tx_offset: recorded.tx_offset, + }) + } + // Another request won after rollback. Return its outcome unchanged. + other => other, + }); + (outcome, trapped) + } + + fn call_reducer_with_tx_inner( + &mut self, + tx: Option, + params: CallReducerParams, + inst: &mut I, + on_success: Option, ) -> (ReducerCallResultWithTxOffset, bool) { let CallReducerParams { timestamp, @@ -1126,7 +1218,7 @@ impl InstanceCommon { // However, that does not necessarily apply to e.g., V8. let trapped = matches!(result.call_result, Err(ExecutionError::Trap(_))); - let (status, mut reducer_return_value) = match result.call_result { + let (mut status, mut reducer_return_value) = match result.call_result { Err(ExecutionError::Recoverable(err) | ExecutionError::Trap(err)) => { inst.log_traceback("reducer", reducer_name, &err); @@ -1170,7 +1262,16 @@ impl InstanceCommon { } }; - // Only re-evaluate and update views if the reducer's execution was successful + // Views must see the hook's writes as well as the reducer's. A failure in + // either the hook or view execution rolls back all of these changes. + if matches!(status, EventStatus::Committed(_)) + && let Some(on_success) = on_success + && let Err(err) = on_success(&mut tx, &reducer_return_value) + { + status = EventStatus::FailedInternal(err.to_string()); + } + + // Only re-evaluate and update views if the reducer and success hook succeeded. let (out, n_views_evaluated, trapped) = if !trapped && matches!(status, EventStatus::Committed(_)) { self.call_views_with_tx(tx, caller_identity, inst, timestamp) } else { diff --git a/crates/core/src/subscription/module_subscription_actor.rs b/crates/core/src/subscription/module_subscription_actor.rs index f5ba88774b2..81b00e8b2ae 100644 --- a/crates/core/src/subscription/module_subscription_actor.rs +++ b/crates/core/src/subscription/module_subscription_actor.rs @@ -2383,7 +2383,7 @@ mod tests { send_worker_queue, BsatnRowListBuilderPool::new(), ); - let auth = AuthCtx::new(owner, sender.auth.claims.identity); + let auth = AuthCtx::new(owner, sender.auth.identity); let subscribe = ws_v1::Subscribe { query_strings: [sql.into()].into(), diff --git a/crates/datastore/src/locking_tx_datastore/datastore.rs b/crates/datastore/src/locking_tx_datastore/datastore.rs index 6840e630f41..79d9c79c708 100644 --- a/crates/datastore/src/locking_tx_datastore/datastore.rs +++ b/crates/datastore/src/locking_tx_datastore/datastore.rs @@ -1114,17 +1114,17 @@ pub(crate) mod tests { use crate::locking_tx_datastore::tx_state::PendingSchemaChange; use crate::system_tables::{ system_tables, IdentityViaU256, StColumnRow, StConnectionCredentialsFields, StConstraintData, - StConstraintFields, StConstraintRow, StEventTableFields, StIndexAlgorithm, StIndexFields, StIndexRow, - StOutboundMsgRow, StOutboundStreamRow, StRowLevelSecurityFields, StScheduledFields, StSequenceFields, - StSequenceRow, StTableRow, StVarFields, StViewArgFields, StViewFields, ST_CLIENT_ID, ST_CLIENT_NAME, - ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ACCESSOR_NAME, ST_COLUMN_ID, ST_COLUMN_NAME, ST_CONNECTION_CREDENTIALS_ID, - ST_CONNECTION_CREDENTIALS_NAME, ST_CONSTRAINT_ID, ST_CONSTRAINT_NAME, ST_EVENT_TABLE_ID, ST_EVENT_TABLE_NAME, - ST_INDEX_ACCESSOR_ID, ST_INDEX_ACCESSOR_NAME, ST_INDEX_ID, ST_INDEX_NAME, ST_MODULE_NAME, - ST_RESERVED_SEQUENCE_RANGE, ST_ROW_LEVEL_SECURITY_ID, ST_ROW_LEVEL_SECURITY_NAME, ST_SCHEDULED_ID, - ST_SCHEDULED_NAME, ST_SEQUENCE_ID, ST_SEQUENCE_NAME, ST_TABLE_ACCESSOR_ID, ST_TABLE_ACCESSOR_NAME, - ST_TABLE_NAME, ST_VAR_ID, ST_VAR_NAME, ST_VIEW_ARG_ID, ST_VIEW_ARG_NAME, ST_VIEW_COLUMN_ID, - ST_VIEW_COLUMN_NAME, ST_VIEW_ID, ST_VIEW_NAME, ST_VIEW_PARAM_ID, ST_VIEW_PARAM_NAME, ST_VIEW_SUB_ID, - ST_VIEW_SUB_NAME, + StConstraintFields, StConstraintRow, StEventTableFields, StInboundMsgResultStatus, StInboundMsgRow, + StInboundStreamRow, StIndexAlgorithm, StIndexFields, StIndexRow, StOutboundMsgRow, StOutboundStreamRow, + StRowLevelSecurityFields, StScheduledFields, StSequenceFields, StSequenceRow, StTableRow, StVarFields, + StViewArgFields, StViewFields, ST_CLIENT_ID, ST_CLIENT_NAME, ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ACCESSOR_NAME, + ST_COLUMN_ID, ST_COLUMN_NAME, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_NAME, ST_CONSTRAINT_ID, + ST_CONSTRAINT_NAME, ST_EVENT_TABLE_ID, ST_EVENT_TABLE_NAME, ST_INDEX_ACCESSOR_ID, ST_INDEX_ACCESSOR_NAME, + ST_INDEX_ID, ST_INDEX_NAME, ST_MODULE_NAME, ST_RESERVED_SEQUENCE_RANGE, ST_ROW_LEVEL_SECURITY_ID, + ST_ROW_LEVEL_SECURITY_NAME, ST_SCHEDULED_ID, ST_SCHEDULED_NAME, ST_SEQUENCE_ID, ST_SEQUENCE_NAME, + ST_TABLE_ACCESSOR_ID, ST_TABLE_ACCESSOR_NAME, ST_TABLE_NAME, ST_VAR_ID, ST_VAR_NAME, ST_VIEW_ARG_ID, + ST_VIEW_ARG_NAME, ST_VIEW_COLUMN_ID, ST_VIEW_COLUMN_NAME, ST_VIEW_ID, ST_VIEW_NAME, ST_VIEW_PARAM_ID, + ST_VIEW_PARAM_NAME, ST_VIEW_SUB_ID, ST_VIEW_SUB_NAME, }; use crate::system_tables::{ ST_ENV_ID, ST_ENV_NAME, ST_INBOUND_MSG_ID, ST_INBOUND_MSG_NAME, ST_INBOUND_STREAM_ID, ST_INBOUND_STREAM_NAME, @@ -1133,6 +1133,7 @@ pub(crate) mod tests { }; use crate::traits::{IsolationLevel, MutTx}; use crate::Result; + use bytes::Bytes; use core::{fmt, mem}; use itertools::Itertools; use pretty_assertions::{assert_eq, assert_matches}; @@ -1142,7 +1143,7 @@ pub(crate) mod tests { use spacetimedb_lib::error::ResultTest; use spacetimedb_lib::st_var::StVarValue; use spacetimedb_lib::{resolved_type_via_v9, Identity, ScheduleAt, TimeDuration}; - use spacetimedb_primitives::{col_list, ArgId, ColId, ColSet, ScheduleId, ViewId}; + use spacetimedb_primitives::{col_list, ArgId, ColId, ColSet, ScheduleId, TableId, ViewId}; use spacetimedb_sats::algebraic_value::ser::value_serialize; use spacetimedb_sats::bsatn::{to_vec, ToBsatn}; use spacetimedb_sats::layout::RowTypeLayout; @@ -3138,6 +3139,118 @@ pub(crate) mod tests { Ok(()) } + #[test] + fn test_inbound_idc_records_replays_and_trims_outcomes() -> ResultTest<()> { + let datastore = get_datastore()?; + let mut tx = begin_mut_tx(&datastore); + // Bootstrap defers indexes until replay completes; duplicate checks require this one. + tx.create_index(crate::system_tables::st_inbound_msg_schema().indexes[0].clone(), true)?; + let sender = Identity::ONE; + let stream_id = 7; + + assert_eq!(tx.inbound_idc_applied_prefix(sender, stream_id)?, 0); + assert!(tx.inbound_idc_outcome(sender, stream_id, 1)?.is_none()); + + tx.record_inbound_idc_outcome( + sender, + stream_id, + 1, + StInboundMsgResultStatus::Ok, + Bytes::from_static(b"ok"), + )?; + assert_eq!(tx.inbound_idc_applied_prefix(sender, stream_id)?, 1); + + let outcome = tx + .inbound_idc_outcome(sender, stream_id, 1)? + .expect("outcome should be retained for replay"); + assert_eq!(outcome.result_status, StInboundMsgResultStatus::Ok); + assert_eq!(&outcome.result_payload[..], b"ok"); + for (status, payload) in [ + (StInboundMsgResultStatus::Ok, Bytes::from_static(b"ok")), + (StInboundMsgResultStatus::Err, Bytes::from_static(b"changed")), + ] { + assert!( + tx.record_inbound_idc_outcome(sender, stream_id, 1, status, payload) + .is_err(), + "IDC outcomes are immutable and retries must replay the first decision ({status:?})", + ); + assert_eq!(tx.inbound_idc_outcome(sender, stream_id, 1)?, Some(outcome.clone())); + } + + tx.trim_inbound_idc_outcomes(sender, stream_id, 1)?; + assert_eq!(tx.inbound_idc_applied_prefix(sender, stream_id)?, 1); + assert!(tx.inbound_idc_outcome(sender, stream_id, 1)?.is_none()); + + let stream_rows = tx + .iter(ST_INBOUND_STREAM_ID)? + .map(StInboundStreamRow::try_from) + .collect::>>()?; + assert_eq!(stream_rows.len(), 1); + assert_eq!(stream_rows[0].sender_identity, IdentityViaU256(sender)); + assert_eq!(stream_rows[0].stream_id, stream_id); + assert_eq!(stream_rows[0].applied_prefix, 1); + + let msg_rows = tx + .iter(ST_INBOUND_MSG_ID)? + .map(StInboundMsgRow::try_from) + .collect::>>()?; + assert!(msg_rows.is_empty()); + + Ok(()) + } + + #[test] + fn test_inbound_idc_trim_range_and_repeated_acknowledgments() -> ResultTest<()> { + // The range scan must handle both the committed index and transaction-local rows. + for commit_before_trim in [false, true] { + let datastore = get_datastore()?; + let mut tx = begin_mut_tx(&datastore); + let sender = Identity::ONE; + let stream_id = 7; + let streams = [ + (sender, stream_id), + (sender, 6), + (sender, 8), + (Identity::ZERO, stream_id), + ]; + for (sender, stream_id) in streams { + for seq in 1..=3 { + tx.record_inbound_idc_outcome(sender, stream_id, seq, StInboundMsgResultStatus::Ok, Bytes::new())?; + } + } + if commit_before_trim { + datastore.commit_mut_tx(tx)?; + tx = begin_mut_tx(&datastore); + } + + // Zero acknowledges nothing; repeated or older acknowledgments are harmless. + tx.trim_inbound_idc_outcomes(sender, stream_id, 0)?; + assert!(tx.inbound_idc_outcome(sender, stream_id, 1)?.is_some()); + for ack_prefix in [1, 1, 0] { + tx.trim_inbound_idc_outcomes(sender, stream_id, ack_prefix)?; + assert!(tx.inbound_idc_outcome(sender, stream_id, 1)?.is_none()); + assert!(tx.inbound_idc_outcome(sender, stream_id, 2)?.is_some()); + assert!(tx.inbound_idc_outcome(sender, stream_id, 3)?.is_some()); + } + + // Even the largest bound must stay within this sender and stream. + tx.trim_inbound_idc_outcomes(sender, stream_id, u64::MAX)?; + tx.trim_inbound_idc_outcomes(sender, stream_id, u64::MAX)?; + for (other_sender, other_stream) in streams { + let retained = (other_sender, other_stream) != (sender, stream_id); + for seq in 1..=3 { + assert_eq!( + tx.inbound_idc_outcome(other_sender, other_stream, seq)?.is_some(), + retained + ); + } + // Trimming results must not allow an old sequence to execute again. + assert_eq!(tx.inbound_idc_applied_prefix(other_sender, other_stream)?, 3); + } + } + Ok(()) + } + #[test] fn test_row_level_security() -> ResultTest<()> { let (_, mut tx, table_id) = setup_table()?; diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 324cc680d84..d06a0366289 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -22,14 +22,15 @@ use crate::{ system_tables::{ with_sys_table_buf, StClientFields, StClientRow, StColumnAccessorFields, StColumnAccessorRow, StColumnFields, StColumnRow, StConstraintFields, StConstraintRow, StEventTableFields, StEventTableRow, StFields as _, + StInboundMsgFields, StInboundMsgResultStatus, StInboundMsgRow, StInboundStreamFields, StInboundStreamRow, StIndexAccessorFields, StIndexAccessorRow, StIndexFields, StIndexRow, StOutboundMsgFields, StOutboundMsgRow, StOutboundStreamFields, StOutboundStreamRow, StOutboxFields, StOutboxRow, StRowLevelSecurityFields, StRowLevelSecurityRow, StScheduledFields, StScheduledRow, StSequenceFields, StSequenceRow, StTableAccessorFields, StTableAccessorRow, StTableFields, StTableRow, SystemTable, INITIAL_OUTBOUND_STREAM_ACK_PREFIX, INITIAL_OUTBOUND_STREAM_NEXT_SEQ, ST_CLIENT_ID, ST_COLUMN_ACCESSOR_ID, - ST_COLUMN_ID, ST_CONSTRAINT_ID, ST_EVENT_TABLE_ID, ST_INDEX_ACCESSOR_ID, ST_INDEX_ID, ST_OUTBOUND_MSG_ID, - ST_OUTBOUND_STREAM_ID, ST_OUTBOX_ID, ST_ROW_LEVEL_SECURITY_ID, ST_SCHEDULED_ID, ST_SEQUENCE_ID, - ST_TABLE_ACCESSOR_ID, ST_TABLE_ID, + ST_COLUMN_ID, ST_CONSTRAINT_ID, ST_EVENT_TABLE_ID, ST_INBOUND_MSG_ID, ST_INBOUND_STREAM_ID, + ST_INDEX_ACCESSOR_ID, ST_INDEX_ID, ST_OUTBOUND_MSG_ID, ST_OUTBOUND_STREAM_ID, ST_OUTBOX_ID, + ST_ROW_LEVEL_SECURITY_ID, ST_SCHEDULED_ID, ST_SEQUENCE_ID, ST_TABLE_ACCESSOR_ID, ST_TABLE_ID, }, }; use crate::{execution_context::ExecutionContext, system_tables::StViewColumnRow}; @@ -38,6 +39,8 @@ use crate::{ locking_tx_datastore::state_view::ScanOrIndex, traits::{InsertFlags, RowTypeForTable, TxData, UpdateFlags}, }; +use anyhow::anyhow; +use bytes::Bytes; use core::{cell::RefCell, iter, ops::RangeBounds}; use itertools::Either; use rand_core::RngCore; @@ -3356,7 +3359,7 @@ impl MutTxId { &mut self, identity: Identity, connection_id: ConnectionId, - jwt_payload: &str, + jwt_payload: Option<&str>, ) -> Result<()> { let row = &StClientRow { identity: identity.into(), @@ -3369,7 +3372,10 @@ impl MutTxId { "[{identity}]: insert_st_client: failed to insert client ({identity}, {connection_id}), error: {e}" ); })?; - self.insert_st_client_credentials(connection_id, jwt_payload) + if let Some(jwt_payload) = jwt_payload { + self.insert_st_client_credentials(connection_id, jwt_payload)?; + } + Ok(()) } fn insert_st_client_credentials(&mut self, connection_id: ConnectionId, jwt_payload: &str) -> Result<()> { @@ -3462,6 +3468,109 @@ impl MutTxId { }) } + fn st_inbound_stream_row(&self, sender: Identity, stream_id: u64) -> Result> { + let key = AlgebraicValue::Product(product![IdentityViaU256(sender), stream_id]); + self.iter_by_col_eq( + ST_INBOUND_STREAM_ID, + col_list![StInboundStreamFields::SenderIdentity, StInboundStreamFields::StreamId], + &key, + )? + .map(StInboundStreamRow::try_from) + .next() + .transpose() + } + + /// Return the highest contiguous inbound sequence already applied for a stream. + pub fn inbound_idc_applied_prefix(&self, sender: Identity, stream_id: u64) -> Result { + Ok(self + .st_inbound_stream_row(sender, stream_id)? + .map_or(0, |row| row.applied_prefix)) + } + + /// Look up a retained inbound result for replaying a duplicate delivery. + pub fn inbound_idc_outcome(&self, sender: Identity, stream_id: u64, seq: u64) -> Result> { + let key = AlgebraicValue::Product(product![IdentityViaU256(sender), stream_id, seq]); + self.iter_by_col_eq( + ST_INBOUND_MSG_ID, + col_list![ + StInboundMsgFields::SenderIdentity, + StInboundMsgFields::StreamId, + StInboundMsgFields::Seq + ], + &key, + )? + .map(StInboundMsgRow::try_from) + .next() + .transpose() + } + + /// Store the result of applying an inbound IDC message. + /// + /// The applied prefix is kept separately from retained message results so old + /// results can be trimmed after ack without allowing old messages to run again. + /// Outcomes are immutable: retries must replay the first durable decision. + pub fn record_inbound_idc_outcome( + &mut self, + sender: Identity, + stream_id: u64, + seq: u64, + result_status: StInboundMsgResultStatus, + result_payload: Bytes, + ) -> Result<()> { + if let Some(row) = self.st_inbound_stream_row(sender, stream_id)? { + self.delete_by_row_value(ST_INBOUND_STREAM_ID, &row.clone().into())?; + } + self.insert_via_serialize_bsatn( + ST_INBOUND_STREAM_ID, + &StInboundStreamRow { + sender_identity: IdentityViaU256(sender), + stream_id, + applied_prefix: seq, + }, + )?; + + let (_, inserted, _) = self.insert_via_serialize_bsatn( + ST_INBOUND_MSG_ID, + &StInboundMsgRow { + sender_identity: IdentityViaU256(sender), + stream_id, + seq, + result_status, + result_payload, + }, + )?; + if matches!(inserted, RowRefInsertion::Existed(_)) { + return Err( + anyhow!("IDC outcome already exists for sender {sender}, stream {stream_id}, seq {seq}").into(), + ); + } + Ok(()) + } + + /// Delete retained inbound results the sender has acknowledged. + pub fn trim_inbound_idc_outcomes(&mut self, sender: Identity, stream_id: u64, ack_prefix: u64) -> Result<()> { + let start = AlgebraicValue::Product(product![IdentityViaU256(sender), stream_id, 0u64]); + let end = AlgebraicValue::Product(product![IdentityViaU256(sender), stream_id, ack_prefix]); + // Bound all three index columns so repeated acknowledgments do not scan + // newer outcomes or copy their payloads. + let rows = self + .iter_by_col_range( + ST_INBOUND_MSG_ID, + col_list![ + StInboundMsgFields::SenderIdentity, + StInboundMsgFields::StreamId, + StInboundMsgFields::Seq + ], + start..=end, + )? + .map(|row| row.pointer()) + .collect::>(); + for row in rows { + self.delete(ST_INBOUND_MSG_ID, row)?; + } + Ok(()) + } + /// Read the stable IDC message id from the user-visible outbox row. fn outbox_message_id(outbox_table_id: TableId, schema: &TableSchema, row_ref: RowRef<'_>) -> Result { let msg_col = schema diff --git a/crates/datastore/src/system_tables/idc.rs b/crates/datastore/src/system_tables/idc.rs index b4c2c3ad8d2..136d25f1630 100644 --- a/crates/datastore/src/system_tables/idc.rs +++ b/crates/datastore/src/system_tables/idc.rs @@ -132,6 +132,7 @@ impl From for ProductValue { pub enum StInboundMsgResultStatus { Ok = 1, Err = 2, + BudgetExceeded = 3, } impl TryFrom for StInboundMsgResultStatus { @@ -140,6 +141,7 @@ impl TryFrom for StInboundMsgResultStatus { match value { 1 => Ok(Self::Ok), 2 => Ok(Self::Err), + 3 => Ok(Self::BudgetExceeded), _ => Err("invalid st_inbound_msg result status"), } } diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 662e1da6ed7..185017da7a1 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -4431,7 +4431,7 @@ mod tests { assert_eq!(user_table_names, expected_table_names); db.with_auto_commit(Workload::ForTests, |tx| { - tx.insert_st_client(Identity::ZERO, ConnectionId::ZERO, "invalid_jwt") + tx.insert_st_client(Identity::ZERO, ConnectionId::ZERO, Some("invalid_jwt")) .unwrap(); Ok::<(), DBError>(()) })?; diff --git a/crates/schema/src/def.rs b/crates/schema/src/def.rs index 7c6494c0ac2..8b59bb469de 100644 --- a/crates/schema/src/def.rs +++ b/crates/schema/src/def.rs @@ -46,12 +46,12 @@ use spacetimedb_lib::db::raw_def::v9::{ }; use spacetimedb_lib::db::view::{extract_view_return_product_type_ref, ViewKind}; use spacetimedb_lib::environment::EnvironmentSchema; -use spacetimedb_lib::{ProductType, RawModuleDef}; +use spacetimedb_lib::{hash_bytes, ProductType, RawModuleDef}; use spacetimedb_primitives::{ ColId, ColList, ColOrCols, ColSet, HttpHandlerId, ProcedureId, ReducerId, TableId, ViewFnPtr, }; use spacetimedb_sats::raw_identifier::RawIdentifier; -use spacetimedb_sats::{AlgebraicType, AlgebraicTypeRef, AlgebraicValue, Typespace}; +use spacetimedb_sats::{bsatn, AlgebraicType, AlgebraicTypeRef, AlgebraicValue, Typespace}; pub mod deserialize; pub mod error; @@ -2440,6 +2440,88 @@ pub struct ReducerDef { pub err_return_type: AlgebraicType, } +impl ReducerDef { + /// Parameter types with schema names. + pub fn params(&self) -> ProductType { + let mut params = self.params.clone(); + for (param, (name, _)) in params.elements.iter_mut().zip(self.params_for_generate.elements.iter()) { + param.name = Some(name.clone().into()); + } + params + } +} + +const REDUCER_SIGNATURE_HASH_DOMAIN: &[u8] = b"spacetimedb:idc:reducer-signature:v1\0"; + +/// Computes the IDC signature hash from parameter names, order, types, and both return types. +/// +/// The encoding is the versioned domain prefix followed by BSATN +/// `(AlgebraicType::Product(params), ok_return_type, err_return_type, reducer_typespace)`, +/// hashed with Keccak-256. Reducer Typespace are a `Vec` in first-use order, +/// traversing parameters, then the success type, then the error type, kind of reducer local +/// typespace. +pub fn reducer_signature_hash( + params: &ProductType, + ok_return_type: &AlgebraicType, + err_return_type: &AlgebraicType, + typespace: &Typespace, +) -> spacetimedb_lib::Hash { + fn normalize( + ty: &mut AlgebraicType, + typespace: &Typespace, + refs: &mut BTreeMap, + reducer_typespace: &mut Typespace, + ) { + match ty { + AlgebraicType::Ref(original) => { + // Number reachable types in first-use order so unrelated types and + // their positions in the module's typespace do not affect the hash. + let local_idx = if let Some(local_idx) = refs.get(original) { + *local_idx + } else { + // Reserve the id before following references so recursive types + // can refer back to this definition while it is being copied. + let local_idx = reducer_typespace.add(AlgebraicType::unit()); + refs.insert(*original, local_idx); + let mut definition = typespace[*original].clone(); + normalize(&mut definition, typespace, refs, reducer_typespace); + reducer_typespace[local_idx] = definition; + local_idx + }; + *original = local_idx; + } + AlgebraicType::Product(product) => { + for field in &mut product.elements { + normalize(&mut field.algebraic_type, typespace, refs, reducer_typespace); + } + } + AlgebraicType::Sum(sum) => { + for variant in &mut sum.variants { + normalize(&mut variant.algebraic_type, typespace, refs, reducer_typespace); + } + } + AlgebraicType::Array(array) => normalize(&mut array.elem_ty, typespace, refs, reducer_typespace), + _ => {} + } + } + + let mut params = AlgebraicType::Product(params.clone()); + let mut ok_return_type = ok_return_type.clone(); + let mut err_return_type = err_return_type.clone(); + let mut refs = BTreeMap::new(); + let mut reducer_typespace = Typespace::default(); + for ty in [&mut params, &mut ok_return_type, &mut err_return_type] { + normalize(ty, typespace, &mut refs, &mut reducer_typespace); + } + let mut bytes = REDUCER_SIGNATURE_HASH_DOMAIN.to_vec(); + bsatn::to_writer( + &mut bytes, + &(params, ok_return_type, err_return_type, reducer_typespace.types), + ) + .expect("validated signature should serialize"); + hash_bytes(&bytes) +} + impl From for RawReducerDefV9 { fn from(val: ReducerDef) -> Self { RawReducerDefV9 { @@ -2933,4 +3015,174 @@ mod tests { assert!(def.reducer_by_name("nonexistent").is_none()); assert!(def.reducer_by_name("nonamespace.auth_verify").is_none()); } + + #[test] + fn reducer_signature_hash_tracks_reachable_types_not_global_ids() { + #[derive(Clone, Copy, Debug)] + enum SignaturePart { + Parameter, + OkReturn, + ErrReturn, + } + + fn hash_for(mut typespace: Typespace, value_type: AlgebraicType, part: SignaturePart) -> spacetimedb_lib::Hash { + // Reserve a slot so Node can refer to itself through its children. + let node = typespace.add(AlgebraicType::unit()); + typespace[node] = AlgebraicType::product([ + ("value", value_type), + ("children", AlgebraicType::array(AlgebraicType::Ref(node))), + ]); + + // Put the recursive type in one signature position at a time, keeping + // the other positions fixed so each traversal path is checked separately. + let node = AlgebraicType::Ref(node); + let (param_type, ok_return_type, err_return_type) = match part { + SignaturePart::Parameter => (node, AlgebraicType::unit(), AlgebraicType::unit()), + SignaturePart::OkReturn => (AlgebraicType::unit(), node, AlgebraicType::unit()), + SignaturePart::ErrReturn => (AlgebraicType::unit(), AlgebraicType::unit(), node), + }; + let params = ProductType::from([("node", param_type)]); + reducer_signature_hash(¶ms, &ok_return_type, &err_return_type, &typespace) + } + + // This unused definition moves Node from Ref(0) to Ref(1) without + // changing its structure or making another type reachable from it. + let mut with_unrelated_type = Typespace::default(); + with_unrelated_type.add(AlgebraicType::product([("other", AlgebraicType::Bool)])); + + for part in [ + SignaturePart::Parameter, + SignaturePart::OkReturn, + SignaturePart::ErrReturn, + ] { + let original = hash_for(Typespace::default(), AlgebraicType::U64, part); + let renumbered = hash_for(with_unrelated_type.clone(), AlgebraicType::U64, part); + let changed = hash_for(Typespace::default(), AlgebraicType::String, part); + + assert_eq!(original, renumbered, "unrelated types must not affect {part:?}"); + assert_ne!(original, changed, "changing Node.value must affect {part:?}"); + } + } + + #[test] + fn reducer_signature_hash_uses_schema_parameter_names() { + use spacetimedb_lib::db::raw_def::v10::RawModuleDefV10Builder; + + // The parameter references Message, which contains an array of MessagePart + // references. Hashing must follow both levels, including the nested fields. + let mut builder = RawModuleDefV10Builder::new(); + let part_type = + AlgebraicType::product([("sender", AlgebraicType::identity()), ("text", AlgebraicType::String)]); + let part_ref = builder.add_algebraic_type([], "MessagePart", part_type.clone(), false); + let message_type = AlgebraicType::product([ + ("label", AlgebraicType::option(AlgebraicType::String)), + ("parts", AlgebraicType::array(AlgebraicType::Ref(part_ref))), + ]); + let message_ref = builder.add_algebraic_type([], "Message", message_type.clone(), false); + builder.add_reducer( + "receive", + ProductType::from([ + ("messageBody", AlgebraicType::Ref(message_ref)), + ("requestId", AlgebraicType::U64), + ]), + ); + // These type references belong to the child, not the parent typespace. + let mut parent = RawModuleDefV10Builder::new(); + parent.add_submodule("child", builder.finish()); + let def: ModuleDef = parent.finish().try_into().expect("valid module"); + let (_, reducer, owning) = def.reducer_by_name_with_module("child.receive").unwrap(); + // Validation applies the module's naming policy to the source parameter names. + let expected_params = ProductType::from([ + ("message_body", AlgebraicType::Ref(message_ref)), + ("request_id", AlgebraicType::U64), + ]); + assert_eq!(reducer.params(), expected_params); + + let hash = reducer_signature_hash( + &reducer.params(), + &reducer.ok_return_type, + &reducer.err_return_type, + owning.typespace(), + ); + // Rebuild just the referenced definitions, without the surrounding module. + let mut typespace: Typespace = [part_type, message_type].into_iter().collect(); + // Source-language spelling is irrelevant when the schema names agree. + assert_eq!( + hash, + reducer_signature_hash( + &expected_params, + &reducer.ok_return_type, + &reducer.err_return_type, + &typespace, + ) + ); + + // Keep the parameter names and reference ids fixed, but change MessagePart.text + // from String to U64. The hash must include the referenced definition's contents. + typespace[part_ref] = + AlgebraicType::product([("sender", AlgebraicType::identity()), ("text", AlgebraicType::U64)]); + assert_ne!( + hash, + reducer_signature_hash( + &expected_params, + &reducer.ok_return_type, + &reducer.err_return_type, + &typespace, + ) + ); + } + + #[test] + fn reducer_signature_hash_tracks_return_types() { + let hash_for = |ok, err| reducer_signature_hash(&ProductType::unit(), &ok, &err, &Typespace::default()); + // With no parameters, only the success and error types contribute to these differences. + let base = hash_for(AlgebraicType::unit(), AlgebraicType::String); + // Changing either return type independently changes the signature. + assert_ne!(base, hash_for(AlgebraicType::U64, AlgebraicType::String)); + assert_ne!(base, hash_for(AlgebraicType::unit(), AlgebraicType::U64)); + // Success and error are distinct positions, even when the same two types are used. + assert_ne!(base, hash_for(AlgebraicType::String, AlgebraicType::unit())); + } + + #[test] + fn reducer_signature_hash_tracks_contract() { + // Keep return types fixed to isolate changes to the parameters. + let hash_for = |params: ProductType| { + reducer_signature_hash( + ¶ms, + &AlgebraicType::unit(), + &AlgebraicType::String, + &Typespace::default(), + ) + }; + let base = hash_for(ProductType::from([ + ("a", AlgebraicType::U64), + ("b", AlgebraicType::String), + ])); + + // BSATN arguments are positional, so reordering parameters changes the contract. + assert_ne!( + base, + hash_for(ProductType::from([ + ("b", AlgebraicType::String), + ("a", AlgebraicType::U64) + ])) + ); + // Names also matter: outbox columns are matched to reducer parameters by name. + assert_ne!( + base, + hash_for(ProductType::from([ + ("renamed", AlgebraicType::U64), + ("b", AlgebraicType::String) + ])) + ); + // Keeping names and order unchanged must not hide a parameter type change. + assert_ne!( + base, + hash_for(ProductType::from([ + ("a", AlgebraicType::U32), + ("b", AlgebraicType::String) + ])) + ); + } } diff --git a/crates/testing/Cargo.toml b/crates/testing/Cargo.toml index 55c5795f533..28d394b19fb 100644 --- a/crates/testing/Cargo.toml +++ b/crates/testing/Cargo.toml @@ -34,6 +34,8 @@ serde.workspace = true futures.workspace = true [dev-dependencies] +axum.workspace = true +tower = { version = "0.5", features = ["util"] } env_logger.workspace = true serial_test.workspace = true diff --git a/crates/testing/src/modules.rs b/crates/testing/src/modules.rs index 98ad9c21f9b..7a89174be9d 100644 --- a/crates/testing/src/modules.rs +++ b/crates/testing/src/modules.rs @@ -69,6 +69,10 @@ pub struct ModuleHandle { } impl ModuleHandle { + pub fn env(&self) -> Arc { + self.env.clone() + } + /// Publish a complete configuration through the standalone control API. pub async fn republish_environment( &self, @@ -348,7 +352,7 @@ impl CompiledModule { with_runtime(move |runtime| { runtime.block_on(async { let module = self.load_module_with_environment(config, None, environment).await; - let env = module.env.clone(); + let env = module.env(); let db_identity = module.db_identity; let routine_result = AssertUnwindSafe(routine(module)).catch_unwind().await.map(drop); finish_module_test(&env, db_identity, routine_result).await; @@ -362,7 +366,7 @@ impl CompiledModule { { with_runtime(move |runtime| { let module = runtime.block_on(async { self.load_module(config, None).await }); - let env = module.env.clone(); + let env = module.env(); let db_identity = module.db_identity; let func_result = std::panic::catch_unwind(AssertUnwindSafe(|| func(runtime, &module))); drop(module); diff --git a/crates/testing/tests/idc_auth.rs b/crates/testing/tests/idc_auth.rs new file mode 100644 index 00000000000..103215da701 --- /dev/null +++ b/crates/testing/tests/idc_auth.rs @@ -0,0 +1,139 @@ +use axum::body::{to_bytes, Body}; +use axum::http::{Request, StatusCode}; +use spacetimedb::auth::idc; +use spacetimedb::db::{Config, Storage}; +use spacetimedb_client_api::routes::database::DatabaseRoutes; +use spacetimedb_client_api::NodeDelegate; +use spacetimedb_lib::sats::product; +use spacetimedb_lib::{bsatn, Hash, Identity}; +use spacetimedb_schema::def::reducer_signature_hash; +use spacetimedb_testing::modules::{CompilationMode, CompiledModule}; +use tower::ServiceExt; + +#[test] +#[serial_test::serial] +fn idc_receiver_runs_connection_lifecycle_without_jwt() { + CompiledModule::compile("idc-auth-test", CompilationMode::Debug).with_module_async( + Config { + storage: Storage::Memory, + page_pool_max_size: None, + }, + |module| async move { + let host = module.client.module(); + let (_, reducer, owning) = host.info.module_def.reducer_by_name_with_module("receive").unwrap(); + let signature_hash = reducer_signature_hash( + &reducer.params(), + &reducer.ok_return_type, + &reducer.err_return_type, + owning.typespace(), + ); + let app = DatabaseRoutes::default() + .into_router(module.env()) + .with_state(module.env()); + let clients_before = host.relational_db().connected_clients().unwrap(); + + let call_idc = + |sender: Identity, seq: u64, fail: bool, hash: Hash, status: StatusCode, error: &'static str| { + let app = app.clone(); + let env = module.env(); + let db_identity = module.db_identity; + async move { + let token = idc::sign(env.jwt_auth_provider(), sender, db_identity).unwrap(); + let request = Request::builder() + .method("POST") + .uri(format!( + "/{}/call-from-database/receive?stream_id=1&seq={seq}&signature_hash={}", + db_identity.to_hex(), + hash.to_hex(), + )) + .header("Authorization", format!("Bearer {token}")) + .header("Content-Type", "application/octet-stream") + .body(Body::from(bsatn::to_vec(&product![fail]).unwrap())) + .unwrap(); + let response = app.oneshot(request).await.unwrap(); + let actual_status = response.status(); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8_lossy(&body); + assert_eq!(actual_status, status, "{body}"); + assert!(body.contains(error), "{body}"); + } + }; + + let assert_counts = async |connected: u64, received: u64, disconnected: u64| { + let app = app.clone(); + let db_identity = module.db_identity; + // Use a normal HTTP call to verify ordinary clients still receive a JWT. + let request = Request::builder() + .method("POST") + .uri(format!("/{}/call/assert_counts", db_identity.to_hex())) + .header("Content-Type", "application/json") + .body(Body::from(format!("[{connected},{received},{disconnected}]"))) + .unwrap(); + let response = app.oneshot(request).await.unwrap(); + let status = response.status(); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(status, StatusCode::OK, "{}", String::from_utf8_lossy(&body)); + assert_eq!(host.relational_db().connected_clients().unwrap(), clients_before); + }; + + // Rejection in client_connected must prevent delivery. Accepted requests + // must disconnect after success, replay, or failure, without exposing a JWT. + // Untrusted senders are rejected before the target reducer runs. + call_idc( + Identity::ZERO, + 1, + false, + signature_hash, + StatusCode::FORBIDDEN, + "untrusted database", + ) + .await; + assert_counts(0, 0, 0).await; + + // A trusted sender runs the reducer and disconnects cleanly afterward. + call_idc(Identity::ONE, 1, false, signature_hash, StatusCode::OK, "").await; + assert_counts(1, 1, 1).await; + + // A duplicate delivery replays the stored outcome without rerunning the reducer. + call_idc(Identity::ONE, 1, false, signature_hash, StatusCode::OK, "").await; + assert_counts(2, 1, 2).await; + + // A reducer-level rejection is terminal but still runs the connection lifecycle. + call_idc( + Identity::ONE, + 2, + true, + signature_hash, + StatusCode::UNPROCESSABLE_ENTITY, + "delivery rejected", + ) + .await; + assert_counts(3, 1, 3).await; + + // Changing the arguments cannot replace an already-recorded failure. + // The reducer would succeed now, but this sequence must not run again. + call_idc( + Identity::ONE, + 2, + false, + signature_hash, + StatusCode::UNPROCESSABLE_ENTITY, + "delivery rejected", + ) + .await; + assert_counts(4, 1, 4).await; + + // A signature mismatch is recorded as a terminal outcome without reducer execution. + call_idc( + Identity::ONE, + 3, + false, + spacetimedb_lib::Hash::from_byte_array([0; 32]), + StatusCode::UNPROCESSABLE_ENTITY, + "", + ) + .await; + assert_counts(5, 1, 5).await; + }, + ); +} diff --git a/crates/testing/tests/idc_signature.rs b/crates/testing/tests/idc_signature.rs new file mode 100644 index 00000000000..dcaa36038c6 --- /dev/null +++ b/crates/testing/tests/idc_signature.rs @@ -0,0 +1,214 @@ +//! Signature failures are durable decisions, not transport failures. Retries must +//! replay those decisions even when the reducer schema has changed. +use spacetimedb::host::{FunctionArgs, IdcDeliveryFailure, IdcReducerCallOutcome}; +use spacetimedb_lib::{hash_bytes, ConnectionId, Hash, Identity}; +use spacetimedb_schema::def::reducer_signature_hash; +use spacetimedb_testing::modules::{CompilationMode, CompiledModule, DEFAULT_CONFIG, IN_MEMORY_CONFIG}; + +/// A duplicate must replay, not execute and fail while inserting its outcome. +#[test] +#[serial_test::serial] +fn concurrent_idc_deliveries_commit_once() { + CompiledModule::compile("module-test", CompilationMode::Debug).with_module_async( + IN_MEMORY_CONFIG, + |handle| async move { + let host = handle.client.module(); + let (_, reducer, _) = host.info.module_def.reducer_by_name_with_module("add").unwrap(); + let hash = host.info.reducer_signature_hash(&reducer.name); + let args = spacetimedb_lib::bsatn::to_vec(&spacetimedb_lib::sats::product!["idc duplicate", 42u8]).unwrap(); + let calls = (0..8).map(|_| { + host.call_idc_reducer( + Identity::ZERO, + ConnectionId::from_u128(1), + "add", + FunctionArgs::Bsatn(args.clone().into()), + 1, + 1, + 0, + hash, + ) + }); + let outcomes = tokio::time::timeout(std::time::Duration::from_secs(10), futures::future::join_all(calls)) + .await + .unwrap(); + let mut applied = 0; + for outcome in outcomes { + match outcome.unwrap() { + IdcReducerCallOutcome::Recorded(result) => { + assert!(result.result.outcome.into_result().is_ok()); + applied += 1; + } + IdcReducerCallOutcome::Replayed(result) => assert!(result.result_payload.is_empty()), + IdcReducerCallOutcome::AlreadyAcked => panic!("no request acknowledged this message"), + } + } + assert_eq!(applied, 1); + + // add allocates a new row ID each time, so duplicate committed effects + // would leave multiple rows even though the arguments are identical. + let rows = spacetimedb::sql::execute::run( + host.relational_db().clone(), + "SELECT * FROM person WHERE name = 'idc duplicate'".into(), + spacetimedb_lib::identity::AuthCtx::for_current(host.info.owner_identity), + Some(host.info.subscriptions.clone()), + Some(host.clone()), + &mut vec![], + ) + .await + .unwrap(); + assert_eq!(rows.rows.len(), 1); + }, + ); +} + +#[test] +#[serial_test::serial] +fn terminal_signature_errors_are_replayed_and_do_not_stall_the_stream() { + CompiledModule::compile("module-test", CompilationMode::Debug).with_module_async( + DEFAULT_CONFIG, + |handle| async move { + let host = handle.client.module(); + let connection = ConnectionId::from_u128(1); + let outdated = hash_bytes(b"outdated signature"); + let call = |name: &'static str, seq, hash: Hash| { + let host = host.clone(); + async move { + host.call_idc_reducer( + Identity::ZERO, + connection, + name, + FunctionArgs::Bsatn(vec![255].into()), + 42, + seq, + 0, + hash, + ) + .await + } + }; + + // A later sequence cannot skip over the first undecided message, + // even when that later message would fail signature validation. + assert!(matches!( + call("add", 2, outdated).await, + Err(IdcDeliveryFailure::OutOfOrder { expected: 1, actual: 2 }) + )); + + // The first message has the wrong signature, so the receiver records a + // terminal outcome for seq 1 instead of retrying it forever. + let IdcReducerCallOutcome::Recorded(failed) = call("add", 1, outdated).await.unwrap() else { + panic!("new message must record a terminal decision"); + }; + let error = failed.result.outcome.into_result().unwrap_err().to_string(); + assert!(error.contains("signature mismatch"), "{error}"); + let offset = failed.tx_offset.await.unwrap(); + + // Retrying seq 1 replays the stored signature error despite a failed reducer lookup. + // A changed route/schema and invalid arguments cannot replace the answer. + let IdcReducerCallOutcome::Replayed(replayed) = + call("removed_reducer", 1, hash_bytes(b"different signature")) + .await + .unwrap() + else { + panic!("decided message must replay despite a failed reducer lookup"); + }; + assert_eq!(&replayed.result_payload[..], error.as_bytes()); + assert!( + replayed.tx_offset >= offset, + "replay durability must cover the original decision" + ); + + // The next sequence can now advance and records its own terminal error. + let IdcReducerCallOutcome::Recorded(missing) = call("removed_reducer", 2, outdated).await.unwrap() else { + panic!("missing reducer must produce a terminal decision"); + }; + assert!(missing + .result + .outcome + .into_result() + .unwrap_err() + .to_string() + .contains("IDC reducer does not exist: removed_reducer")); + + let (_, reducer, owning) = host.info.module_def.reducer_by_name_with_module("add").unwrap(); + let hash = reducer_signature_hash( + &reducer.params(), + &reducer.ok_return_type, + &reducer.err_return_type, + owning.typespace(), + ); + let IdcReducerCallOutcome::Recorded(invalid_args) = call("add", 3, hash).await.unwrap() else { + panic!("invalid arguments must produce a terminal decision"); + }; + let error = invalid_args.result.outcome.into_result().unwrap_err().to_string(); + assert!(error.contains("invalid arguments for reducer add:"), "{error}"); + assert!(!error.contains("signature mismatch"), "{error}"); + + // Matching hashes do not make malformed arguments valid. Keep the + // decoding error so retries report the same failure. + let IdcReducerCallOutcome::Replayed(replayed) = call("add", 3, hash).await.unwrap() else { + panic!("invalid arguments must replay their recorded error"); + }; + assert_eq!(&replayed.result_payload[..], error.as_bytes()); + + // Terminal errors for earlier messages must not stall a later valid delivery. + let (_, reducer, owning) = host.info.module_def.reducer_by_name_with_module("say_hello").unwrap(); + let hash = reducer_signature_hash( + &reducer.params(), + &reducer.ok_return_type, + &reducer.err_return_type, + owning.typespace(), + ); + let IdcReducerCallOutcome::Recorded(valid) = host + .call_idc_reducer( + Identity::ZERO, + connection, + "say_hello", + FunctionArgs::Nullary, + 42, + 4, + 3, + hash, + ) + .await + .unwrap() + else { + panic!("valid message following terminal errors must run"); + }; + assert!(valid.result.outcome.into_result().is_ok()); + let offset = valid.tx_offset.await.unwrap(); + + // Successful outcomes are replayed too, and acked outcomes are not resurrected. + let IdcReducerCallOutcome::Replayed(replayed) = call("removed_reducer", 4, outdated).await.unwrap() else { + panic!("successful decisions must also survive schema changes"); + }; + assert!(replayed.result_payload.is_empty()); + assert!(replayed.tx_offset >= offset); + assert!(matches!( + call("removed_reducer", 1, outdated).await.unwrap(), + IdcReducerCallOutcome::AlreadyAcked + )); + + // Cleanup commits separately even when the accompanying delivery is + // out of order and its sequence-check transaction is rolled back. + assert!(matches!( + host.call_idc_reducer( + Identity::ZERO, + connection, + "say_hello", + FunctionArgs::Nullary, + 42, + 6, + 4, + hash, + ) + .await, + Err(IdcDeliveryFailure::OutOfOrder { expected: 5, actual: 6 }) + )); + assert!(matches!( + call("removed_reducer", 4, outdated).await.unwrap(), + IdcReducerCallOutcome::AlreadyAcked + )); + }, + ); +} diff --git a/modules/idc-auth-test/Cargo.toml b/modules/idc-auth-test/Cargo.toml new file mode 100644 index 00000000000..0299ebd1036 --- /dev/null +++ b/modules/idc-auth-test/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "idc-auth-test" +version = "0.0.0" +edition.workspace = true +license-file = "../../LICENSE.txt" +publish = false + +[lib] +crate-type = ["cdylib"] + +[dependencies] +spacetimedb.workspace = true diff --git a/modules/idc-auth-test/src/lib.rs b/modules/idc-auth-test/src/lib.rs new file mode 100644 index 00000000000..76a249be2be --- /dev/null +++ b/modules/idc-auth-test/src/lib.rs @@ -0,0 +1,85 @@ +use spacetimedb::{ConnectionId, Identity, ReducerContext, Table}; + +#[spacetimedb::table(accessor = idc_event)] +pub struct IdcEvent { + #[primary_key] + #[auto_inc] + id: u64, + connection_id: ConnectionId, + phase: String, +} + +fn record_event(ctx: &ReducerContext, phase: &str) { + ctx.db.idc_event().insert(IdcEvent { + id: 0, + connection_id: ctx.connection_id().unwrap(), + phase: phase.into(), + }); +} + +fn assert_connected(ctx: &ReducerContext) { + let connection_id = ctx.connection_id().unwrap(); + assert!(ctx + .db + .idc_event() + .iter() + .any(|event| { event.connection_id == connection_id && event.phase == "connected" })); +} + +fn check_auth(ctx: &ReducerContext) -> Result<(), String> { + if ctx.sender() != Identity::ONE { + return Err("untrusted database".into()); + } + assert!(ctx.connection_id().is_some()); + assert!(!ctx.sender_auth().is_internal()); + assert!(ctx.sender_auth().jwt().is_none()); + Ok(()) +} + +#[spacetimedb::reducer(client_connected)] +pub fn connected(ctx: &ReducerContext) -> Result<(), String> { + // Ordinary clients still authenticate with a module-visible JWT. + if ctx.sender_auth().jwt().is_some() { + return Ok(()); + } + check_auth(ctx)?; + record_event(ctx, "connected"); + Ok(()) +} + +#[spacetimedb::reducer(client_disconnected)] +pub fn disconnected(ctx: &ReducerContext) -> Result<(), String> { + if ctx.sender_auth().jwt().is_some() { + return Ok(()); + } + check_auth(ctx)?; + assert_connected(ctx); + record_event(ctx, "disconnected"); + Ok(()) +} + +#[spacetimedb::reducer] +pub fn receive(ctx: &ReducerContext, fail: bool) -> Result<(), String> { + check_auth(ctx)?; + assert_connected(ctx); + if fail { + return Err("delivery rejected".into()); + } + record_event(ctx, "received"); + Ok(()) +} + +#[spacetimedb::reducer] +pub fn assert_counts(ctx: &ReducerContext, connected: u64, received: u64, disconnected: u64) { + assert!(ctx.sender_auth().jwt().is_some()); + for (phase, expected) in [ + ("connected", connected), + ("received", received), + ("disconnected", disconnected), + ] { + assert_eq!( + ctx.db.idc_event().iter().filter(|event| event.phase == phase).count() as u64, + expected + ); + } +}