From 73403fcd88a92e8fd07a1028602cd999b615c38d Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Mon, 24 Aug 2026 00:06:20 -0500 Subject: [PATCH 1/4] fix(celld): make command lifecycle durable and fast Keep long-running consumers alive across idle polls, route every Todo transition to one cell, persist fenced cell command replays, enforce CellByKey row policies, and run the real browser lifecycle in celld CI. Refs [[incidents/pr-206-e2e-ui-command-latency-1]] --- .github/workflows/integration-celld.yaml | 124 +++++- src/bus/nats.rs | 45 ++- src/bus/nats_bus.rs | 12 +- src/command_dispatch/host.rs | 54 ++- src/command_ledger/record.rs | 155 +++++++- src/graphql/compile/mod.rs | 1 + src/graphql/compile/projection.rs | 344 +++++++++++++++- src/graphql/mod.rs | 1 - src/graphql/read_store.rs | 41 +- src/graphql/schema.rs | 17 +- src/in_memory_repo/repository.rs | 30 ++ src/microsvc/cell_host/causal.rs | 367 ++++++++++++++++++ src/microsvc/cell_host/cell.rs | 33 +- src/microsvc/cell_host/command.rs | 110 ++++-- src/microsvc/cell_host/mod.rs | 9 +- src/microsvc/cell_host/store.rs | 44 ++- src/microsvc/cell_host/tests.rs | 120 +++++- src/microsvc/mod.rs | 4 +- src/microsvc/service/routes.rs | 268 ++++++++++++- src/microsvc/workers.rs | 23 +- tests/celld/main.rs | 184 ++++++--- tests/celld/worker/src/lib.rs | 300 +++++++++----- .../crates/graphql-service/src/host.rs | 12 +- .../src/handlers/ingestors/zitadel/scrape.rs | 52 ++- .../src/handlers/ingestors/zitadel_scrape.rs | 4 +- .../e2e-celld/crates/todo-service/src/host.rs | 75 +++- .../e2e-celld/crates/todo-service/src/lib.rs | 4 +- tests/e2e-ui/crates/service/src/host.rs | 8 +- 28 files changed, 2189 insertions(+), 252 deletions(-) create mode 100644 src/microsvc/cell_host/causal.rs diff --git a/.github/workflows/integration-celld.yaml b/.github/workflows/integration-celld.yaml index 798be1eb8..743de603b 100644 --- a/.github/workflows/integration-celld.yaml +++ b/.github/workflows/integration-celld.yaml @@ -5,7 +5,10 @@ name: celld (live + e2e-celld) # # Local parity: # make -C tests/e2e-celld test -# make -C tests/e2e-ui up-celld-nats && make -C tests/e2e-ui test-celld +# make -C tests/e2e-ui up && make -C tests/e2e-ui up-celld-nats +# WATCH=0 WATCH_WORKER=0 make -C tests/e2e-celld run +# E2E_UI_ORIGIN=http://localhost:5180 npx --prefix tests/e2e-ui playwright test \ +# todos.user.spec.ts chat.user.spec.ts --project chromium-user # # Default `cargo test` (quality) still runs fixture-only celld checks and # skips live HTTP unless CELLD_URL is set. This job sets CELLD_URL / NATS_URL. @@ -14,8 +17,9 @@ on: env: CARGO_TERM_COLOR: always - CELLD_HTTP_PORT: "18080" - CELLD_URL: http://127.0.0.1:18080 + # Zitadel owns :18080 in the browser topology. + CELLD_HTTP_PORT: "18880" + CELLD_URL: http://127.0.0.1:18880 NATS_PORT: "14222" NATS_URL: nats://127.0.0.1:14222 AZURE_STORAGE_USE_EMULATOR: "true" @@ -67,6 +71,14 @@ jobs: - uses: actions/setup-node@v4 with: node-version: "22" + cache: npm + cache-dependency-path: | + js/package-lock.json + tests/e2e-ui/package-lock.json + tests/e2e-ui/ui/package-lock.json + + - name: Install host tools + run: sudo apt-get update && sudo apt-get install -y jq openssl curl - name: Install esbuild run: npm install -g esbuild @@ -81,6 +93,9 @@ jobs: curl -fsSL https://celld.dev/install.sh | sh echo "$HOME/.local/bin" >> "$GITHUB_PATH" + - name: Bring up Postgres + Zitadel and bootstrap OIDC + run: make -C tests/e2e-ui up + - name: Bring up Azurite + celld + NATS run: | command -v celld @@ -90,6 +105,95 @@ jobs: - name: Live celld HTTP + NATS profile tests run: make -C tests/e2e-ui test-celld + - name: Build e2e-celld API and UI + run: | + cargo build --manifest-path tests/e2e-celld/Cargo.toml \ + -p e2e-celld-runner --bin e2e-celld + make -C tests/e2e-ui ui-install + npm install --prefix tests/e2e-ui + + - name: Start e2e-celld API + UI + run: | + set -euo pipefail + set -a + # shellcheck disable=SC1091 + . tests/e2e-ui/e2e-ui.env + set +a + export BIND=0.0.0.0:8791 + export CELLD_URL=http://127.0.0.1:18880 + export NATS_URL=nats://127.0.0.1:14222 + export AUTH_URL=http://localhost:5180 + export AUTH_USE_SECURE_COOKIES=false + export AUTH_TRUST_HOST=true + + tests/e2e-celld/target/debug/e2e-celld \ + > tests/e2e-celld/.ci-runner.log 2>&1 & + echo $! > tests/e2e-celld/.ci-runner.pid + + ok=0 + for i in $(seq 1 120); do + code=$(curl -s -o /dev/null -w '%{http_code}' -X POST \ + "http://127.0.0.1:8791/graphql" \ + -H 'content-type: application/json' \ + -d '{"query":"{ __typename }"}' 2>/dev/null || echo 000) + if [ "$code" = "200" ] || [ "$code" = "401" ]; then ok=1; break; fi + sleep 0.5 + done + if [ "$ok" != "1" ]; then + echo "e2e-celld API failed to become ready" + tail -120 tests/e2e-celld/.ci-runner.log + exit 1 + fi + + cd tests/e2e-ui/ui + PUBLIC_E2E_PROFILE=celld-nats \ + E2E_API_ORIGIN=http://127.0.0.1:8791 \ + npm run dev -- --host localhost --port 5180 \ + > ../.ci-celld-ui.log 2>&1 & + echo $! > ../.ci-celld-ui.pid + cd ../../.. + + ok=0 + for i in $(seq 1 60); do + code=$(curl -s -o /dev/null -w '%{http_code}' \ + "http://localhost:5180/" 2>/dev/null || echo 000) + if [ "$code" = "200" ] || [ "$code" = "302" ] || [ "$code" = "303" ]; then + ok=1 + break + fi + sleep 0.5 + done + if [ "$ok" != "1" ]; then + echo "e2e-celld UI failed to become ready (last HTTP $code)" + tail -100 tests/e2e-ui/.ci-celld-ui.log + exit 1 + fi + + - name: Install Playwright + Chromium + working-directory: tests/e2e-ui + run: npx playwright install chromium --with-deps + + - name: Todo + Chat browser lifecycle through celld + working-directory: tests/e2e-ui + run: >- + npx playwright test todos.user.spec.ts chat.user.spec.ts + --project chromium-user + env: + E2E_UI_ORIGIN: http://localhost:5180 + E2E_API_ORIGIN: http://127.0.0.1:8791 + CI: true + + - name: Upload celld Playwright report + if: failure() + uses: actions/upload-artifact@v4 + with: + name: celld-playwright-report + path: | + tests/e2e-ui/playwright-report + tests/e2e-ui/test-results + if-no-files-found: ignore + retention-days: 7 + - name: Dump logs on failure if: failure() run: | @@ -99,9 +203,23 @@ jobs: echo '=== NATS profile compose ===' docker compose -f tests/e2e-ui/celld-nats-profile/docker-compose.yml ps -a || true docker compose -f tests/e2e-ui/celld-nats-profile/docker-compose.yml logs --tail=80 || true + echo '=== e2e-celld API ===' + tail -180 tests/e2e-celld/.ci-runner.log || true + echo '=== e2e-celld UI ===' + tail -100 tests/e2e-ui/.ci-celld-ui.log || true + echo '=== Postgres + Zitadel ===' + docker compose -f tests/e2e-ui/docker/docker-compose.yml ps -a || true + docker compose -f tests/e2e-ui/docker/docker-compose.yml logs --tail=100 || true - name: Tear down if: always() run: | + [ -f tests/e2e-ui/.ci-celld-ui.pid ] && \ + kill "$(cat tests/e2e-ui/.ci-celld-ui.pid)" 2>/dev/null || true + [ -f tests/e2e-celld/.ci-runner.pid ] && \ + kill "$(cat tests/e2e-celld/.ci-runner.pid)" 2>/dev/null || true + lsof -ti:5180 2>/dev/null | xargs -r kill -9 2>/dev/null || true + lsof -ti:8791 2>/dev/null | xargs -r kill -9 2>/dev/null || true make -C tests/e2e-ui down-celld-nats || true make -C tests/e2e-ui down-celld || true + docker compose -f tests/e2e-ui/docker/docker-compose.yml down -v || true diff --git a/src/bus/nats.rs b/src/bus/nats.rs index 6362a2c5a..554c2d349 100644 --- a/src/bus/nats.rs +++ b/src/bus/nats.rs @@ -105,6 +105,7 @@ pub struct NatsJetStreamSource { consumer: Consumer, fetch_timeout: Duration, strip_prefix: Option, + idle_poll: Duration, } impl NatsJetStreamSource { @@ -114,6 +115,7 @@ impl NatsJetStreamSource { consumer, fetch_timeout: Duration::from_millis(500), strip_prefix: None, + idle_poll: Duration::ZERO, } } @@ -134,6 +136,12 @@ impl NatsJetStreamSource { self } + /// Keep `recv` retrying after an empty fetch instead of draining to idle. + pub fn with_idle_poll(mut self, idle_poll: Duration) -> Self { + self.idle_poll = idle_poll; + self + } + /// Connect to a NATS server URL, then create/open the stream + consumer. pub async fn connect( url: &str, @@ -188,22 +196,27 @@ impl MessageSource for NatsJetStreamSource { } async fn recv(&mut self) -> Result, TransportError> { - let mut batch = self - .consumer - .batch() - .max_messages(1) - .expires(self.fetch_timeout) - .messages() - .await - .map_err(|err| retryable("nats fetch", err))?; - - match batch.next().await { - Some(Ok(message)) => Ok(Some(NatsReceived::from_jetstream( - message, - self.strip_prefix.as_deref(), - ))), - Some(Err(err)) => Err(retryable("nats batch message", err)), - None => Ok(None), + loop { + let mut batch = self + .consumer + .batch() + .max_messages(1) + .expires(self.fetch_timeout) + .messages() + .await + .map_err(|err| retryable("nats fetch", err))?; + + match batch.next().await { + Some(Ok(message)) => { + return Ok(Some(NatsReceived::from_jetstream( + message, + self.strip_prefix.as_deref(), + ))) + } + Some(Err(err)) => return Err(retryable("nats batch message", err)), + None if self.idle_poll.is_zero() => return Ok(None), + None => continue, + } } } } diff --git a/src/bus/nats_bus.rs b/src/bus/nats_bus.rs index c9ae5d5c5..020ab9811 100644 --- a/src/bus/nats_bus.rs +++ b/src/bus/nats_bus.rs @@ -44,6 +44,7 @@ pub struct NatsBus { evt_publisher: Arc, topology: BusTopologyConfig, fetch_timeout: Duration, + idle_poll: Duration, } /// Awaitable builder returned by [`NatsBus::connect`]. @@ -107,6 +108,7 @@ impl NatsBus { evt_publisher: Arc::new(evt_publisher), topology: BusTopologyConfig::default(), fetch_timeout: DEFAULT_FETCH_TIMEOUT, + idle_poll: Duration::ZERO, } } @@ -162,6 +164,13 @@ impl NatsBus { self } + /// Keep `listen`/`subscribe` running after an empty JetStream fetch. + /// Drain-to-idle is for tests; long-running hosts must set this. + pub fn with_idle_poll(mut self, idle_poll: Duration) -> Self { + self.idle_poll = idle_poll; + self + } + /// Sanitize the group into a valid NATS consumer-name token. Consumer names /// cannot contain `.`, `*`, `>`, or whitespace, so map them to `_`. fn durable_base(group: &str) -> String { @@ -231,7 +240,8 @@ impl NatsBus { .map_err(|err| retryable("nats get_or_create_consumer", err))?; Ok(NatsJetStreamSource::new(consumer) .with_fetch_timeout(self.fetch_timeout) - .with_strip_prefix(strip_prefix)) + .with_strip_prefix(strip_prefix) + .with_idle_poll(self.idle_poll)) } /// Shared consume path for `listen` (commands) and `subscribe` (events): diff --git a/src/command_dispatch/host.rs b/src/command_dispatch/host.rs index 88172422a..ad2447728 100644 --- a/src/command_dispatch/host.rs +++ b/src/command_dispatch/host.rs @@ -6,6 +6,7 @@ use std::sync::Arc; use crate::graphql::identity::VerifiedPrincipal; use crate::graphql::protocol::ProtocolResponseAccumulator; +use crate::microsvc::cell_host::{CELL_PRINCIPAL_PARTITION_HEADER, CELL_SERVICE_ID_HEADER}; use crate::microsvc::{ CausalCommandPublicStatus, CausalDispatchError, CausalDispatchResult, Service, Session, ROLE_KEY, USER_ID_KEY, @@ -157,19 +158,58 @@ impl HttpCommandHost { input: Value, session: &Session, ) -> Result<(u16, Value), CausalDispatchError> { - let mut request = self - .client - .post(format!("{}/{command}", self.base)) - .json(&serde_json::json!({ - "commandId": command_id, - "input": input, - })); + self.post_wait_path_inner(command, command_id, input, session, None) + .await + } + + /// POST a cell wait-path command with identity derived by the verified + /// GraphQL host. These headers are part of the trusted internal boundary, + /// not values copied from public request headers or command input. + pub async fn post_cell_wait_path( + &self, + command: &str, + command_id: &str, + input: Value, + session: &Session, + service_id: &str, + principal_partition: &str, + ) -> Result<(u16, Value), CausalDispatchError> { + self.post_wait_path_inner( + command, + command_id, + input, + session, + Some((service_id, principal_partition)), + ) + .await + } + + async fn post_wait_path_inner( + &self, + command: &str, + command_id: &str, + input: Value, + session: &Session, + cell_identity: Option<(&str, &str)>, + ) -> Result<(u16, Value), CausalDispatchError> { + let mut request = + self.client + .post(format!("{}/{command}", self.base)) + .json(&serde_json::json!({ + "commandId": command_id, + "input": input, + })); if let Some(user) = session.user_id() { request = request.header(USER_ID_KEY, user); } if let Some(roles) = session.get(ROLE_KEY) { request = request.header(ROLE_KEY, roles); } + if let Some((service_id, principal_partition)) = cell_identity { + request = request + .header(CELL_SERVICE_ID_HEADER, service_id) + .header(CELL_PRINCIPAL_PARTITION_HEADER, principal_partition); + } let response = request.send().await.map_err(|err| { CausalDispatchError::Internal(format!("wait-path HTTP failed: {err}")) })?; diff --git a/src/command_ledger/record.rs b/src/command_ledger/record.rs index 6f62f15cd..8ccc9629c 100644 --- a/src/command_ledger/record.rs +++ b/src/command_ledger/record.rs @@ -1,7 +1,8 @@ -use std::time::{Duration, SystemTime}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use base64::Engine as _; +use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::projection_protocol::{ResolvedProjectionObligation, SameTransactionProjectionEvidence}; @@ -62,6 +63,113 @@ impl CommandLedgerRecord { }) } + /// Stable opaque key used by a cell host's private SQLite table. + pub(crate) fn durable_cell_key(&self) -> String { + let material = format!( + "{}\0{}\0{}", + self.key.service_id(), + self.key.principal_partition(), + self.key.command_id() + ); + format!("v1.{}", URL_SAFE_NO_PAD.encode(material)) + } + + /// Versioned cell-storage representation. This is deliberately separate + /// from the public command receipt and retains the complete fenced row. + pub(crate) fn durable_cell_json(&self) -> Result { + self.validate_stored_shape()?; + let wire = DurableCellCommandRecordV1 { + version: 1, + service_id: self.key.service_id().to_string(), + principal_partition: self.key.principal_partition().to_string(), + command_id: self.key.command_id().to_string(), + command_name: self.command_name.clone(), + contract_fingerprint: self.contract_fingerprint.as_bytes().to_vec(), + input_hash: self.input_hash.as_bytes().to_vec(), + state: self.state.as_str().to_string(), + causation_id: self.causation_id.as_str().to_string(), + attempt_token: self + .attempt_token + .as_ref() + .map(|token| token.as_str().to_string()), + attempt_number: self.attempt_number, + lease_expires_at_ms: self + .lease_expires_at + .map(system_time_to_unix_millis) + .transpose()?, + outcome_json: self.outcome_json.clone(), + created_at_ms: system_time_to_unix_millis(self.created_at)?, + updated_at_ms: system_time_to_unix_millis(self.updated_at)?, + completed_at_ms: self + .completed_at + .map(system_time_to_unix_millis) + .transpose()?, + retention_expires_at_ms: system_time_to_unix_millis(self.retention_expires_at)?, + compacted_at_ms: self + .compacted_at + .map(system_time_to_unix_millis) + .transpose()?, + }; + serde_json::to_string(&wire).map_err(|error| { + CommandLedgerError::Corrupt(format!( + "cell command ledger row could not be encoded: {error}" + )) + }) + } + + pub(crate) fn from_durable_cell_json(body: &str) -> Result { + let wire: DurableCellCommandRecordV1 = serde_json::from_str(body).map_err(|error| { + CommandLedgerError::Corrupt(format!("cell command ledger row is invalid JSON: {error}")) + })?; + if wire.version != 1 { + return Err(CommandLedgerError::Corrupt(format!( + "cell command ledger row version `{}` is unsupported", + wire.version + ))); + } + let key = CommandLedgerKey::new( + wire.service_id, + super::PrincipalPartitionId::new(wire.principal_partition)?, + super::CommandId::parse(wire.command_id)?, + )?; + let record = Self { + key, + command_name: wire.command_name, + contract_fingerprint: CommandContractFingerprint::try_from_slice( + &wire.contract_fingerprint, + )?, + input_hash: CanonicalInputHash::try_from_slice(&wire.input_hash)?, + state: CommandLedgerState::parse(&wire.state)?, + causation_id: CausationId::parse_stored(wire.causation_id)?, + attempt_token: wire + .attempt_token + .map(AttemptToken::parse_stored) + .transpose()?, + attempt_number: wire.attempt_number, + lease_expires_at: wire + .lease_expires_at_ms + .map(system_time_from_unix_millis) + .transpose()?, + outcome_json: wire.outcome_json, + created_at: system_time_from_unix_millis(wire.created_at_ms)?, + updated_at: system_time_from_unix_millis(wire.updated_at_ms)?, + completed_at: wire + .completed_at_ms + .map(system_time_from_unix_millis) + .transpose()?, + retention_expires_at: system_time_from_unix_millis(wire.retention_expires_at_ms)?, + compacted_at: wire + .compacted_at_ms + .map(system_time_from_unix_millis) + .transpose()?, + }; + record.validate_stored_shape()?; + if record.state.is_replayable() { + record.replay()?; + } + Ok(record) + } + pub(crate) fn acquired_attempt(&self) -> Result { let token = self.attempt_token.as_ref().ok_or_else(|| { CommandLedgerError::Corrupt(format!( @@ -499,6 +607,51 @@ impl CommandLedgerRecord { } } +#[derive(Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct DurableCellCommandRecordV1 { + version: u16, + service_id: String, + principal_partition: String, + command_id: String, + command_name: String, + contract_fingerprint: Vec, + input_hash: Vec, + state: String, + causation_id: String, + attempt_token: Option, + attempt_number: u64, + lease_expires_at_ms: Option, + outcome_json: Option, + created_at_ms: u64, + updated_at_ms: u64, + completed_at_ms: Option, + retention_expires_at_ms: u64, + compacted_at_ms: Option, +} + +fn system_time_to_unix_millis(value: SystemTime) -> Result { + let millis = value + .duration_since(UNIX_EPOCH) + .map_err(|_| { + CommandLedgerError::Corrupt( + "cell command ledger timestamp precedes the Unix epoch".into(), + ) + })? + .as_millis(); + u64::try_from(millis).map_err(|_| { + CommandLedgerError::Corrupt("cell command ledger timestamp exceeds u64 millis".into()) + }) +} + +fn system_time_from_unix_millis(value: u64) -> Result { + UNIX_EPOCH + .checked_add(Duration::from_millis(value)) + .ok_or_else(|| { + CommandLedgerError::Corrupt("cell command ledger timestamp is out of range".into()) + }) +} + fn checked_deadline( now: SystemTime, duration: Duration, diff --git a/src/graphql/compile/mod.rs b/src/graphql/compile/mod.rs index 884a0ef2c..d4ce551f1 100644 --- a/src/graphql/compile/mod.rs +++ b/src/graphql/compile/mod.rs @@ -37,3 +37,4 @@ pub(crate) use dialect::{ }; #[allow(unused_imports)] pub(crate) use evidence::{ExtractedQueryEvidence, QueryRecordEvidence, QueryResponsePathSegment}; +pub(crate) use projection::cell_row_matches; diff --git a/src/graphql/compile/projection.rs b/src/graphql/compile/projection.rs index bcdd9101d..d277d743c 100644 --- a/src/graphql/compile/projection.rs +++ b/src/graphql/compile/projection.rs @@ -1,3 +1,4 @@ +use std::cmp::Ordering; use std::collections::BTreeMap; use async_graphql::Value; @@ -7,9 +8,10 @@ use crate::microsvc::Session; use crate::table::{ColumnType, TableSchema}; use super::super::engine::EngineInner; +use super::super::filter::{CmpOp, FilterExpr, LitValue, Operand}; use super::super::naming::{is_valid_graphql_name, scalar_type_name}; use super::super::permissions::ReadPermission; -use super::binds::{value_to_bind, BindValue}; +use super::binds::{operand_to_bind, value_to_bind, BindValue}; use super::dialect::{placeholder, SqlDialect}; use super::evidence::{ ExtractedQueryEvidence, QueryEvidenceFieldPlan, QueryEvidenceKeyPlan, QueryEvidenceNode, @@ -63,6 +65,8 @@ pub enum QueryPlan { CellByKey { model: String, pk: BTreeMap, + /// Role row policy with every claim resolved from the trusted session. + row_filter: Option, }, } @@ -85,13 +89,15 @@ pub fn compile_query( inner, session, role, model_name, kind, selection, )?)), crate::graphql::read_store::ReadStoreKind::CellByKey => { - compile_cell_by_key(inner, model_name, kind, selection) + compile_cell_by_key(inner, session, role, model_name, kind, selection) } } } fn compile_cell_by_key( inner: &EngineInner, + session: &Session, + role: &str, model_name: &str, kind: RootKind, selection: &SelectionNode, @@ -100,6 +106,11 @@ fn compile_cell_by_key( .catalog .get(model_name) .ok_or_else(|| format!("unknown model `{model_name}`"))?; + let permission = inner + .permissions + .get(&(model_name.to_string(), role.to_string())) + .map(|entry| &entry.permission) + .ok_or_else(|| format!("role `{role}` has no permission on `{model_name}`"))?; match kind { RootKind::List => { return Err( @@ -133,6 +144,11 @@ fn compile_cell_by_key( ); } } + let row_filter = permission + .row_filter + .as_ref() + .map(|filter| resolve_cell_row_filter(&entry.schema, session, filter)) + .transpose()?; let mut pk = BTreeMap::new(); for column in &entry.schema.primary_key.columns { let value = selection @@ -153,9 +169,333 @@ fn compile_cell_by_key( Ok(QueryPlan::CellByKey { model: model_name.to_string(), pk, + row_filter, + }) +} + +/// Resolve a cell row policy before the remote GET so missing or malformed +/// claims cannot turn row existence into an authorization side channel. +fn resolve_cell_row_filter( + schema: &TableSchema, + session: &Session, + filter: &FilterExpr, +) -> Result { + Ok(match filter { + FilterExpr::And(items) => FilterExpr::And( + items + .iter() + .map(|item| resolve_cell_row_filter(schema, session, item)) + .collect::>()?, + ), + FilterExpr::Or(items) => FilterExpr::Or( + items + .iter() + .map(|item| resolve_cell_row_filter(schema, session, item)) + .collect::>()?, + ), + FilterExpr::Not(item) => { + FilterExpr::Not(Box::new(resolve_cell_row_filter(schema, session, item)?)) + } + FilterExpr::Cmp { column, op, rhs } => { + let column_schema = cell_policy_column(schema, column)?; + match op { + CmpOp::Eq | CmpOp::Neq + if matches!( + column_schema.column_type, + ColumnType::Text + | ColumnType::Timestamp + | ColumnType::Boolean + | ColumnType::Integer + | ColumnType::UnsignedInteger + | ColumnType::Float + ) => {} + CmpOp::Gt | CmpOp::Gte | CmpOp::Lt | CmpOp::Lte + if matches!( + column_schema.column_type, + ColumnType::Integer | ColumnType::UnsignedInteger | ColumnType::Float + ) => {} + _ => { + return Err(format!( + "cell-by-key row policy operator `{op:?}` is unsupported for column `{column}`" + )); + } + } + FilterExpr::Cmp { + column: column.clone(), + op: *op, + rhs: resolve_cell_operand(rhs, session, &column_schema.column_type)?, + } + } + FilterExpr::In { + column, + values, + negated, + } => { + let column_schema = cell_policy_column(schema, column)?; + if !matches!( + column_schema.column_type, + ColumnType::Text + | ColumnType::Timestamp + | ColumnType::Boolean + | ColumnType::Integer + | ColumnType::UnsignedInteger + | ColumnType::Float + ) { + return Err(format!( + "cell-by-key row policy IN is unsupported for column `{column}`" + )); + } + FilterExpr::In { + column: column.clone(), + values: values + .iter() + .map(|value| resolve_cell_operand(value, session, &column_schema.column_type)) + .collect::>()?, + negated: *negated, + } + } + FilterExpr::IsNull { column, is_null } => { + cell_policy_column(schema, column)?; + FilterExpr::IsNull { + column: column.clone(), + is_null: *is_null, + } + } + FilterExpr::Rel { field, .. } => { + return Err(format!( + "cell-by-key row policy cannot traverse relationship `{field}`" + )); + } }) } +fn cell_policy_column<'a>( + schema: &'a TableSchema, + column: &str, +) -> Result<&'a crate::table::TableColumn, String> { + schema + .columns + .iter() + .find(|candidate| candidate.column_name == column) + .ok_or_else(|| format!("unknown cell row-policy column `{column}`")) +} + +fn resolve_cell_operand( + operand: &Operand, + session: &Session, + column_type: &ColumnType, +) -> Result { + let literal = match operand_to_bind(operand, session, column_type)? { + BindValue::Null => LitValue::Null, + BindValue::Bool(value) => LitValue::Bool(value), + BindValue::I64(value) => LitValue::I64(value), + BindValue::F64(value) if value.is_finite() => LitValue::F64(value), + BindValue::F64(value) => { + return Err(format!( + "cell-by-key row-policy float `{value}` must be finite" + )); + } + BindValue::Text(value) => LitValue::String(value), + BindValue::Json(value) => LitValue::Json(value), + BindValue::Bytes(_) => { + return Err("cell-by-key row policies do not support byte operands".into()); + } + }; + Ok(Operand::Lit(literal)) +} + +/// Apply the already-resolved scalar policy to a sealed cell row. Only an +/// exact SQL-style TRUE authorizes the row; FALSE, NULL/unknown, malformed +/// fields, and unsupported material all fail closed. +pub(crate) fn cell_row_matches(schema: &TableSchema, filter: &FilterExpr, row: &JsonValue) -> bool { + let JsonValue::Object(row) = row else { + return false; + }; + matches!(evaluate_cell_filter(schema, filter, row), CellTruth::True) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum CellTruth { + True, + False, + Unknown, +} + +impl CellTruth { + fn not(self) -> Self { + match self { + Self::True => Self::False, + Self::False => Self::True, + Self::Unknown => Self::Unknown, + } + } +} + +fn evaluate_cell_filter( + schema: &TableSchema, + filter: &FilterExpr, + row: &serde_json::Map, +) -> CellTruth { + match filter { + FilterExpr::And(items) => { + let mut result = CellTruth::True; + for item in items { + match evaluate_cell_filter(schema, item, row) { + CellTruth::False => return CellTruth::False, + CellTruth::Unknown => result = CellTruth::Unknown, + CellTruth::True => {} + } + } + result + } + FilterExpr::Or(items) => { + let mut result = CellTruth::False; + for item in items { + match evaluate_cell_filter(schema, item, row) { + CellTruth::True => return CellTruth::True, + CellTruth::Unknown => result = CellTruth::Unknown, + CellTruth::False => {} + } + } + result + } + FilterExpr::Not(item) => evaluate_cell_filter(schema, item, row).not(), + FilterExpr::Cmp { column, op, rhs } => { + let Some((column_type, left)) = cell_row_value(schema, row, column) else { + return CellTruth::Unknown; + }; + let Operand::Lit(right) = rhs else { + return CellTruth::Unknown; + }; + evaluate_cell_comparison(&column_type, left, *op, right) + } + FilterExpr::In { + column, + values, + negated, + } => { + if values.is_empty() { + return if *negated { + CellTruth::True + } else { + CellTruth::False + }; + } + let Some((column_type, left)) = cell_row_value(schema, row, column) else { + return CellTruth::Unknown; + }; + let mut unknown = false; + for value in values { + let Operand::Lit(right) = value else { + unknown = true; + continue; + }; + match cell_values_equal(&column_type, left, right) { + Some(true) => { + return if *negated { + CellTruth::False + } else { + CellTruth::True + }; + } + Some(false) => {} + None => unknown = true, + } + } + if unknown { + CellTruth::Unknown + } else if *negated { + CellTruth::True + } else { + CellTruth::False + } + } + FilterExpr::IsNull { column, is_null } => { + let Some((_, value)) = cell_row_value(schema, row, column) else { + return CellTruth::Unknown; + }; + if value.is_null() == *is_null { + CellTruth::True + } else { + CellTruth::False + } + } + FilterExpr::Rel { .. } => CellTruth::Unknown, + } +} + +fn cell_row_value<'a>( + schema: &TableSchema, + row: &'a serde_json::Map, + column: &str, +) -> Option<(ColumnType, &'a JsonValue)> { + let column_schema = schema + .columns + .iter() + .find(|candidate| candidate.column_name == column)?; + let value = row + .get(&column_schema.field_name) + .or_else(|| row.get(&column_schema.column_name))?; + Some((column_schema.column_type.clone(), value)) +} + +fn evaluate_cell_comparison( + column_type: &ColumnType, + left: &JsonValue, + op: CmpOp, + right: &LitValue, +) -> CellTruth { + let matched = match op { + CmpOp::Eq => cell_values_equal(column_type, left, right), + CmpOp::Neq => cell_values_equal(column_type, left, right).map(|equal| !equal), + CmpOp::Gt => cell_values_order(column_type, left, right).map(|order| order.is_gt()), + CmpOp::Gte => cell_values_order(column_type, left, right).map(|order| order.is_ge()), + CmpOp::Lt => cell_values_order(column_type, left, right).map(|order| order.is_lt()), + CmpOp::Lte => cell_values_order(column_type, left, right).map(|order| order.is_le()), + CmpOp::Like | CmpOp::Ilike | CmpOp::Contains | CmpOp::ContainedIn | CmpOp::HasKey => None, + }; + match matched { + Some(true) => CellTruth::True, + Some(false) => CellTruth::False, + None => CellTruth::Unknown, + } +} + +fn cell_values_equal(column_type: &ColumnType, left: &JsonValue, right: &LitValue) -> Option { + if left.is_null() || matches!(right, LitValue::Null) { + return None; + } + Some(match (column_type, right) { + (ColumnType::Text | ColumnType::Timestamp, LitValue::String(right)) => { + left.as_str()? == right + } + (ColumnType::Boolean, LitValue::Bool(right)) => left.as_bool()? == *right, + (ColumnType::Integer, LitValue::I64(right)) => left.as_i64()? == *right, + (ColumnType::UnsignedInteger, LitValue::I64(right)) if *right >= 0 => { + left.as_u64()? == *right as u64 + } + (ColumnType::Float, LitValue::F64(right)) => left.as_f64()? == *right, + (ColumnType::Float, LitValue::I64(right)) => left.as_f64()? == *right as f64, + _ => return None, + }) +} + +fn cell_values_order( + column_type: &ColumnType, + left: &JsonValue, + right: &LitValue, +) -> Option { + match (column_type, right) { + (ColumnType::Integer, LitValue::I64(right)) => left.as_i64()?.partial_cmp(right), + (ColumnType::UnsignedInteger, LitValue::I64(right)) if *right >= 0 => { + left.as_u64()?.partial_cmp(&(*right as u64)) + } + (ColumnType::Float, LitValue::F64(right)) => left.as_f64()?.partial_cmp(right), + (ColumnType::Float, LitValue::I64(right)) => left.as_f64()?.partial_cmp(&(*right as f64)), + _ => None, + } +} + /// Compile a root field selection into one SQL statement. pub fn compile_root( inner: &EngineInner, diff --git a/src/graphql/mod.rs b/src/graphql/mod.rs index dad928fc6..0710e8568 100644 --- a/src/graphql/mod.rs +++ b/src/graphql/mod.rs @@ -70,7 +70,6 @@ pub use permissions::{ }; pub use types::{GraphqlInputType, GraphqlOutputType, GraphqlTypeDef, GraphqlTypeField}; -#[cfg(feature = "graphql")] pub(crate) mod command_input; #[cfg(feature = "graphql")] mod compile; diff --git a/src/graphql/read_store.rs b/src/graphql/read_store.rs index 83541b62f..bee7f45ce 100644 --- a/src/graphql/read_store.rs +++ b/src/graphql/read_store.rs @@ -195,7 +195,12 @@ mod tests { } fn blob_perms() -> ModelPermissions { - ModelPermissions::new().grant("user", read().all_columns()) + ModelPermissions::new().grant( + "user", + read() + .all_columns() + .rows(col("owner_id").eq(claim("x-user-id"))), + ) } fn todo_perms() -> ModelPermissions { @@ -404,6 +409,40 @@ mod tests { assert_eq!(data["blob_games_by_pk"]["score"], 9); } + #[tokio::test] + async fn graphql_by_id_hides_cell_rows_outside_the_role_policy() { + let cells = MapCellByKey::new(); + cells.insert( + "game-bob", + json!({ "game_id": "game-bob", "owner_id": "bob", "score": 9 }), + ); + cells.insert( + "game-malformed", + json!({ "game_id": "game-malformed", "score": 10 }), + ); + let engine = GraphqlEngine::builder(pool()) + .roles(&["user"]) + .model::(blob_perms()) + .model::(user_perms()) + .read_store::(ReadStore::CellByKey(Arc::new(cells))) + .build() + .unwrap(); + + for game_id in ["game-bob", "game-malformed"] { + let response = engine + .execute( + &session_user(), + Request::new(format!( + "{{ blob_games_by_pk(game_id: \"{game_id}\") {{ game_id score }} }}" + )), + ) + .await; + assert!(response.errors.is_empty(), "{response:?}"); + let data = response.data.into_json().unwrap(); + assert!(data["blob_games_by_pk"].is_null(), "{data}"); + } + } + #[tokio::test] async fn graphql_owner_join_fails_on_cell_store() { let cells = MapCellByKey::new(); diff --git a/src/graphql/schema.rs b/src/graphql/schema.rs index c94de546a..bf091bda0 100644 --- a/src/graphql/schema.rs +++ b/src/graphql/schema.rs @@ -681,6 +681,7 @@ async fn execute_cell_by_key( inner: &EngineInner, model: &str, pk: &BTreeMap, + row_filter: Option<&super::filter::FilterExpr>, selection: &compile::SelectionNode, ) -> Result { let getter = inner @@ -690,6 +691,16 @@ async fn execute_cell_by_key( let Some(row) = getter.get_sealed_row(pk).await? else { return Ok(Value::Null); }; + if let Some(filter) = row_filter { + let schema = &inner + .catalog + .get(model) + .ok_or_else(|| format!("unknown model `{model}`"))? + .schema; + if !compile::cell_row_matches(schema, filter, &row) { + return Ok(Value::Null); + } + } let mut out = serde_json::Map::new(); for child in &selection.children { if child.field_name == "__typename" { @@ -736,7 +747,11 @@ async fn resolve_root( let plan = compile::compile_query(&inner, &session, &role, model, kind, &selection) .map_err(|e| client_error("BAD_REQUEST", sanitize_compile_error(&e)))?; let value = match plan { - QueryPlan::CellByKey { model, pk } => execute_cell_by_key(&inner, &model, &pk, &selection) + QueryPlan::CellByKey { + model, + pk, + row_filter, + } => execute_cell_by_key(&inner, &model, &pk, row_filter.as_ref(), &selection) .await .map_err(|e| client_error("BAD_REQUEST", sanitize_compile_error(&e)))?, QueryPlan::Sql(plan) => { diff --git a/src/in_memory_repo/repository.rs b/src/in_memory_repo/repository.rs index 131afc4cd..defa9a17f 100644 --- a/src/in_memory_repo/repository.rs +++ b/src/in_memory_repo/repository.rs @@ -204,6 +204,36 @@ impl InMemoryRepository { Ok(()) } + pub(crate) fn clone_command_ledger(&self) -> Result, RepositoryError> { + Ok(self + .command_ledger + .read() + .map_err(|_| RepositoryError::LockPoisoned("command ledger read"))? + .values() + .cloned() + .collect()) + } + + pub(crate) fn replace_command_ledger( + &self, + records: Vec, + ) -> Result<(), RepositoryError> { + let mut ledger = HashMap::with_capacity(records.len()); + for record in records { + let key = record.key.clone(); + if ledger.insert(key, record).is_some() { + return Err(RepositoryError::Model( + "cell command ledger restore contains a duplicate key".into(), + )); + } + } + *self + .command_ledger + .write() + .map_err(|_| RepositoryError::LockPoisoned("command ledger write"))? = ledger; + Ok(()) + } + /// Whether a consumer inbox receipt for `(consumer, message_id)` is recorded. pub fn inbox_contains(&self, consumer: &str, message_id: &str) -> bool { self.inbox_store diff --git a/src/microsvc/cell_host/causal.rs b/src/microsvc/cell_host/causal.rs new file mode 100644 index 000000000..a27a3ecd4 --- /dev/null +++ b/src/microsvc/cell_host/causal.rs @@ -0,0 +1,367 @@ +//! Feature-free causal receipt and recovery for aggregate-cell wait paths. +//! +//! GraphQL owns its richer projection receipt, but the cell itself still owns +//! the durable command reservation and event/outbox commit. Keeping this layer +//! free of the `graphql` Cargo feature lets workers-rs cells use the exact same +//! fenced command ledger without pulling an HTTP server runtime into wasm. + +use std::time::Duration; + +use serde_json::Value; + +use crate::command_ledger::{ + AttemptFence, CausalCommitBatch, CausalTransactionalCommit, CommandAttempt, CommandId, + CommandLedgerError, CommandLedgerKey, CommandLedgerState, CommandLedgerStore, CommandLookup, + CommandLookupScope, CommandReplay, PrincipalPartitionId, TerminalCommandState, +}; +use crate::microsvc::HandlerError; +use crate::repository::CommitBatch; + +/// Internal wait-path header carrying the executable service identity. +/// +/// Public ingress must strip this header. The GraphQL celld command host sets +/// it from the locally bound [`Service`](crate::microsvc::Service). +pub const CELL_SERVICE_ID_HEADER: &str = "x-distributed-service-id"; + +/// Internal wait-path header carrying the verified-principal partition. +/// +/// This is an opaque server-derived value, never a public command argument. +pub const CELL_PRINCIPAL_PARTITION_HEADER: &str = "x-distributed-principal-partition"; + +/// Trusted command-ledger identity supplied by the cell's authenticated host. +/// +/// `principal_partition` is the opaque, server-derived partition produced by +/// the verified ingress. A public client must never be allowed to choose it. +#[derive(Clone, Debug)] +pub struct CellCommandIdentity { + key: CommandLedgerKey, +} + +impl CellCommandIdentity { + pub fn new( + service_id: impl Into, + principal_partition: impl Into, + command_id: impl AsRef, + ) -> Result { + let command_id = CommandId::parse(command_id).map_err(internal_ledger_error)?; + let principal_partition = + PrincipalPartitionId::new(principal_partition).map_err(internal_ledger_error)?; + let key = CommandLedgerKey::new(service_id, principal_partition, command_id) + .map_err(internal_ledger_error)?; + Ok(Self { key }) + } + + pub fn service_id(&self) -> &str { + self.key.service_id() + } + + pub fn command_id(&self) -> &str { + self.key.command_id() + } + + pub(crate) fn key(&self) -> &CommandLedgerKey { + &self.key + } +} + +/// Exact terminal cell receipt recovered from the command ledger. +#[derive(Clone, Debug, PartialEq)] +pub struct CellDispatchResult { + payload: Value, + command_id: String, + causation_id: String, + state: String, + replayed: bool, +} + +impl CellDispatchResult { + pub fn payload(&self) -> &Value { + &self.payload + } + + pub fn command_id(&self) -> &str { + &self.command_id + } + + pub fn causation_id(&self) -> &str { + &self.causation_id + } + + pub fn state(&self) -> &str { + &self.state + } + + pub fn replayed(&self) -> bool { + self.replayed + } +} + +/// Stable error vocabulary for the feature-free cell wait path. +#[derive(Debug)] +pub enum CellDispatchError { + BadRequest(String), + Unauthorized, + Forbidden, + CommandIdReuse, + InProgress, + Expired, + Rejected { + code: &'static str, + status: u16, + message: String, + }, + Internal(String), +} + +impl CellDispatchError { + pub fn code(&self) -> &'static str { + match self { + Self::BadRequest(_) => "BAD_REQUEST", + Self::Unauthorized => "UNAUTHORIZED", + Self::Forbidden => "FORBIDDEN", + Self::CommandIdReuse => "COMMAND_ID_REUSE", + Self::InProgress => "COMMAND_IN_PROGRESS", + Self::Expired => "COMMAND_EXPIRED", + Self::Rejected { code, .. } => code, + Self::Internal(_) => "INTERNAL", + } + } + + pub fn status_code(&self) -> u16 { + match self { + Self::BadRequest(_) => 400, + Self::Unauthorized => 401, + Self::Forbidden => 403, + Self::CommandIdReuse | Self::InProgress => 409, + Self::Expired => 410, + Self::Rejected { status, .. } => *status, + Self::Internal(_) => 500, + } + } + + pub fn client_message(&self) -> String { + match self { + Self::BadRequest(message) => message.clone(), + Self::Unauthorized => "missing authenticated principal".into(), + Self::Forbidden => "command is not allowed".into(), + Self::CommandIdReuse => "command ID was already used for different input".into(), + Self::InProgress => "command is already in progress".into(), + Self::Expired => "command ID has expired".into(), + Self::Rejected { message, .. } => message.clone(), + Self::Internal(_) => "internal error".into(), + } + } +} + +impl std::fmt::Display for CellDispatchError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Internal(detail) => formatter.write_str(detail), + _ => formatter.write_str(&self.client_message()), + } + } +} + +impl std::error::Error for CellDispatchError {} + +pub(crate) fn handler_error_code(error: &HandlerError) -> &'static str { + match error.status_code() { + 400 => "BAD_REQUEST", + 401 => "UNAUTHORIZED", + 403 => "FORBIDDEN", + 404 => "NOT_FOUND", + _ => "REJECTED", + } +} + +pub(crate) fn internal_ledger_error(error: CommandLedgerError) -> CellDispatchError { + match error { + CommandLedgerError::Invalid(message) => CellDispatchError::BadRequest(message), + other => CellDispatchError::Internal(other.to_string()), + } +} + +pub(crate) fn replay_result( + replay: CommandReplay, + replayed: bool, +) -> Result { + match replay.state { + CommandLedgerState::Succeeded + | CommandLedgerState::SucceededPendingProjection + | CommandLedgerState::Atomic + | CommandLedgerState::ProjectionFailed => Ok(CellDispatchResult { + payload: replay.outcome, + command_id: replay.command_id.as_str().to_string(), + causation_id: replay.causation_id.as_str().to_string(), + state: replay.state.as_str().to_string(), + replayed, + }), + CommandLedgerState::Rejected => replay_rejection(replay.outcome), + CommandLedgerState::InProgress + | CommandLedgerState::RetryableUnknown + | CommandLedgerState::Expired => Err(CellDispatchError::Internal( + "stored cell replay has a non-terminal state".into(), + )), + } +} + +fn replay_rejection(outcome: Value) -> Result { + let error = outcome + .get("error") + .and_then(Value::as_object) + .ok_or_else(|| CellDispatchError::Internal("stored cell rejection is malformed".into()))?; + let code = match error.get("code").and_then(Value::as_str) { + Some("BAD_REQUEST") => "BAD_REQUEST", + Some("UNAUTHORIZED") => "UNAUTHORIZED", + Some("FORBIDDEN") => "FORBIDDEN", + Some("NOT_FOUND") => "NOT_FOUND", + Some("REJECTED") => "REJECTED", + _ => { + return Err(CellDispatchError::Internal( + "stored cell rejection code is invalid".into(), + )); + } + }; + let status = error + .get("status") + .and_then(Value::as_u64) + .and_then(|status| u16::try_from(status).ok()) + .filter(|status| (400..500).contains(status)) + .ok_or_else(|| { + CellDispatchError::Internal("stored cell rejection status is invalid".into()) + })?; + let message = error + .get("message") + .and_then(Value::as_str) + .ok_or_else(|| { + CellDispatchError::Internal("stored cell rejection message is invalid".into()) + })? + .to_string(); + Err(CellDispatchError::Rejected { + code, + status, + message, + }) +} + +pub(crate) async fn commit_rejection( + repository: &R, + attempt: CommandAttempt, + retention: Duration, + code: &'static str, + status: u16, + message: String, +) -> Result +where + R: CommandLedgerStore + CausalTransactionalCommit + Send + Sync, +{ + let outcome = serde_json::json!({ + "error": { + "code": code, + "status": status, + "message": message, + } + }); + let fence = attempt.fence(); + let completion = attempt + .complete(TerminalCommandState::Rejected, outcome, retention) + .map_err(internal_ledger_error)?; + match repository + .commit_causal_batch(CausalCommitBatch::new(CommitBatch::empty(), completion)) + .await + { + Ok(()) => Err(CellDispatchError::Rejected { + code, + status, + message, + }), + Err(error) => recover_commit_error(repository, fence, error.to_string()).await, + } +} + +pub(crate) async fn load_committed_result( + repository: &R, + fence: &AttemptFence, + replayed: bool, +) -> Result +where + R: CommandLedgerStore + Send + Sync, +{ + match repository + .lookup_command(fence.key(), CommandLookupScope::Attempt(fence)) + .await + .map_err(internal_ledger_error)? + { + CommandLookup::Replay(replay) => replay_result(replay, replayed), + CommandLookup::Expired => Err(CellDispatchError::Expired), + CommandLookup::InProgress { .. } + | CommandLookup::RetryableUnknown { .. } + | CommandLookup::Unknown => Err(CellDispatchError::Internal( + "committed cell command has no exact durable replay receipt".into(), + )), + } +} + +pub(crate) async fn abandon_attempt( + repository: &R, + attempt: CommandAttempt, + detail: String, +) -> Result +where + R: CommandLedgerStore + Send + Sync, +{ + let fence = attempt.fence(); + match repository.mark_retryable_unknown(fence.clone()).await { + Ok(()) => Err(CellDispatchError::Internal(detail)), + Err(CommandLedgerError::AttemptFenced { .. }) => { + resolve_ambiguous_lookup(repository, fence, detail).await + } + Err(error) => Err(CellDispatchError::Internal(format!( + "{detail}; failed to mark cell command retryable: {error}" + ))), + } +} + +pub(crate) async fn recover_commit_error( + repository: &R, + fence: AttemptFence, + detail: String, +) -> Result +where + R: CommandLedgerStore + Send + Sync, +{ + resolve_ambiguous_lookup(repository, fence, detail).await +} + +async fn resolve_ambiguous_lookup( + repository: &R, + fence: AttemptFence, + detail: String, +) -> Result +where + R: CommandLedgerStore + Send + Sync, +{ + match repository + .lookup_command(fence.key(), CommandLookupScope::Attempt(&fence)) + .await + { + Ok(CommandLookup::Replay(replay)) => replay_result(replay, false), + Ok(CommandLookup::Expired) => Err(CellDispatchError::Expired), + Ok(CommandLookup::RetryableUnknown { .. }) => Err(CellDispatchError::Internal(detail)), + Ok(CommandLookup::InProgress { .. }) => { + match repository.mark_retryable_unknown(fence).await { + Ok(()) => Err(CellDispatchError::Internal(detail)), + Err(CommandLedgerError::AttemptFenced { .. }) => Err(CellDispatchError::InProgress), + Err(error) => Err(CellDispatchError::Internal(format!( + "{detail}; cell command recovery failed: {error}" + ))), + } + } + Ok(CommandLookup::Unknown) => Err(CellDispatchError::Internal(format!( + "{detail}; cell command ledger row disappeared" + ))), + Err(error) => Err(CellDispatchError::Internal(format!( + "{detail}; cell command outcome lookup failed: {error}" + ))), + } +} diff --git a/src/microsvc/cell_host/cell.rs b/src/microsvc/cell_host/cell.rs index a2bab1736..7d3e0b3ca 100644 --- a/src/microsvc/cell_host/cell.rs +++ b/src/microsvc/cell_host/cell.rs @@ -5,7 +5,8 @@ use std::collections::HashMap; use serde_json::Value; -use super::store::{CellStreamStore, DurableCellEvents, DurableCellSnapshot}; +use super::causal::{CellCommandIdentity, CellDispatchError, CellDispatchResult}; +use super::store::{CellStreamStore, DurableCellCommand, DurableCellEvents, DurableCellSnapshot}; use crate::aggregate::{Aggregate, AggregateRepository}; use crate::microsvc::error::HandlerError; use crate::microsvc::service::{PortableCommand, Routes}; @@ -105,6 +106,23 @@ where .await } + /// Dispatch a wait-path command through this cell's fenced command ledger. + /// + /// Same principal/command ID plus the same canonical typed input replays + /// the original payload without invoking the handler. Reusing the ID for a + /// different command or input fails before domain effects can commit. + pub async fn dispatch_idempotent( + &self, + command: &str, + identity: &CellCommandIdentity, + input: Value, + session: Session, + ) -> Result { + self.routes + .dispatch_cell_causal(command, identity, input, session, &self.shard) + .await + } + /// Load this cell's aggregate from the private stream store. /// /// HTTP GET on the cell host is a stream load, not a GraphQL/projector @@ -155,6 +173,19 @@ where .restore_durable_snapshots(snapshots) } + /// Command-ledger rows for Durable Object SQLite. + pub fn durable_commands(&self) -> Result, RepositoryError> { + self.routes.repo().repo().durable_commands() + } + + /// Restore command-ledger rows before accepting another wait-path request. + pub fn restore_durable_commands( + &self, + commands: Vec, + ) -> Result<(), RepositoryError> { + self.routes.repo().repo().restore_durable_commands(commands) + } + /// Read the repository snapshot cache for this cell's shard. pub async fn cached_snapshot(&self) -> Result, RepositoryError> { SnapshotStore::get_snapshot(self.routes.repo().repo(), &self.shard).await diff --git a/src/microsvc/cell_host/command.rs b/src/microsvc/cell_host/command.rs index dda2939a5..513c2e9cc 100644 --- a/src/microsvc/cell_host/command.rs +++ b/src/microsvc/cell_host/command.rs @@ -19,6 +19,8 @@ use crate::microsvc::{ CausalCommandPublicStatus, CausalDispatchError, CausalDispatchResult, Service, Session, }; +const COMPLETED_STATUS_CACHE_LIMIT: usize = 4_096; + /// One aggregate's cell wait-path: command names, URL kind, shard id, payload. #[derive(Clone, Copy)] pub struct CelldRoute { @@ -53,7 +55,7 @@ pub struct CelldCommandHost

{ local: LocalCommandHost, routes: Vec, pending: Arc>>, - completed: Arc>>, + completed: Arc>>, } impl

CelldCommandHost

@@ -86,6 +88,59 @@ where .iter() .find(|route| route.commands.contains(&command)) } + + fn service_id(&self) -> Result<&str, CausalDispatchError> { + self.local.service().name().ok_or_else(|| { + CausalDispatchError::Internal( + "celld command host requires a named executable service".into(), + ) + }) + } + + fn remember_completed(&self, key: (String, String), status: CausalCommandPublicStatus) { + let Ok(mut completed) = self.completed.lock() else { + return; + }; + if completed.len() >= COMPLETED_STATUS_CACHE_LIMIT && !completed.contains_key(&key) { + if let Some(evicted) = completed.keys().next().cloned() { + completed.remove(&evicted); + } + } + completed.insert(key, status); + } +} + +fn remote_dispatch_error(status: u16, body: &Value) -> CausalDispatchError { + let message = body + .get("error") + .and_then(Value::as_str) + .unwrap_or("wait-path rejected") + .to_string(); + match body.get("code").and_then(Value::as_str) { + Some("BAD_REQUEST") => CausalDispatchError::BadRequest(message), + Some("FORBIDDEN") => CausalDispatchError::Forbidden, + Some("COMMAND_ID_REUSE") => CausalDispatchError::CommandIdReuse, + Some("COMMAND_IN_PROGRESS") => CausalDispatchError::InProgress, + Some("COMMAND_EXPIRED") => CausalDispatchError::Expired, + Some("INTERNAL") => { + CausalDispatchError::Internal(format!("cell wait-path failed with HTTP {status}")) + } + Some("UNAUTHORIZED") => CausalDispatchError::Rejected { + code: "UNAUTHORIZED", + status, + message, + }, + Some("NOT_FOUND") => CausalDispatchError::Rejected { + code: "NOT_FOUND", + status, + message, + }, + _ => CausalDispatchError::Rejected { + code: "REJECTED", + status, + message, + }, + } } #[async_trait] @@ -108,17 +163,28 @@ where .invoke(command, command_id, input, session, principal, protocol) .await; }; - let shard = (route.shard)(&input).filter(|value| !value.is_empty()).ok_or_else(|| { - CausalDispatchError::BadRequest(format!( - "{} id required for celld wait-path", - route.kind - )) - })?; + let shard = (route.shard)(&input) + .filter(|value| !value.is_empty()) + .ok_or_else(|| { + CausalDispatchError::BadRequest(format!( + "{} id required for celld wait-path", + route.kind + )) + })?; + let service_id = self.service_id()?.to_string(); + let principal_partition = principal.partition_for_service(&service_id); let http = self .http .retarget(format!("{}/{}/{}", self.celld_url, route.kind, shard)); let (status, body) = http - .post_wait_path(command, command_id, input.clone(), &session) + .post_cell_wait_path( + command, + command_id, + input.clone(), + &session, + &service_id, + &principal_partition, + ) .await?; let outbox = CausalDispatchResult::outbox_from_wait_path(&body); if !outbox.is_empty() { @@ -128,20 +194,10 @@ where } drain_cell_outbox(&http, &self.publisher, &outbox).await; if status >= 400 { - let message = body - .get("error") - .and_then(Value::as_str) - .unwrap_or("wait-path rejected") - .to_string(); - return Err(CausalDispatchError::Rejected { - code: "REJECTED", - status, - message, - }); + return Err(remote_dispatch_error(status, &body)); } - let remote = CausalDispatchResult::from_wait_path_wire(body).map_err(|error| { - CausalDispatchError::Internal(format!("wait-path decode: {error}")) - })?; + let remote = CausalDispatchResult::from_wait_path_wire(body) + .map_err(|error| CausalDispatchError::Internal(format!("wait-path decode: {error}")))?; let payload = (route.payload)(command, &input, remote.payload(), &session); let mut remote = remote.with_payload(payload); if let Some(protocol) = protocol { @@ -149,10 +205,11 @@ where .local .service() .seal_wait_path_dispatch(command, &protocol, remote)?; - if let Ok(mut guard) = self.completed.lock() { - guard.insert(command_id.to_string(), remote.public_status()); - } } + self.remember_completed( + (principal_partition, command_id.to_string()), + remote.public_status(), + ); Ok(remote) } @@ -163,11 +220,14 @@ where principal: VerifiedPrincipal, protocol: Option, ) -> Result { + let service_id = self.service_id()?; + let principal_partition = principal.partition_for_service(service_id); + let key = (principal_partition, command_id.to_string()); if let Some(status) = self .completed .lock() .ok() - .and_then(|guard| guard.get(command_id).cloned()) + .and_then(|guard| guard.get(&key).cloned()) { return Ok(status); } diff --git a/src/microsvc/cell_host/mod.rs b/src/microsvc/cell_host/mod.rs index 0ea84ddb8..92b07222d 100644 --- a/src/microsvc/cell_host/mod.rs +++ b/src/microsvc/cell_host/mod.rs @@ -12,15 +12,19 @@ //! GraphQL wait-path + cell SQLite outbox drain (`CelldCommandHost`) is //! the same for every aggregate: routes only supply kind, shard, and payload. +pub(crate) mod causal; mod cell; -mod store; #[cfg(feature = "graphql")] mod command; #[cfg(feature = "graphql")] mod outbox; +mod store; +pub use causal::{ + CellCommandIdentity, CellDispatchError, CellDispatchResult, CELL_PRINCIPAL_PARTITION_HEADER, + CELL_SERVICE_ID_HEADER, +}; pub use cell::{instance_name, parent_cell_name, AggregateCell, CellNamespace}; -pub use store::{CellStreamStore, DurableCellEvents, DurableCellSnapshot}; #[cfg(feature = "graphql")] pub use command::{CelldCommandHost, CelldRoute}; #[cfg(feature = "graphql")] @@ -28,6 +32,7 @@ pub use outbox::{ accept_outbox_drain, drain_cell_outbox, outbox_alarm_handler, spawn_cell_outbox_drain_loop, CellOutboxDrainHandler, CELL_OUTBOX_DRAIN_PATH, }; +pub use store::{CellStreamStore, DurableCellCommand, DurableCellEvents, DurableCellSnapshot}; #[cfg(test)] mod tests; diff --git a/src/microsvc/cell_host/store.rs b/src/microsvc/cell_host/store.rs index 8f2469433..7ccb1a21e 100644 --- a/src/microsvc/cell_host/store.rs +++ b/src/microsvc/cell_host/store.rs @@ -16,6 +16,7 @@ use crate::command_ledger::{ }; use crate::entity::{Entity, EventRecord}; use crate::microsvc::HasOutboxStore; +use crate::outbox::OutboxMessage; use crate::projection_protocol::{ ProjectionChangeCursor, ProjectionChangeRead, ProjectionCheckpoint, ProjectionCommitBatch, ProjectionCommitResult, ProjectionFailure, ProjectionFailureBatch, ProjectionFailureLocation, @@ -33,7 +34,6 @@ use crate::repository::{ TransactionalCommit, }; use crate::snapshot::SnapshotRecord; -use crate::outbox::OutboxMessage; use crate::{InMemoryOutboxStore, InMemoryRepository}; use serde::{Deserialize, Serialize}; @@ -82,6 +82,13 @@ pub struct DurableCellSnapshot { pub payload: Vec, } +/// One versioned command-ledger row for Durable Object SQLite persistence. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct DurableCellCommand { + pub id: String, + pub body: String, +} + #[derive(Clone)] pub struct CellStreamStore { ownership: CellOwnership, @@ -255,6 +262,41 @@ impl CellStreamStore { ) } + /// Fenced command rows committed with this cell's domain effects. + pub fn durable_commands(&self) -> Result, RepositoryError> { + self.inner + .clone_command_ledger()? + .into_iter() + .map(|record| { + let id = record.durable_cell_key(); + let body = record + .durable_cell_json() + .map_err(|error| RepositoryError::Model(error.to_string()))?; + Ok(DurableCellCommand { id, body }) + }) + .collect() + } + + /// Restore the complete command ledger before accepting another request. + pub fn restore_durable_commands( + &self, + commands: Vec, + ) -> Result<(), RepositoryError> { + let mut records = Vec::with_capacity(commands.len()); + for command in commands { + let record = + crate::command_ledger::CommandLedgerRecord::from_durable_cell_json(&command.body) + .map_err(|error| RepositoryError::Model(error.to_string()))?; + if record.durable_cell_key() != command.id { + return Err(RepositoryError::Model( + "cell command ledger row key does not match its body".into(), + )); + } + records.push(record); + } + self.inner.replace_command_ledger(records) + } + fn ensure_batch(&self, batch: &CommitBatch<'_>) -> Result<(), RepositoryError> { for stream in &batch.streams { self.ensure_identity(&stream.identity)?; diff --git a/src/microsvc/cell_host/tests.rs b/src/microsvc/cell_host/tests.rs index 3930b4bd4..56bc8b879 100644 --- a/src/microsvc/cell_host/tests.rs +++ b/src/microsvc/cell_host/tests.rs @@ -1,4 +1,7 @@ -use super::{instance_name, parent_cell_name, AggregateCell, CellNamespace, CellStreamStore}; +use super::{ + instance_name, parent_cell_name, AggregateCell, CellCommandIdentity, CellDispatchError, + CellNamespace, CellStreamStore, +}; use crate::aggregate::{Aggregate, AggregateRepository}; use crate::entity::Entity; use crate::graphql::{typed_command, PreparedCommand, Succeeded}; @@ -269,6 +272,121 @@ async fn cell_dispatches_complete_with_the_same_portable_command_as_soa() { assert_eq!(loaded.entity.snapshot_version(), 2); } +#[tokio::test] +async fn cell_wait_path_replays_the_same_command_without_new_domain_effects() { + let cell = AggregateCell::::new("item-ledger") + .unwrap() + .mount(Create) + .mount(Complete); + let identity = CellCommandIdentity::new( + "cell-test-service", + "principal-alice", + "0190a000-0000-7000-8000-000000000401", + ) + .unwrap(); + let input = json!({ "id": "item-ledger", "title": "once" }); + + let first = cell + .dispatch_idempotent( + "cell_item.create", + &identity, + input.clone(), + owner_session(), + ) + .await + .expect("first dispatch"); + let replay = cell + .dispatch_idempotent("cell_item.create", &identity, input, owner_session()) + .await + .expect("same-input replay"); + + assert!(!first.replayed()); + assert!(replay.replayed()); + assert_eq!(replay.payload(), first.payload()); + assert_eq!(replay.causation_id(), first.causation_id()); + let events = cell.durable_events().unwrap(); + assert_eq!( + events + .iter() + .map(|stream| stream.events.len()) + .sum::(), + 1, + "replay must not invoke the handler or append another event" + ); + assert_eq!( + events[0].events[0].causation_id(), + Some(first.causation_id()) + ); + + let durable_commands = cell.durable_commands().expect("export command ledger"); + assert_eq!(durable_commands.len(), 1); + let restored = AggregateCell::::new("item-ledger") + .unwrap() + .mount(Create) + .mount(Complete); + restored + .restore_durable_events(events) + .expect("restore domain events"); + restored + .restore_durable_commands(durable_commands) + .expect("restore command ledger"); + let replay_after_restart = restored + .dispatch_idempotent( + "cell_item.create", + &identity, + json!({ "id": "item-ledger", "title": "once" }), + owner_session(), + ) + .await + .expect("durable replay after restart"); + assert!(replay_after_restart.replayed()); + assert_eq!(replay_after_restart.causation_id(), first.causation_id()); + assert_eq!( + restored + .durable_events() + .unwrap() + .iter() + .map(|stream| stream.events.len()) + .sum::(), + 1 + ); +} + +#[tokio::test] +async fn cell_wait_path_rejects_command_id_reuse_with_different_input() { + let cell = AggregateCell::::new("item-conflict") + .unwrap() + .mount(Create) + .mount(Complete); + let identity = CellCommandIdentity::new( + "cell-test-service", + "principal-alice", + "0190a000-0000-7000-8000-000000000402", + ) + .unwrap(); + cell.dispatch_idempotent( + "cell_item.create", + &identity, + json!({ "id": "item-conflict", "title": "first" }), + owner_session(), + ) + .await + .unwrap(); + + let error = cell + .dispatch_idempotent( + "cell_item.create", + &identity, + json!({ "id": "item-conflict", "title": "different" }), + owner_session(), + ) + .await + .unwrap_err(); + assert!(matches!(error, CellDispatchError::CommandIdReuse)); + assert_eq!(error.code(), "COMMAND_ID_REUSE"); + assert_eq!(error.status_code(), 409); +} + #[tokio::test] async fn cell_complete_rejects_a_different_shard_id() { let cell = AggregateCell::::new("item-1") diff --git a/src/microsvc/mod.rs b/src/microsvc/mod.rs index 84a6221a8..55e8798d1 100644 --- a/src/microsvc/mod.rs +++ b/src/microsvc/mod.rs @@ -113,7 +113,9 @@ pub use service::GraphqlServiceBindError; feature = "rabbitmq", feature = "kafka", ))] -pub use workers::{spawn_outbox_publish_loop, spawn_service_consumer_loop}; +pub use workers::{ + spawn_outbox_publish_loop, spawn_service_consumer_loop, CONSUMER_IDLE_POLL, +}; pub use service::{ direct_read_model, invoke_transition, require_loaded, CausalCommandContext, CausalCommitBuilder, CausalRepository, CommandRequest, CommandResponse, DeliveryKind, DirectReadModelProjection, diff --git a/src/microsvc/service/routes.rs b/src/microsvc/service/routes.rs index 82f5c9dba..af488c495 100644 --- a/src/microsvc/service/routes.rs +++ b/src/microsvc/service/routes.rs @@ -23,24 +23,25 @@ use super::handlers::{ use crate::aggregate::Aggregate; use crate::application::{CommandMount, CommandMountRegistrar, CommandSpec}; use crate::bus::{Bus, Message, MessageKind, MessagePublisher, OrderedDelivery, TransportError}; -#[cfg(feature = "graphql")] use crate::command_ledger::{ - CanonicalInputHash, CausalCommitBatch, CausalRepositoryIdentity, CausalTransactionalCommit, - CommandContractFingerprint, CommandId, CommandLedgerKey, CommandLedgerStore, CommandLookup, - CommandLookupScope, CommandReservation, PrincipalPartitionId, ReservationOutcome, - TerminalCommandState, + CanonicalInputHash, CausalCommitBatch, CausalTransactionalCommit, CommandContractFingerprint, + CommandLedgerStore, CommandReservation, ReservationOutcome, TerminalCommandState, }; #[cfg(feature = "graphql")] +use crate::command_ledger::{ + CausalRepositoryIdentity, CommandId, CommandLedgerKey, CommandLookup, CommandLookupScope, + PrincipalPartitionId, +}; use crate::graphql::command_contract::CommandConsistency; use crate::graphql::command_contract::{ CommandEventSet, CommandOutcome, CompiledInputDefaults, TypedCommandContract, }; -#[cfg(feature = "graphql")] use crate::graphql::command_input::canonicalize_command_input; #[cfg(feature = "graphql")] use crate::graphql::identity::VerifiedPrincipal; use crate::graphql::{command_transition, GraphqlInputType, SurfaceProjector, TypedCommand}; use crate::microsvc::causal::CausalWorkspace; +use crate::microsvc::cell_host::{CellCommandIdentity, CellDispatchError, CellDispatchResult}; use crate::microsvc::context::Context; use crate::microsvc::dependencies::{ CausalProjectionRouteDependencies, CausalRouteDependencies, ConfigurableOutboxPublisher, @@ -188,6 +189,8 @@ pub(super) type CausalHandlerFuture<'a> = pub(super) type CausalStatusFuture<'a> = Pin< Box> + Send + 'a>, >; +pub(super) type CellCausalHandlerFuture<'a> = + Pin> + Send + 'a>>; pub(super) trait ErasedCausalHandler: Send + Sync { fn contract(&self) -> &TypedCommandContract; @@ -212,6 +215,15 @@ pub(super) trait ErasedCausalHandler: Send + Sync { protocol: Option, ) -> CausalHandlerFuture<'a>; + fn dispatch_cell_causal<'a>( + &'a self, + dependencies: &'a D, + identity: &'a CellCommandIdentity, + input: Value, + session: Session, + shard: &'a StreamIdentity, + ) -> CellCausalHandlerFuture<'a>; + #[cfg(feature = "graphql")] #[allow(dead_code)] fn lookup<'a>( @@ -1023,6 +1035,32 @@ impl Routes { } } + /// Dispatch a typed cell command through the same fenced ledger contract + /// as the in-process causal wait path. + pub(in crate::microsvc) async fn dispatch_cell_causal( + &self, + command: &str, + identity: &CellCommandIdentity, + input: Value, + session: Session, + shard: &StreamIdentity, + ) -> Result { + let handler = self + .handlers + .get(&MessageKind::Command) + .and_then(|handlers| handlers.get(command)); + match handler { + Some(RegisteredHandler::Causal(handler)) => { + handler + .dispatch_cell_causal(&self.dependencies, identity, input, session, shard) + .await + } + Some(_) | None => Err(CellDispatchError::BadRequest(format!( + "`{command}` is not a typed causal command" + ))), + } + } + pub(in crate::microsvc) fn is_command_only(&self) -> bool { self.projectors.is_empty() && self.modeled_local_services.is_empty() @@ -1498,6 +1536,224 @@ where &self.contract } + fn dispatch_cell_causal<'a>( + &'a self, + dependencies: &'a D, + identity: &'a CellCommandIdentity, + input: Value, + session: Session, + shard: &'a StreamIdentity, + ) -> CellCausalHandlerFuture<'a> { + Box::pin(async move { + match crate::application::admit_command_session( + &self.contract.roles, + session.user_id(), + &session.roles(), + ) { + Ok(()) => {} + Err("unauthenticated") => return Err(CellDispatchError::Unauthorized), + Err(_) => return Err(CellDispatchError::Forbidden), + } + if self.contract.consistency == CommandConsistency::Atomic { + return Err(CellDispatchError::BadRequest( + "atomic typed commands require a same-transaction relational projection host" + .into(), + )); + } + + let canonical = canonicalize_command_input(&self.contract.input, input) + .map_err(|error| CellDispatchError::BadRequest(error.to_string()))?; + let typed = canonical + .decode::() + .map_err(|error| CellDispatchError::BadRequest(error.to_string()))?; + let (input, wire, input_digest) = typed.into_parts(); + let policy = CausalCommandPolicy::default(); + let reservation = CommandReservation::new( + identity.key().clone(), + self.contract.name.clone(), + CommandContractFingerprint::new(self.contract.fingerprint_bytes()), + CanonicalInputHash::new(input_digest), + policy.attempt_lease, + policy.replay_retention, + ) + .map_err(crate::microsvc::cell_host::causal::internal_ledger_error)?; + + let aggregate_repository = dependencies.__causal_aggregate_repository(); + let repository = aggregate_repository.repo(); + let attempt = match repository + .reserve_command(reservation) + .await + .map_err(crate::microsvc::cell_host::causal::internal_ledger_error)? + { + ReservationOutcome::Acquired(attempt) => attempt, + ReservationOutcome::InProgress { .. } => { + return Err(CellDispatchError::InProgress); + } + ReservationOutcome::Replay(replay) => { + return crate::microsvc::cell_host::causal::replay_result(replay, true); + } + ReservationOutcome::Conflict => return Err(CellDispatchError::CommandIdReuse), + ReservationOutcome::Expired => return Err(CellDispatchError::Expired), + }; + + let payload = serde_json::to_vec(&wire).map_err(|error| { + CellDispatchError::Internal(format!( + "canonical cell command input could not be encoded: {error}" + )) + })?; + let mut metadata = session + .variables() + .iter() + .filter(|(name, _)| !name.eq_ignore_ascii_case(crate::trace_context::CAUSATION_ID)) + .map(|(name, value)| (name.clone(), value.clone())) + .collect::>(); + metadata.push(( + crate::trace_context::CAUSATION_ID.to_string(), + attempt.causation_id().as_str().to_string(), + )); + let message = Message { + id: Some(identity.command_id().to_string()), + name: self.contract.name.clone(), + kind: MessageKind::Command, + payload, + content_type: "application/json".into(), + metadata, + }; + + let workspace = CausalWorkspace::new(aggregate_repository); + let context = CausalCommandContext::new(&message, &session, &workspace); + if self.guard.as_ref().is_some_and(|guard| !guard(&context)) { + return crate::microsvc::cell_host::causal::commit_rejection( + repository, + attempt, + policy.replay_retention, + "REJECTED", + 422, + format!("guard rejected command: {}", self.contract.name), + ) + .await; + } + + let mut prepared = match (self.handle)(&context, input).await { + Ok(prepared) => prepared, + Err(error) if error.status_code() < 500 => { + return crate::microsvc::cell_host::causal::commit_rejection( + repository, + attempt, + policy.replay_retention, + crate::microsvc::cell_host::causal::handler_error_code(&error), + error.status_code(), + error.client_facing_message(), + ) + .await; + } + Err(error) => { + return crate::microsvc::cell_host::causal::abandon_attempt( + repository, + attempt, + error.to_string(), + ) + .await; + } + }; + + let mut parts = match workspace.into_parts() { + Ok(parts) => parts, + Err(error) => { + return crate::microsvc::cell_host::causal::abandon_attempt( + repository, + attempt, + error.to_string(), + ) + .await; + } + }; + if let Err(error) = parts.prepare_domain_publications(attempt.causation_id().as_str()) { + return crate::microsvc::cell_host::causal::abandon_attempt( + repository, + attempt, + error.to_string(), + ) + .await; + } + if let Err(error) = parts.validate_prepared(&self.contract, &mut prepared) { + return crate::microsvc::cell_host::causal::abandon_attempt( + repository, + attempt, + error.to_string(), + ) + .await; + } + + let replay_payload = prepared.serialized_payload().clone(); + let batch = match parts.prepare_commit_batch() { + Ok(batch) => batch, + Err(error) => { + return crate::microsvc::cell_host::causal::abandon_attempt( + repository, + attempt, + format!("cell causal commit batch preparation failed: {error}"), + ) + .await; + } + }; + let foreign_stream = batch + .streams + .iter() + .find(|stream| stream.identity != *shard) + .map(|stream| stream.identity.to_string()); + if let Some(foreign_stream) = foreign_stream { + drop(batch); + return crate::microsvc::cell_host::causal::abandon_attempt( + repository, + attempt, + format!("cell `{shard}` cannot commit stream `{foreign_stream}`"), + ) + .await; + } + + let fence = attempt.fence(); + let completion = attempt + .complete( + TerminalCommandState::Succeeded, + replay_payload.clone(), + policy.replay_retention, + ) + .map_err(crate::microsvc::cell_host::causal::internal_ledger_error)?; + match repository + .commit_causal_batch(CausalCommitBatch::new(batch, completion)) + .await + { + Ok(()) => { + parts.mark_committed_state().map_err(|error| { + CellDispatchError::Internal(format!( + "committed cell workspace cleanup failed: {error}" + )) + })?; + let (_committed, serialized) = prepared.finalize_after_commit(); + let result = crate::microsvc::cell_host::causal::load_committed_result( + repository, &fence, false, + ) + .await?; + if result.payload() != &serialized { + return Err(CellDispatchError::Internal( + "durable cell replay differs from the committed handler payload".into(), + )); + } + Ok(result) + } + Err(error) => { + crate::microsvc::cell_host::causal::recover_commit_error( + repository, + fence, + error.to_string(), + ) + .await + } + } + }) + } + #[cfg(feature = "graphql")] fn contract_mut(&mut self) -> &mut TypedCommandContract { &mut self.contract diff --git a/src/microsvc/workers.rs b/src/microsvc/workers.rs index 3a4b5a987..1b41e68dc 100644 --- a/src/microsvc/workers.rs +++ b/src/microsvc/workers.rs @@ -40,7 +40,20 @@ pub fn spawn_outbox_publish_loop( .spawn(); } -/// Spawn a service consumer loop that re-runs the bus handler continuously. +/// Idle poll for long-running SQL `listen`/`subscribe` hosts. +/// +/// Drain-to-idle is for tests. A host that lets `Service::run` return `Ok(())` +/// would otherwise reconstruct routes and bootstrap projectors on every quiet +/// stretch — seconds of delay on the next Eventual command. +pub const CONSUMER_IDLE_POLL: Duration = Duration::from_millis(25); + +/// Spawn the bus consumer for a long-running host. +/// +/// `build_service` constructs the heavy route/projector graph **once**, then +/// again only after `run` fails. A successful return means the bus drained to +/// idle; that is a host bug for SQL buses (use `with_idle_poll` / +/// [`CONSUMER_IDLE_POLL`]). We log and stop instead of reconstructing, so an +/// idle drain cannot hide behind a rebuild storm. pub fn spawn_service_consumer_loop(build_service: F) where F: Fn() -> Service + Send + Sync + 'static, @@ -49,7 +62,13 @@ where loop { let service = build_service(); match service.run(RunOptions::idempotent()).await { - Ok(()) => tokio::time::sleep(Duration::from_millis(25)).await, + Ok(()) => { + eprintln!( + "consumer: bus drained to idle; not reconstructing Service. \ + Long-running SQL hosts must call with_idle_poll({CONSUMER_IDLE_POLL:?})" + ); + return; + } Err(e) => { eprintln!("consumer: {e}"); tokio::time::sleep(Duration::from_millis(100)).await; diff --git a/tests/celld/main.rs b/tests/celld/main.rs index 5eef57d34..312ce9614 100644 --- a/tests/celld/main.rs +++ b/tests/celld/main.rs @@ -7,8 +7,12 @@ use std::path::Path; use std::time::Duration; +use distributed::cell_host::{CELL_PRINCIPAL_PARTITION_HEADER, CELL_SERVICE_ID_HEADER}; use serde_json::Value; +const TEST_SERVICE_ID: &str = "celld-live-test"; +const TEST_PRINCIPAL_PARTITION: &str = "test-principal-alice"; + #[path = "../support/env.rs"] mod env_support; @@ -58,6 +62,9 @@ fn worker_declares_sqlite_todo_and_chat_cells() { assert!(source.contains("outbox.complete")); assert!(source.contains("outbox.drain")); assert!(source.contains("CREATE TABLE IF NOT EXISTS cell_outbox")); + assert!(source.contains("CREATE TABLE IF NOT EXISTS cell_commands")); + assert!(source.contains("dispatch_idempotent")); + assert!(source.contains("restore_durable_commands")); assert!(source.contains("sealed_row")); assert!(source.contains("new_with_snapshots")); assert!(source.contains("restore_durable_events")); @@ -100,7 +107,7 @@ fn compose_file_does_not_use_minio() { } #[tokio::test] -async fn live_todo_cell_create_complete_and_isolate() { +async fn live_todo_cell_create_complete_reopen_archive_and_isolate() { let Some(base) = env_support::broker_env("CELLD_URL", "celld live Todo cell") else { return; }; @@ -115,17 +122,19 @@ async fn live_todo_cell_create_complete_and_isolate() { let a = unique_todo(); let b = unique_todo(); - let created = client - .post(format!("{base}/todo/{a}/todo.create")) - .header("x-user-id", "alice") - .header("x-roles", "user") - .json(&serde_json::json!({ - "commandId": "0190a000-0000-7000-8000-000000000201", - "input": { "title": "ship celld" } - })) - .send() - .await - .expect("create"); + let created = trusted_cell_request( + client + .post(format!("{base}/todo/{a}/todo.create")) + .header("x-user-id", "alice") + .header("x-roles", "user"), + ) + .json(&serde_json::json!({ + "commandId": "0190a000-0000-7000-8000-000000000201", + "input": { "title": "ship celld" } + })) + .send() + .await + .expect("create"); assert_eq!(created.status(), 201, "{}", created.text().await.unwrap()); let created: Value = created.json().await.unwrap(); assert_eq!(created["payload"]["id"], a); @@ -134,18 +143,60 @@ async fn live_todo_cell_create_complete_and_isolate() { created["receipt"]["commandId"], "0190a000-0000-7000-8000-000000000201" ); + let causation_id = created["receipt"]["causationId"] + .as_str() + .expect("causationId") + .to_string(); - let completed = client - .post(format!("{base}/todo/{a}/todo.complete")) - .header("x-user-id", "alice") - .header("x-roles", "user") - .json(&serde_json::json!({ - "commandId": "0190a000-0000-7000-8000-000000000202", - "input": {} - })) - .send() - .await - .expect("complete"); + let replay = trusted_cell_request( + client + .post(format!("{base}/todo/{a}/todo.create")) + .header("x-user-id", "alice") + .header("x-roles", "user"), + ) + .json(&serde_json::json!({ + "commandId": "0190a000-0000-7000-8000-000000000201", + "input": { "title": "ship celld" } + })) + .send() + .await + .expect("replay create"); + assert_eq!(replay.status(), 201, "{}", replay.text().await.unwrap()); + let replay: Value = replay.json().await.unwrap(); + assert_eq!(replay["receipt"]["replayed"], true); + assert_eq!(replay["receipt"]["causationId"], causation_id); + assert_eq!(replay["payload"], created["payload"]); + + let conflict = trusted_cell_request( + client + .post(format!("{base}/todo/{a}/todo.create")) + .header("x-user-id", "alice") + .header("x-roles", "user"), + ) + .json(&serde_json::json!({ + "commandId": "0190a000-0000-7000-8000-000000000201", + "input": { "title": "different input" } + })) + .send() + .await + .expect("conflicting create"); + assert_eq!(conflict.status(), 409, "{}", conflict.text().await.unwrap()); + let conflict: Value = conflict.json().await.unwrap(); + assert_eq!(conflict["code"], "COMMAND_ID_REUSE"); + + let completed = trusted_cell_request( + client + .post(format!("{base}/todo/{a}/todo.complete")) + .header("x-user-id", "alice") + .header("x-roles", "user"), + ) + .json(&serde_json::json!({ + "commandId": "0190a000-0000-7000-8000-000000000202", + "input": {} + })) + .send() + .await + .expect("complete"); assert_eq!( completed.status(), 200, @@ -159,6 +210,40 @@ async fn live_todo_cell_create_complete_and_isolate() { "0190a000-0000-7000-8000-000000000202" ); + let reopened = trusted_cell_request( + client + .post(format!("{base}/todo/{a}/todo.reopen")) + .header("x-user-id", "alice") + .header("x-roles", "user"), + ) + .json(&serde_json::json!({ + "commandId": "0190a000-0000-7000-8000-000000000203", + "input": {} + })) + .send() + .await + .expect("reopen"); + assert_eq!(reopened.status(), 200, "{}", reopened.text().await.unwrap()); + let reopened: Value = reopened.json().await.unwrap(); + assert_eq!(reopened["payload"]["status"], "open"); + + let archived = trusted_cell_request( + client + .post(format!("{base}/todo/{a}/todo.archive")) + .header("x-user-id", "alice") + .header("x-roles", "user"), + ) + .json(&serde_json::json!({ + "commandId": "0190a000-0000-7000-8000-000000000204", + "input": {} + })) + .send() + .await + .expect("archive"); + assert_eq!(archived.status(), 200, "{}", archived.text().await.unwrap()); + let archived: Value = archived.json().await.unwrap(); + assert_eq!(archived["payload"]["status"], "archived"); + let got: Value = client .get(format!("{base}/todo/{a}")) .send() @@ -168,7 +253,7 @@ async fn live_todo_cell_create_complete_and_isolate() { .await .unwrap(); assert_eq!(got["title"], "ship celld"); - assert_eq!(got["status"], "completed"); + assert_eq!(got["status"], "archived"); let other = client .get(format!("{base}/todo/{b}")) @@ -195,21 +280,24 @@ async fn live_chat_cell_post_and_isolate() { let b = unique_chat(); let created_at = unix_millis(); - let posted = client - .post(format!("{base}/chat/{a}/chat.post")) - .header("x-user-id", "alice") - .json(&serde_json::json!({ - "commandId": "0190a000-0000-7000-8000-000000000301", - "input": { - "message_id": a, - "room_id": "lobby", - "body": "hello from a cell", - "created_at": created_at, - } - })) - .send() - .await - .expect("post"); + let posted = trusted_cell_request( + client + .post(format!("{base}/chat/{a}/chat.post")) + .header("x-user-id", "alice") + .header("x-roles", "user"), + ) + .json(&serde_json::json!({ + "commandId": "0190a000-0000-7000-8000-000000000301", + "input": { + "message_id": a, + "room_id": "lobby", + "body": "hello from a cell", + "created_at": created_at, + } + })) + .send() + .await + .expect("post"); assert_eq!(posted.status(), 201, "{}", posted.text().await.unwrap()); let posted: Value = posted.json().await.unwrap(); assert_eq!(posted["payload"]["message_id"], a); @@ -234,19 +322,17 @@ async fn live_chat_cell_post_and_isolate() { let pending = posted["outbox"].as_array().cloned().unwrap_or_default(); if !pending.is_empty() { - let ids: Vec = pending.iter().filter_map(|row| row.get("id").cloned()).collect(); + let ids: Vec = pending + .iter() + .filter_map(|row| row.get("id").cloned()) + .collect(); let complete = client .post(format!("{base}/chat/{a}/outbox.complete")) .json(&serde_json::json!({ "ids": ids })) .send() .await .expect("outbox.complete"); - assert_eq!( - complete.status(), - 200, - "{}", - complete.text().await.unwrap() - ); + assert_eq!(complete.status(), 200, "{}", complete.text().await.unwrap()); let drained: Value = client .post(format!("{base}/chat/{a}/outbox.drain")) .json(&serde_json::json!({ @@ -294,6 +380,12 @@ fn unix_millis() -> String { .to_string() } +fn trusted_cell_request(request: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + request + .header(CELL_SERVICE_ID_HEADER, TEST_SERVICE_ID) + .header(CELL_PRINCIPAL_PARTITION_HEADER, TEST_PRINCIPAL_PARTITION) +} + async fn wait_healthy(client: &reqwest::Client, base: &str) { let deadline = std::time::Instant::now() + Duration::from_secs(30); loop { diff --git a/tests/celld/worker/src/lib.rs b/tests/celld/worker/src/lib.rs index a06e38f3c..21e9b7515 100644 --- a/tests/celld/worker/src/lib.rs +++ b/tests/celld/worker/src/lib.rs @@ -8,12 +8,18 @@ use std::time::Duration; use chat_domain::{post, ChatMessage, ChatMessageState}; -use distributed::cell_host::{AggregateCell, DurableCellEvents, DurableCellSnapshot}; -use distributed::microsvc::{HandlerError, Session, ROLE_KEY, USER_ID_KEY}; +use distributed::cell_host::{ + AggregateCell, CellCommandIdentity, CellDispatchError, CellDispatchResult, DurableCellCommand, + DurableCellEvents, DurableCellSnapshot, CELL_PRINCIPAL_PARTITION_HEADER, + CELL_SERVICE_ID_HEADER, +}; +use distributed::microsvc::{Session, ROLE_KEY, USER_ID_KEY}; use distributed::{EventRecord, OutboxMessage, OutboxMessageStatus}; use serde::Deserialize; use serde_json::{json, Value}; -use todo_domain::{complete, create, Todo, TodoState}; +use todo_domain::{ + archive, complete, create, force_archive, purge, rename, reopen, Todo, TodoState, +}; use worker::*; const EVENTS_DDL: &str = "CREATE TABLE IF NOT EXISTS cell_events ( @@ -38,6 +44,11 @@ const OUTBOX_DDL: &str = "CREATE TABLE IF NOT EXISTS cell_outbox ( body TEXT NOT NULL )"; +const COMMANDS_DDL: &str = "CREATE TABLE IF NOT EXISTS cell_commands ( + id TEXT PRIMARY KEY, + body TEXT NOT NULL +)"; + #[durable_object] pub struct TodoCell { cell: AggregateCell, @@ -57,17 +68,26 @@ impl DurableObject for TodoCell { .expect("create cell_snapshots"); sql.exec(SEALED_DDL, None).expect("create cell_sealed"); sql.exec(OUTBOX_DDL, None).expect("create cell_outbox"); + sql.exec(COMMANDS_DDL, None).expect("create cell_commands"); let shard = state.id().name().unwrap_or_else(|| "todo".to_string()); let cell = AggregateCell::::new_with_snapshots(shard.clone(), 1) .expect("todo cell identity") .mount(create()) - .mount(complete()); + .mount(rename()) + .mount(complete()) + .mount(reopen()) + .mount(archive()) + .mount(force_archive()) + .mount(purge()); if let Ok(events) = load_events(&sql) { let _ = cell.restore_durable_events(events); } if let Ok(snapshots) = load_snapshots(&sql) { let _ = cell.restore_durable_snapshots(snapshots); } + if let Ok(commands) = load_commands(&sql) { + let _ = cell.restore_durable_commands(commands); + } Self { cell, sql, @@ -106,13 +126,24 @@ impl DurableObject for TodoCell { ) .await } - (Method::Post, Some("todo.complete")) => { - complete_todo( + (Method::Post, Some(command)) + if matches!( + command, + "todo.rename" + | "todo.complete" + | "todo.reopen" + | "todo.archive" + | "todo.force_archive" + | "todo.purge" + ) => + { + transition_todo( &self.sql, &self.storage, &self.env, &self.cell, &id, + command, &mut req, ) .await @@ -150,6 +181,7 @@ impl DurableObject for ChatCell { sql.exec(EVENTS_DDL, None).expect("create cell_events"); sql.exec(SEALED_DDL, None).expect("create cell_sealed"); sql.exec(OUTBOX_DDL, None).expect("create cell_outbox"); + sql.exec(COMMANDS_DDL, None).expect("create cell_commands"); let shard = state.id().name().unwrap_or_else(|| "chat".to_string()); let cell = AggregateCell::::new(shard.clone()) .expect("chat cell identity") @@ -157,6 +189,9 @@ impl DurableObject for ChatCell { if let Ok(events) = load_events(&sql) { let _ = cell.restore_durable_events(events); } + if let Ok(commands) = load_commands(&sql) { + let _ = cell.restore_durable_commands(commands); + } Self { cell, sql, @@ -225,7 +260,7 @@ async fn main(req: Request, env: Env, _ctx: Context) -> Result { (Some("chat"), Some(id)) => ("CHAT", id), _ => { return Response::error( - "cells: GET|POST /todo/:id[/todo.create|todo.complete|outbox.drain|outbox.complete] GET|POST /chat/:id[/chat.post|outbox.drain|outbox.complete]\n", + "cells: GET|POST /todo/:id[/todo.|outbox.drain|outbox.complete] GET|POST /chat/:id[/chat.post|outbox.drain|outbox.complete]\n", 404, ); } @@ -283,31 +318,39 @@ async fn post_chat( ) -> Result { let session = request_session(req); let body = req.json::().await.unwrap_or(json!({})); - let (command_id, mut input) = wait_path_parts(&body); + let (command_id, mut input) = match wait_path_parts(&body) { + Ok(parts) => parts, + Err(error) => return map_cell_error(error, cell), + }; + let identity = match request_cell_identity(req, &command_id) { + Ok(identity) => identity, + Err(error) => return map_cell_error(error, cell), + }; if input.get("message_id").and_then(Value::as_str).is_none() { input .as_object_mut() .map(|object| object.insert("message_id".into(), json!(id))); } - match cell.dispatch("chat.post", input, session).await { - Ok(payload) => { + match cell + .dispatch_idempotent("chat.post", &identity, input, session) + .await + { + Ok(dispatch) => { seal_chat_from_load(cell).await; persist_chat_copy(sql, cell)?; arm_drain_alarm(storage, env, has_pending(cell)).await; - wait_path_ok(payload, command_id, 201, outbox_wire(cell)) + wait_path_ok( + dispatch.payload().clone(), + &dispatch, + 201, + outbox_wire(cell), + ) } - Err(HandlerError::Rejected(message)) if message.contains("already exists") => { + Err(error) => { + persist_chat_copy(sql, cell)?; arm_drain_alarm(storage, env, has_pending(cell)).await; - json_status( - json!({ - "error": "already exists", - "id": id, - "outbox": outbox_wire(cell), - }), - 409, - ) + map_cell_error(error, cell) } - Err(error) => map_handler_error(error), } } @@ -328,6 +371,9 @@ fn restore_chat_copy( let events = load_events(sql).map_err(|error| error.to_string())?; cell.restore_durable_events(events) .map_err(|error| error.to_string())?; + let commands = load_commands(sql).map_err(|error| error.to_string())?; + cell.restore_durable_commands(commands) + .map_err(|error| error.to_string())?; let outbox = load_outbox(sql).map_err(|error| error.to_string())?; cell.restore_durable_outbox(outbox) .map_err(|error| error.to_string())?; @@ -363,6 +409,7 @@ fn persist_chat_copy(sql: &SqlStorage, cell: &AggregateCell) -> Res )?; } } + persist_commands(sql, cell)?; persist_outbox(sql, cell)?; sql.exec("DELETE FROM cell_sealed", None)?; if let Ok(Some(row)) = cell.sealed_row() { @@ -387,34 +434,60 @@ async fn get_todo(cell: &AggregateCell, id: &str) -> Result { } } -fn wait_path_parts(body: &Value) -> (Option, Value) { +fn wait_path_parts(body: &Value) -> std::result::Result<(String, Value), CellDispatchError> { let command_id = body .get("commandId") .and_then(Value::as_str) .map(str::trim) .filter(|value| !value.is_empty()) - .map(str::to_string); + .map(str::to_string) + .ok_or_else(|| CellDispatchError::BadRequest("commandId is required".into()))?; let input = body.get("input").cloned().unwrap_or_else(|| body.clone()); - (command_id, input) + Ok((command_id, input)) +} + +fn request_cell_identity( + req: &Request, + command_id: &str, +) -> std::result::Result { + let service_id = required_internal_header(req, CELL_SERVICE_ID_HEADER)?; + let principal_partition = required_internal_header(req, CELL_PRINCIPAL_PARTITION_HEADER)?; + CellCommandIdentity::new(service_id, principal_partition, command_id) +} + +fn required_internal_header( + req: &Request, + name: &str, +) -> std::result::Result { + req.headers() + .get(name) + .map_err(|error| { + CellDispatchError::Internal(format!("could not read internal cell header: {error}")) + })? + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + .ok_or(CellDispatchError::Unauthorized) } fn wait_path_ok( payload: Value, - command_id: Option, + dispatch: &CellDispatchResult, status: u16, outbox: Value, ) -> Result { - match command_id { - Some(command_id) => json_status( - json!({ - "payload": payload, - "receipt": { "commandId": command_id, "state": "succeeded" }, - "outbox": outbox, - }), - status, - ), - None => json_status(payload, status), - } + json_status( + json!({ + "payload": payload, + "receipt": { + "commandId": dispatch.command_id(), + "causationId": dispatch.causation_id(), + "state": dispatch.state(), + "replayed": dispatch.replayed(), + }, + "outbox": outbox, + }), + status, + ) } fn outbox_wire(cell: &AggregateCell) -> Value @@ -449,64 +522,79 @@ async fn create_todo( req: &mut Request, ) -> Result { let body = req.json::().await.unwrap_or(json!({})); - let (command_id, input) = wait_path_parts(&body); + let (command_id, input) = match wait_path_parts(&body) { + Ok(parts) => parts, + Err(error) => return map_cell_error(error, cell), + }; + let identity = match request_cell_identity(req, &command_id) { + Ok(identity) => identity, + Err(error) => return map_cell_error(error, cell), + }; let title = input .get("title") .and_then(Value::as_str) .unwrap_or("") - .trim(); - if title.is_empty() { - return json_status(json!({ "error": "title required" }), 400); - } + .to_string(); match cell - .dispatch( + .dispatch_idempotent( "todo.create", + &identity, json!({ "todo_id": id, "title": title }), request_session(req), ) .await { - Ok(payload) => { + Ok(dispatch) => { seal_from_load(cell).await; persist_working_copy(sql, cell)?; arm_drain_alarm(storage, env, has_pending(cell)).await; wait_path_ok( - http_from_command(id, &payload, title), - command_id, + http_from_command(id, dispatch.payload(), &title), + &dispatch, 201, outbox_wire(cell), ) } - Err(HandlerError::Rejected(message)) if message.contains("already exists") => { + Err(error) => { + persist_working_copy(sql, cell)?; arm_drain_alarm(storage, env, has_pending(cell)).await; - json_status( - json!({ - "error": "already exists", - "id": id, - "outbox": outbox_wire(cell), - }), - 409, - ) + map_cell_error(error, cell) } - Err(error) => map_handler_error(error), } } -async fn complete_todo( +async fn transition_todo( sql: &SqlStorage, storage: &Storage, env: &Env, cell: &AggregateCell, id: &str, + command: &str, req: &mut Request, ) -> Result { let body = req.json::().await.unwrap_or(json!({})); - let (command_id, _input) = wait_path_parts(&body); + let (command_id, mut input) = match wait_path_parts(&body) { + Ok(parts) => parts, + Err(error) => return map_cell_error(error, cell), + }; + let identity = match request_cell_identity(req, &command_id) { + Ok(identity) => identity, + Err(error) => return map_cell_error(error, cell), + }; + let Some(input_object) = input.as_object_mut() else { + return map_cell_error( + CellDispatchError::BadRequest("input must be an object".into()), + cell, + ); + }; + input_object + .entry("todo_id".to_string()) + .or_insert_with(|| json!(id)); match cell - .dispatch("todo.complete", json!({ "todo_id": id }), request_session(req)) + .dispatch_idempotent(command, &identity, input, request_session(req)) .await { - Ok(payload) => { + Ok(dispatch) => { seal_from_load(cell).await; persist_working_copy(sql, cell)?; arm_drain_alarm(storage, env, has_pending(cell)).await; @@ -518,23 +606,17 @@ async fn complete_todo( .map(|todo| TodoState::from(&todo).title) .unwrap_or_default(); wait_path_ok( - http_from_command(id, &payload, &title), - command_id, + http_from_command(id, dispatch.payload(), &title), + &dispatch, 200, outbox_wire(cell), ) } - Err(HandlerError::NotFound(_)) => { - json_status(json!({ "error": "not found", "id": id }), 404) - } - Err(HandlerError::Rejected(message)) if message.to_lowercase().contains("not found") => { - json_status(json!({ "error": "not found", "id": id }), 404) + Err(error) => { + persist_working_copy(sql, cell)?; + arm_drain_alarm(storage, env, has_pending(cell)).await; + map_cell_error(error, cell) } - Err(HandlerError::Rejected(message)) if message.contains("not open") => json_status( - json!({ "error": "not open", "id": id, "status": "completed" }), - 422, - ), - Err(error) => map_handler_error(error), } } @@ -547,24 +629,32 @@ fn http_todo(state: &TodoState) -> Value { } fn http_from_command(id: &str, payload: &Value, fallback_title: &str) -> Value { - json!({ - "id": payload.get("todo_id").cloned().unwrap_or_else(|| json!(id)), - "todo_id": payload.get("todo_id").cloned().unwrap_or_else(|| json!(id)), - "owner_id": payload.get("owner_id").cloned().unwrap_or(json!("")), - "title": payload.get("title").cloned().unwrap_or_else(|| json!(fallback_title)), - "status": payload.get("status").cloned().unwrap_or_else(|| json!("open")), - }) + let mut body = payload.as_object().cloned().unwrap_or_default(); + body.entry("id".to_string()).or_insert_with(|| json!(id)); + body.entry("todo_id".to_string()) + .or_insert_with(|| json!(id)); + body.entry("owner_id".to_string()) + .or_insert_with(|| json!("")); + body.entry("title".to_string()) + .or_insert_with(|| json!(fallback_title)); + body.entry("status".to_string()) + .or_insert_with(|| json!("open")); + Value::Object(body) } -fn map_handler_error(error: HandlerError) -> Result { - let status = match &error { - HandlerError::NotFound(_) => 404, - HandlerError::Unauthorized(_) | HandlerError::GuardRejected(_) => 401, - HandlerError::Rejected(_) => 422, - HandlerError::DecodeFailed(_) => 400, - _ => 500, - }; - json_status(json!({ "error": error.to_string() }), status) +fn map_cell_error(error: CellDispatchError, cell: &AggregateCell) -> Result +where + A: distributed::Aggregate + Send + Sync + 'static, +{ + let status = error.status_code(); + json_status( + json!({ + "error": error.client_message(), + "code": error.code(), + "outbox": outbox_wire(cell), + }), + status, + ) } fn json_status(body: Value, status: u16) -> Result { @@ -589,6 +679,9 @@ fn restore_working_copy( let snapshots = load_snapshots(sql).map_err(|error| error.to_string())?; cell.restore_durable_snapshots(snapshots) .map_err(|error| error.to_string())?; + let commands = load_commands(sql).map_err(|error| error.to_string())?; + cell.restore_durable_commands(commands) + .map_err(|error| error.to_string())?; let outbox = load_outbox(sql).map_err(|error| error.to_string())?; cell.restore_durable_outbox(outbox) .map_err(|error| error.to_string())?; @@ -636,6 +729,7 @@ fn persist_working_copy(sql: &SqlStorage, cell: &AggregateCell) -> Result< Some(vec![snapshot.stream.into(), body.into()]), )?; } + persist_commands(sql, cell)?; persist_outbox(sql, cell)?; sql.exec("DELETE FROM cell_sealed", None)?; if let Ok(Some(row)) = cell.sealed_row() { @@ -668,6 +762,28 @@ where Ok(()) } +fn persist_commands(sql: &SqlStorage, cell: &AggregateCell) -> Result<()> +where + A: distributed::Aggregate + Send + Sync + 'static, +{ + let rows = cell + .durable_commands() + .map_err(|error| Error::RustError(error.to_string()))?; + sql.exec("DELETE FROM cell_commands", None)?; + for command in rows { + sql.exec( + "INSERT INTO cell_commands (id, body) VALUES (?, ?)", + Some(vec![command.id.into(), command.body.into()]), + )?; + } + Ok(()) +} + +fn load_commands(sql: &SqlStorage) -> Result> { + sql.exec("SELECT id, body FROM cell_commands ORDER BY id", None)? + .to_array() +} + fn outbox_item(message: &OutboxMessage) -> Value { json!({ "id": message.id, @@ -720,11 +836,7 @@ fn ids_from_body(body: &Value) -> Vec { .unwrap_or_default() } -fn mark_outbox_published( - sql: &SqlStorage, - cell: &AggregateCell, - ids: &[String], -) -> Result<()> +fn mark_outbox_published(sql: &SqlStorage, cell: &AggregateCell, ids: &[String]) -> Result<()> where A: distributed::Aggregate + Send + Sync + 'static, { @@ -805,7 +917,9 @@ async fn offer_pending(env: &Env, kind: &str, id: &str, outbox: &Value) { let mut init = RequestInit::new(); init.with_method(Method::Post) .with_headers(headers) - .with_body(Some(worker::wasm_bindgen::JsValue::from_str(&payload.to_string()))); + .with_body(Some(worker::wasm_bindgen::JsValue::from_str( + &payload.to_string(), + ))); if let Ok(req) = Request::new_with_init(&url, &init) { let _ = Fetch::Request(req).send().await; } diff --git a/tests/e2e-celld/crates/graphql-service/src/host.rs b/tests/e2e-celld/crates/graphql-service/src/host.rs index 6d116a6b6..132f72a58 100644 --- a/tests/e2e-celld/crates/graphql-service/src/host.rs +++ b/tests/e2e-celld/crates/graphql-service/src/host.rs @@ -13,7 +13,7 @@ use distributed::bus::MessagePublisher; use distributed::BusPublisher; use distributed::graphql::IdentityConfig; use distributed::microsvc::{ - spawn_outbox_publish_loop, spawn_service_consumer_loop, Service, + spawn_outbox_publish_loop, spawn_service_consumer_loop, Service, CONSUMER_IDLE_POLL, }; use distributed::{PostgresLockManager, PostgresRepository}; @@ -83,7 +83,8 @@ async fn run_postgres( let locks = locks.clone(); let nats = nats.clone(); spawn_service_consumer_loop(move || { - build_service(repo.clone(), locks.clone(), repo.clone()).with_bus(nats.clone()) + build_service(repo.clone(), locks.clone(), repo.clone()) + .with_bus(nats.clone().with_idle_poll(CONSUMER_IDLE_POLL)) }); } spawn_zitadel_scrape(repo.clone()); @@ -129,7 +130,12 @@ async fn connect_nats( fn spawn_zitadel_scrape(repo: R) where - R: distributed::TransactionalCommit + Clone + Send + Sync + 'static, + R: distributed::TransactionalCommit + + distributed::ReadModelWritePlanStore + + Clone + + Send + + Sync + + 'static, { match ZitadelScrapeConfig::from_env() { Some(cfg) if cfg.background_enabled() || cfg.on_start => { diff --git a/tests/e2e-celld/crates/identity-service/src/handlers/ingestors/zitadel/scrape.rs b/tests/e2e-celld/crates/identity-service/src/handlers/ingestors/zitadel/scrape.rs index 2c3c3214e..c988fcabb 100644 --- a/tests/e2e-celld/crates/identity-service/src/handlers/ingestors/zitadel/scrape.rs +++ b/tests/e2e-celld/crates/identity-service/src/handlers/ingestors/zitadel/scrape.rs @@ -6,8 +6,11 @@ use std::env; use std::time::Duration; -use distributed::TransactionalCommit; -use e2e_projections::{ZitadelEmail, ZitadelUserPayload}; +use distributed::read_model::ReadModelWritePlanBuilder; +use distributed::{ReadModelWritePlanStore, TransactionalCommit}; +use e2e_projections::{ + map_zitadel_user_status, map_zitadel_user_upsert, ZitadelEmail, ZitadelUserPayload, +}; use serde::Deserialize; use serde_json::{json, Value}; @@ -92,10 +95,15 @@ pub struct ScrapeReport { } /// List users from Zitadel Management API and publish provider messages for each. -pub async fn scrape_users_to_outbox( - repo: &R, +pub async fn scrape_users_to_outbox( + outbox: &R, + directory: &S, cfg: &ZitadelScrapeConfig, -) -> ScrapeReport { +) -> ScrapeReport +where + R: TransactionalCommit, + S: ReadModelWritePlanStore, +{ let mut report = ScrapeReport::default(); let users = match list_all_users(cfg).await { Ok(u) => u, @@ -111,12 +119,16 @@ pub async fn scrape_users_to_outbox( report.skipped += 1; continue; }; - match publish_mapped_delivery(repo, &mapped).await { + if let Err(e) = materialize_auth_user(directory, &mapped).await { + report.errors.push(format!( + "user {}: auth_users upsert failed: {e}", + mapped.payload.provider_subject + )); + continue; + } + match publish_mapped_delivery(outbox, &mapped).await { Ok(()) => report.published += 1, Err(e) => { - // Content-addressed scrape ids: unchanged profile re-scrape hits the - // outbox unique key. That is the durable "already emitted" cache — - // count as skip, not error. if is_expected_scrape_duplicate(&e) { report.skipped += 1; } else { @@ -131,6 +143,22 @@ pub async fn scrape_users_to_outbox( report } +async fn materialize_auth_user( + store: &S, + mapped: &MappedDelivery, +) -> Result<(), String> { + let name = mapped.message_name.as_str(); + let row = if name.contains("deactivated") || name.contains("reactivated") { + map_zitadel_user_status(name, &mapped.payload) + } else { + map_zitadel_user_upsert(name, &mapped.payload) + }; + let mut plan = ReadModelWritePlanBuilder::new(); + plan.upsert(&row).map_err(|e| e.to_string())?; + plan.commit(store).await.map_err(|e| e.to_string())?; + Ok(()) +} + /// True when publish failed because this scrape delivery id was already committed. /// /// Matches repository `DuplicateOutboxMessageInBatch` display text and common @@ -385,14 +413,14 @@ fn now_ms() -> String { /// Background loop: optional immediate scrape, then every `cfg.interval`. pub fn spawn_scrape_loop(repo: R, cfg: ZitadelScrapeConfig) where - R: TransactionalCommit + Clone + Send + Sync + 'static, + R: TransactionalCommit + ReadModelWritePlanStore + Clone + Send + Sync + 'static, { if !cfg.background_enabled() && !cfg.on_start { return; } tokio::spawn(async move { if cfg.on_start { - let r = scrape_users_to_outbox(&repo, &cfg).await; + let r = scrape_users_to_outbox(&repo, &repo, &cfg).await; eprintln!( "zitadel scrape (start): listed={} published={} skipped={} errors={}", r.listed, @@ -409,7 +437,7 @@ where } loop { tokio::time::sleep(cfg.interval).await; - let r = scrape_users_to_outbox(&repo, &cfg).await; + let r = scrape_users_to_outbox(&repo, &repo, &cfg).await; if r.published > 0 || !r.errors.is_empty() { eprintln!( "zitadel scrape: listed={} published={} skipped={} errors={}", diff --git a/tests/e2e-celld/crates/identity-service/src/handlers/ingestors/zitadel_scrape.rs b/tests/e2e-celld/crates/identity-service/src/handlers/ingestors/zitadel_scrape.rs index 2b24f17d5..8b8e5ba82 100644 --- a/tests/e2e-celld/crates/identity-service/src/handlers/ingestors/zitadel_scrape.rs +++ b/tests/e2e-celld/crates/identity-service/src/handlers/ingestors/zitadel_scrape.rs @@ -26,7 +26,7 @@ pub async fn handle(ctx: &Context<'_, AuthDeps>) -> Result CelldRoute { - CelldRoute::new( - &["todo.create", "todo.complete"], - "todo", - todo_shard, - graphql_todo_payload, - ) + CelldRoute::new(TODO_CELL_COMMANDS, "todo", todo_shard, graphql_todo_payload) } fn todo_shard(input: &Value) -> Option { @@ -31,17 +36,16 @@ fn graphql_todo_payload(command: &str, input: &Value, remote: &Value, session: & .or_else(|| input.get("todo_id")) .cloned() .unwrap_or(json!("")); - let status = remote.get("status").cloned().unwrap_or_else(|| { - if command == "todo.complete" { - json!("completed") - } else { - json!("open") - } - }); - if command == "todo.complete" { - json!({ "todo_id": id, "status": status }) - } else { - json!({ + let status = remote + .get("status") + .cloned() + .unwrap_or_else(|| match command { + "todo.complete" => json!("completed"), + "todo.archive" | "todo.force_archive" => json!("archived"), + _ => json!("open"), + }); + match command { + "todo.create" => json!({ "todo_id": id, "owner_id": remote .get("owner_id") @@ -50,6 +54,39 @@ fn graphql_todo_payload(command: &str, input: &Value, remote: &Value, session: & .unwrap_or(json!("")), "title": remote.get("title").or_else(|| input.get("title")).cloned().unwrap_or(json!("")), "status": status, - }) + }), + "todo.rename" => json!({ + "todo_id": id, + "title": remote.get("title").or_else(|| input.get("title")).cloned().unwrap_or(json!("")), + "status": status, + }), + "todo.complete" | "todo.reopen" | "todo.archive" => { + json!({ "todo_id": id, "status": status }) + } + "todo.force_archive" => json!({ + "todo_id": id, + "owner_id": remote.get("owner_id").cloned().unwrap_or(json!("")), + "status": status, + "archived_by": remote + .get("archived_by") + .cloned() + .or_else(|| session.user_id().map(|id| json!(id))) + .unwrap_or(json!("")), + }), + "todo.purge" => json!({ + "todo_id": id, + "purged": remote.get("purged").cloned().unwrap_or(json!(true)), + }), + _ => remote.clone(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn celld_route_keeps_every_todo_transition_on_one_shard() { + assert_eq!(celld_route().commands, TODO_CELL_COMMANDS); } } diff --git a/tests/e2e-celld/crates/todo-service/src/lib.rs b/tests/e2e-celld/crates/todo-service/src/lib.rs index ce2a51097..294e3b8c7 100644 --- a/tests/e2e-celld/crates/todo-service/src/lib.rs +++ b/tests/e2e-celld/crates/todo-service/src/lib.rs @@ -1,7 +1,7 @@ //! Todo service crate for the celld example. //! -//! Domain commands stay in `todo-domain`. Wait-dispatch create/complete to -//! celld through [`distributed::cell_host::CelldCommandHost`]. +//! Domain commands stay in `todo-domain`. Every Todo aggregate transition is +//! wait-dispatched to celld through [`distributed::cell_host::CelldCommandHost`]. mod bounds; mod handlers; diff --git a/tests/e2e-ui/crates/service/src/host.rs b/tests/e2e-ui/crates/service/src/host.rs index 61ea2dfcb..d16589d09 100644 --- a/tests/e2e-ui/crates/service/src/host.rs +++ b/tests/e2e-ui/crates/service/src/host.rs @@ -13,7 +13,9 @@ use std::time::Duration; use distributed::bus::{PostgresBus, SqliteBus}; use distributed::command_dispatch::LocalCommandDispatcher; use distributed::graphql::IdentityConfig; -use distributed::microsvc::{spawn_outbox_publish_loop, spawn_service_consumer_loop}; +use distributed::microsvc::{ + spawn_outbox_publish_loop, spawn_service_consumer_loop, CONSUMER_IDLE_POLL, +}; use distributed::{PostgresLockManager, PostgresRepository, SqliteLockManager, SqliteRepository}; use crate::{ @@ -78,7 +80,7 @@ async fn run_sqlite( spawn_service_consumer_loop(move || { let bus = SqliteBus::new(repo.pool().clone()) .group(BUS_GROUP) - .with_idle_poll(Duration::from_millis(25)); + .with_idle_poll(CONSUMER_IDLE_POLL); build_service(repo.clone(), locks.clone(), repo.clone()).with_bus(bus) }); } @@ -122,7 +124,7 @@ async fn run_postgres( spawn_service_consumer_loop(move || { let bus = PostgresBus::new(repo.pool().clone()) .group(BUS_GROUP) - .with_idle_poll(Duration::from_millis(25)); + .with_idle_poll(CONSUMER_IDLE_POLL); build_service(repo.clone(), locks.clone(), repo.clone()).with_bus(bus) }); } From 1c0a488c97695d2bf38607e723aa56b5b66529c1 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Mon, 24 Aug 2026 01:00:36 -0500 Subject: [PATCH 2/4] fix(replica): avoid stale celld revalidation races Trust exact authenticated projection deltas instead of refetching solely because they have no obligations. Preserve conservative revalidation for unconditional recovery cases, cover rapid Todo transitions, and keep newly-created Todo controls pending until the durable receipt arrives. --- .../client_compiler/manifest/projections.rs | 7 - .../projection_delta/preview.rs | 5 +- distributed_cli/src/client_compiler/tests.rs | 2 + js/src/replica/command-id.ts | 7 +- js/src/replica/command-runtime/create.ts | 8 +- js/src/replica/index.ts | 1 + js/src/replica/projection-delta/resolve.ts | 4 +- js/tests/replica-command-artifacts.test.mjs | 35 ++++- js/tests/replica-command-runtime.test.mjs | 7 +- src/microsvc/service/causal.rs | 7 +- tests/e2e-ui/e2e/todos.user.spec.ts | 133 +++++++++++++++++- .../ui/src/lib/generated/admin/commands.ts | 16 +-- .../ui/src/lib/generated/admin/manifest.json | 12 +- .../ui/src/lib/generated/user/commands.ts | 14 +- .../ui/src/lib/generated/user/manifest.json | 10 +- tests/e2e-ui/ui/src/routes/todos/+page.svelte | 46 +++++- 16 files changed, 245 insertions(+), 69 deletions(-) diff --git a/distributed_cli/src/client_compiler/manifest/projections.rs b/distributed_cli/src/client_compiler/manifest/projections.rs index f11f176b2..8f57977c9 100644 --- a/distributed_cli/src/client_compiler/manifest/projections.rs +++ b/distributed_cli/src/client_compiler/manifest/projections.rs @@ -319,13 +319,6 @@ pub(crate) fn validate_command_projections( .get(&value.slot) .expect("exact slot coverage was validated"); validate_preview_source(command, &value.source, expected)?; - if matches!( - value.source, - ManifestProjectionPreviewSource::Unknown - | ManifestProjectionPreviewSource::Absent - ) { - requiring_revalidation.insert(command.name.clone()); - } } } if selected_programs.is_empty() { diff --git a/distributed_cli/src/client_compiler/projection_delta/preview.rs b/distributed_cli/src/client_compiler/projection_delta/preview.rs index e199389e7..4d6bb4855 100644 --- a/distributed_cli/src/client_compiler/projection_delta/preview.rs +++ b/distributed_cli/src/client_compiler/projection_delta/preview.rs @@ -165,7 +165,10 @@ impl CompiledCommandProjection { } pub(crate) fn requires_revalidation(&self) -> bool { - !self.preview.recoveries.is_empty() + self.preview + .recoveries + .iter() + .any(|recovery| recovery.condition == PreviewRecoveryCondition::Always) } pub(crate) fn selected_models(&self) -> &BTreeSet { diff --git a/distributed_cli/src/client_compiler/tests.rs b/distributed_cli/src/client_compiler/tests.rs index 78ebc1305..f0f90daeb 100644 --- a/distributed_cli/src/client_compiler/tests.rs +++ b/distributed_cli/src/client_compiler/tests.rs @@ -3221,6 +3221,7 @@ fn command_protocol_and_extensions_are_preserved_exactly() { assert!(!partial_commands.contains("\"op\": \"upsert\"")); assert!(partial_commands.contains("\"condition\": \"if_record_missing\"")); assert!(partial_commands.contains("\"kind\": \"record\"")); + assert!(partial_commands.contains("\"required\": false")); let absent = compile_client(ClientCompileInput::new( absent_value, @@ -3290,6 +3291,7 @@ fn command_protocol_and_extensions_are_preserved_exactly() { assert!(fallback_commands.contains("\"kind\": \"model\"")); assert!(!fallback_commands.contains("\"op\": \"upsert\"")); assert!(!fallback_commands.contains("\"op\": \"patch\"")); + assert!(fallback_commands.contains("\"required\": true")); } let delete = compile_client(ClientCompileInput::new( diff --git a/js/src/replica/command-id.ts b/js/src/replica/command-id.ts index ee714ec34..7546a51d7 100644 --- a/js/src/replica/command-id.ts +++ b/js/src/replica/command-id.ts @@ -1,5 +1,5 @@ -/** Create a browser/Node UUIDv7 command identity. */ -export function createReplicaCommandId(): string { +/** Create a browser/Node UUIDv7 identity for commands or generated record IDs. */ +export function createReplicaUuidV7(): string { const crypto = globalThis.crypto; if (!crypto || typeof crypto.getRandomValues !== 'function') { throw new Error('replica commands require crypto.getRandomValues'); @@ -19,3 +19,6 @@ export function createReplicaCommandId(): string { .slice(6, 8) .join('')}-${hex.slice(8, 10).join('')}-${hex.slice(10).join('')}`; } + +/** Package-internal semantic alias used while preparing command envelopes. */ +export const createReplicaCommandId = createReplicaUuidV7; diff --git a/js/src/replica/command-runtime/create.ts b/js/src/replica/command-runtime/create.ts index 64b4bc067..b353a843e 100644 --- a/js/src/replica/command-runtime/create.ts +++ b/js/src/replica/command-runtime/create.ts @@ -555,8 +555,7 @@ export function createReplicaCommandRuntime< throw new Error('projection delta changed during command replay'); } return Object.freeze({ - requiresRevalidation: - actual.revalidate || actual.obligations.length === 0 + requiresRevalidation: actual.revalidate }); } assertActualProjectionCapabilities( @@ -572,14 +571,13 @@ export function createReplicaCommandRuntime< canonical, operations, revalidation: - actual.revalidate || actual.obligations.length === 0 + actual.revalidate ? actualProjectionRevalidation( prepared.revalidation, actual.delta ) : undefined, - requiresRevalidation: - actual.revalidate || actual.obligations.length === 0 + requiresRevalidation: actual.revalidate }); }; diff --git a/js/src/replica/index.ts b/js/src/replica/index.ts index 4637414eb..f8a96fa92 100644 --- a/js/src/replica/index.ts +++ b/js/src/replica/index.ts @@ -36,6 +36,7 @@ export type { ReplicaOperationInjectedFieldInspection } from './diagnostics.js'; export { createReplicaGraphqlTransport } from './graphql-transport.js'; +export { createReplicaUuidV7 } from './command-id.js'; export type { ReplicaGraphqlTransport, ReplicaGraphqlTransportOptions diff --git a/js/src/replica/projection-delta/resolve.ts b/js/src/replica/projection-delta/resolve.ts index de9408c60..3a3dd2bde 100644 --- a/js/src/replica/projection-delta/resolve.ts +++ b/js/src/replica/projection-delta/resolve.ts @@ -32,7 +32,9 @@ export function prepareCommandProjection( return Object.freeze({ contract, preview: Object.freeze([...preview, ...pure]), - revalidate: contract.preview.recoveries.length !== 0 + revalidate: contract.preview.recoveries.some( + (recovery) => recovery.condition === 'always' + ) }); } catch { // Preview is only a convenience. Missing client authority must never be diff --git a/js/tests/replica-command-artifacts.test.mjs b/js/tests/replica-command-artifacts.test.mjs index 918b0d22c..b4a0f0f6e 100644 --- a/js/tests/replica-command-artifacts.test.mjs +++ b/js/tests/replica-command-artifacts.test.mjs @@ -3,6 +3,7 @@ import { readFileSync } from 'node:fs'; import test from 'node:test'; import { + createReplicaUuidV7, prepareReplicaCommand, ReplicaCommandContractError, verifyReplicaCommandReceipt @@ -136,7 +137,7 @@ function projectionScope(model, ...fields) { }); } -function projectionArtifact(operations, capabilities) { +function projectionArtifact(operations, capabilities, recoveries = []) { return Object.freeze({ version: 2, deltaWireVersion: 1, @@ -178,7 +179,7 @@ function projectionArtifact(operations, capabilities) { }) ) ), - recoveries: Object.freeze([]) + recoveries: Object.freeze(recoveries) }), fallback: 'revalidate' }); @@ -385,15 +386,16 @@ test('explicit defaulted fields are retained and their generators never run', () }); test('compact generated preview patches canonicalize an omitted unset list', () => { + const scope = projectionScope( + 'Todo', + projectionField('id', inputValue(['id'])) + ); const artifact = baseArtifact({ projection: projectionArtifact( [ Object.freeze({ op: 'patch', - scope: projectionScope( - 'Todo', - projectionField('id', inputValue(['id'])) - ), + scope, set: Object.freeze([ projectionField('title', inputValue(['title'])) ]), @@ -411,6 +413,14 @@ test('compact generated preview patches canonicalize an omitted unset list', () patch: true, delete: false }) + ], + [ + Object.freeze({ + occurrence_ordinal: 0, + projection_refs: Object.freeze([0]), + condition: 'if_record_missing', + target: Object.freeze({ kind: 'record', scope }) + }) ] ) }); @@ -441,6 +451,11 @@ test('compact generated preview patches canonicalize an omitted unset list', () const unset = prepared.optimistic.operations[0].unset; assert.equal(Object.isFrozen(unset), true); assert.throws(() => unset.push('title'), TypeError); + assert.equal( + prepared.projection.revalidate, + false, + 'a conditional missing-record fallback is not unconditional revalidation' + ); }); test('real default generators produce canonical values', () => { @@ -454,6 +469,14 @@ test('real default generators produce canonical values', () => { assert.match(prepared.input.code, /^[0-7][0-9A-HJKMNP-TV-Z]{25}$/); }); +test('public UUIDv7 generation supports caller-correlated optimistic record ids', () => { + const id = createReplicaUuidV7(); + assert.match( + id, + /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/ + ); +}); + test('none inputs and typed JSON fields produce exact canonical transport variables', () => { const noInput = prepareReplicaCommand( baseArtifact({ diff --git a/js/tests/replica-command-runtime.test.mjs b/js/tests/replica-command-runtime.test.mjs index 60dacb914..eebbef219 100644 --- a/js/tests/replica-command-runtime.test.mjs +++ b/js/tests/replica-command-runtime.test.mjs @@ -1558,7 +1558,7 @@ test('an allowed unpreviewed event arm can authoritatively replace the preview', runtime.dispose(); }); -test('zero-obligation revalidation includes actual unpreviewed target models', async () => { +test('explicit revalidation includes actual unpreviewed target models', async () => { const modeled = modeledArtifactWithAuditArm(); const replica = new TestReplica(); const runtime = createReplicaCommandRuntime( @@ -1568,6 +1568,7 @@ test('zero-obligation revalidation includes actual unpreviewed target models', a Promise.resolve( envelope(request, { obligations: 0, + revalidate: true, mutation: { op: 'upsert', scope: scope( @@ -1768,8 +1769,8 @@ test('zero, one, and many obligations are server-derived and never predicted key assert.equal(receipt.projected === undefined, count === 0); if (count === 0) { await tick(); - assert.equal(replica.revalidations.length, 1); - assert.equal(replica.layer(receipt.commandId), undefined); + assert.equal(replica.revalidations.length, 0); + assert.equal(replica.layer(receipt.commandId), 'accepted'); } runtime.dispose(); } diff --git a/src/microsvc/service/causal.rs b/src/microsvc/service/causal.rs index b2f2ab1db..ba743aa17 100644 --- a/src/microsvc/service/causal.rs +++ b/src/microsvc/service/causal.rs @@ -318,17 +318,20 @@ impl CausalDispatchResult { // Cell wait-path has no GraphQL command-ledger observations. Keep the // modeled delta so the replica can apply it, but drop expects so // `projected` does not wait on live/status observations this process - // cannot emit. The client revalidates; SQL `@live` still catches up. + // cannot emit. Preserve the delta's own recovery disposition: a fully + // resolved actual delta is sufficient local authority and must not turn + // every successful cell command into a full-query revalidation. let metadata = if metadata.obligations.is_empty() { metadata } else { + let revalidate = metadata.revalidate; crate::graphql::protocol::CommandProjectionMetadataV1::try_new( metadata.issued_at_unix_ms, metadata.expires_at_unix_ms, metadata.delta, metadata.lifecycle_proofs, Vec::new(), - true, + revalidate, ) .map_err(|error| { CausalDispatchError::Internal(format!( diff --git a/tests/e2e-ui/e2e/todos.user.spec.ts b/tests/e2e-ui/e2e/todos.user.spec.ts index d3a97a11f..cb350de29 100644 --- a/tests/e2e-ui/e2e/todos.user.spec.ts +++ b/tests/e2e-ui/e2e/todos.user.spec.ts @@ -181,7 +181,7 @@ test.describe('todos (alice)', () => { await expect(add).toBeDisabled(); }); - test('commands preserve rendered cache while revalidating', async ({ page }) => { + test('commands preserve rendered cache while settling', async ({ page }) => { await page.goto('/todos'); await expect(page.getByRole('heading', { name: /todos/i })).toBeVisible(); @@ -285,15 +285,20 @@ test.describe('todos (alice)', () => { // turns that event value into the upsert that must paint before the held // Eventual response. await expect(openItem).toBeVisible({ timeout: 1_000 }); + await expect(openItem).toHaveAttribute('aria-busy', 'true'); + await expect(openItem).toHaveClass(/item-pending/); + await expect(openItem.locator('.pending-state')).toHaveText('Saving…'); expect( - await page.locator('.board button:disabled').count(), - 'routine command concurrency guards must not flash Todo row controls disabled' - ).toBe(0); + await openItem.locator('button:disabled').count(), + 'a newly created optimistic Todo must not expose actions before its receipt' + ).toBe(3); await createResponse; await expect(openItem).toBeVisible(); + await expect(openItem).toHaveAttribute('aria-busy', 'false'); + await expect(openItem.locator('.pending-state')).toHaveCount(0); expect( await page.locator('.board button:disabled').count(), - 'routine command concurrency guards must not flash Todo row controls disabled' + 'Todo controls must unlock after its durable create receipt' ).toBe(0); expectBinarySorted(await visibleTodoOrders(page)); const todoId = await openItem.getAttribute('data-todo-id'); @@ -419,4 +424,122 @@ test.describe('todos (alice)', () => { await expect(archivedItem).toBeVisible(); await page.unrouteAll({ behavior: 'wait' }); }); + + test('rapid independent complete and reopen commands do not refetch or regress', async ({ + page + }) => { + await page.goto('/todos'); + await expect(page.getByRole('heading', { name: /todos/i })).toBeVisible(); + + const prefix = `rapid transitions ${Date.now()}`; + const titles = Array.from({ length: 6 }, (_, index) => `${prefix} ${index + 1}`); + const todoIds: string[] = []; + for (const title of titles) { + await page.locator('#todo-title').fill(title); + const response = waitForTodoCommand(page, 'todos_create'); + await page.getByRole('button', { name: /^add$/i }).click(); + expect((await response).ok(), 'setup todos_create must succeed').toBeTruthy(); + const item = page.locator('.item', { hasText: title }); + await expect(item).toBeVisible(); + const todoId = await item.getAttribute('data-todo-id'); + expect(todoId).not.toBeNull(); + todoIds.push(todoId!); + } + await page.waitForLoadState('networkidle'); + + let transitionQueries = 0; + let transitionResponses = 0; + page.on('request', (request) => { + const body = request.postData() ?? ''; + if (body.includes('query Todos')) transitionQueries += 1; + }); + page.on('response', (response) => { + const body = response.request().postData() ?? ''; + if (body.includes('todos_complete') || body.includes('todos_reopen')) { + transitionResponses += 1; + } + }); + await page.route('**/graphql', async (route) => { + const body = route.request().postData() ?? ''; + if (!body.includes('todos_complete') && !body.includes('todos_reopen')) { + await route.continue(); + return; + } + const response = await route.fetch(); + await new Promise((resolve) => setTimeout(resolve, 350)); + await route.fulfill({ response }); + }); + + const openPanel = page + .locator('.panel') + .filter({ has: page.getByRole('heading', { name: /^open$/i }) }); + const donePanel = page + .locator('.panel') + .filter({ has: page.getByRole('heading', { name: /^done$/i }) }); + + await startTodoOrderTrace(page); + for (const title of titles) { + await openPanel + .locator('.item', { hasText: title }) + .getByRole('button', { name: /^done$/i }) + .click(); + } + for (const title of titles) { + const item = donePanel.locator('.item', { hasText: title }); + await expect(item).toBeVisible({ timeout: 1_000 }); + await item.getByRole('button', { name: /^reopen$/i }).click(); + } + + for (const title of titles) { + await expect(openPanel.locator('.item', { hasText: title })).toBeVisible({ + timeout: 1_000 + }); + } + await expect + .poll(() => transitionResponses, { timeout: 20_000 }) + .toBe(titles.length * 2); + await page.waitForTimeout(750); + + const frames = await stopTodoOrderTrace(page); + const allDoneFrame = frames.findIndex((frame) => + todoIds.every((todoId) => todoIsIn(frame, todoId, 'done')) + ); + const fullyReopenedFrame = frames.findIndex( + (frame, index) => + index > allDoneFrame && + todoIds.every((todoId) => todoIsIn(frame, todoId, 'open')) + ); + for (const title of titles) { + await expect(openPanel.locator('.item', { hasText: title })).toBeVisible(); + await expect(donePanel.locator('.item', { hasText: title })).toHaveCount(0); + } + expect( + transitionQueries, + 'exact successful Todo deltas must not launch a full-list revalidation' + ).toBe(0); + expect(allDoneFrame, `rapid transitions never reached Done: ${JSON.stringify(frames)}`).toBeGreaterThanOrEqual( + 0 + ); + expect( + fullyReopenedFrame, + `rapid transitions never fully reopened: ${JSON.stringify(frames)}` + ).toBeGreaterThan(allDoneFrame); + expect( + frames + .slice(fullyReopenedFrame) + .every((frame) => todoIds.every((todoId) => todoIsIn(frame, todoId, 'open'))), + `a stale result regressed a fully reopened Todo: ${JSON.stringify(frames)}` + ).toBe(true); + expect( + frames.every( + (frame) => + isBinarySorted(frame.open) && + isBinarySorted(frame.done) && + todoIds.every((todoId) => validTodoTransitionFrame(frame, todoId)) + ), + `rapid transitions rendered an invalid generated order: ${JSON.stringify(frames)}` + ).toBe(true); + + await page.unrouteAll({ behavior: 'wait' }); + }); }); diff --git a/tests/e2e-ui/ui/src/lib/generated/admin/commands.ts b/tests/e2e-ui/ui/src/lib/generated/admin/commands.ts index 5c5e875c6..965f57f5f 100644 --- a/tests/e2e-ui/ui/src/lib/generated/admin/commands.ts +++ b/tests/e2e-ui/ui/src/lib/generated/admin/commands.ts @@ -678,7 +678,7 @@ export const Command_blob_games_start: ReplicaCommandArtifact(null); + let pendingCreateIds = $state>(new Set()); const who = $derived(sessionDisplayName(data.session)); @@ -40,8 +42,13 @@ actionError = null; title = ''; + const todoId = createReplicaUuidV7(); + pendingCreateIds = new Set(pendingCreateIds).add(todoId); try { - await commands.todo.create({ title: text }); + await commands.todo.create({ title: text, todo_id: todoId }); + const next = new Set(pendingCreateIds); + next.delete(todoId); + pendingCreateIds = next; } catch (error) { actionError = error instanceof Error ? error.message : 'create failed'; } @@ -137,16 +144,26 @@ {:else}