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/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/scripts/pack-smoke.mjs b/js/scripts/pack-smoke.mjs index 65ea158fd..0ff98f3eb 100644 --- a/js/scripts/pack-smoke.mjs +++ b/js/scripts/pack-smoke.mjs @@ -430,6 +430,7 @@ assert.deepEqual(Object.keys(replicaSurface).sort(), [ 'createReplicaGraphqlTransport', 'createReplicaIndexMaintenanceRegistry', 'createReplicaIndexedDbPersistence', + 'createReplicaUuidV7', 'createWasmJsonPure', 'decideReplicaPaginationMaintenance', 'evaluateReplicaFilter', 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..307e5a6c7 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 }); }; @@ -1022,6 +1020,18 @@ export function createReplicaCommandRuntime< if (tracker.pending !== undefined) { settleTrackedProjection(tracker, pending); } + } else if ( + metadata.state === 'atomic' && + !prepared.revalidation.required && + !statusRequiresRevalidation + ) { + /* + * An exact terminal delta proves delivery but carries no + * canonical revision. Keep its accepted overlay until a later + * comparable authoritative result seals it, without racing + * sibling commands with a command-triggered query. + */ + settleTrackedProjection(tracker, pending); } else if ( metadata.state === 'atomic' || (metadata.state === 'succeeded' && diff --git a/js/src/replica/distributed-replica/impl-optimistic.ts b/js/src/replica/distributed-replica/impl-optimistic.ts index 8b37acc16..9ce9469eb 100644 --- a/js/src/replica/distributed-replica/impl-optimistic.ts +++ b/js/src/replica/distributed-replica/impl-optimistic.ts @@ -1,4 +1,5 @@ import type { + BaseCacheWriter, CacheEngine, OptimisticLayerReplacement } from '../../internal/cache-engine.js'; @@ -220,9 +221,18 @@ export function confirmOptimisticLayerOn( id: string, update: (writer: ReplicaBaseWriter) => T ): T { - const result = host.engine.confirmOptimisticLayer(id, (writer) => + return confirmOptimisticLayerWithCacheWriterOn(host, id, (writer) => update(baseWriter(writer)) ); +} + +/** Internal confirmation seam for protocol code that must atomically seal indexes. */ +export function confirmOptimisticLayerWithCacheWriterOn( + host: OptimisticHost, + id: string, + update: (writer: BaseCacheWriter) => T +): T { + const result = host.engine.confirmOptimisticLayer(id, update); host.retireDiagnosticLayer(id, 'retired', 'atomic'); host.optimisticReceipts.delete(id); host.syncDiagnostics(); diff --git a/js/src/replica/distributed-replica/impl.ts b/js/src/replica/distributed-replica/impl.ts index 9e8bf6328..f6371fee1 100644 --- a/js/src/replica/distributed-replica/impl.ts +++ b/js/src/replica/distributed-replica/impl.ts @@ -171,6 +171,7 @@ import { import { applyReceiptOnly as applyReceiptOnlyOn, confirmOptimisticLayerOn, + confirmOptimisticLayerWithCacheWriterOn, createOptimisticLayerOn, markOptimisticLayerAcceptedOn, planOptimisticReceipts as planOptimisticReceiptsOn, @@ -686,14 +687,61 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { pendingAnonymousRecordClocks, consumedAnonymousRecordClocks ); + /* + * The Atomic output is a complete authoritative row. Seal every active + * collection membership that the compiler plan can prove from that row in + * the same base transaction; otherwise the record exists by key while a + * warm @load index still omits it until refresh. + */ + const indexMutations = apply + ? this.#directProjectionIndexMutations( + commandId, + model, + recordKey, + fields + ) + : Object.freeze([]); + const indexRevision = + indexMutations.length === 0 + ? undefined + : this.#allocateIndexRevision(); - this.confirmOptimisticLayer(commandId, (writer) => { - if (!apply) return false; - return writer.writeRecord(model, identity, evidence.revision, { - incarnation: evidence.incarnation, - fields - }); - }); + confirmOptimisticLayerWithCacheWriterOn( + this.#optimisticHost(), + commandId, + (writer) => { + if (!apply) return false; + const wrote = writer.writeRecord({ + key: recordKey, + revision: evidence.revision, + incarnation: evidence.incarnation, + fields + }); + if (indexRevision !== undefined) { + for (const mutation of indexMutations) { + switch (mutation.kind) { + case 'write': + writer.writeIndex({ + ...mutation.write, + revision: indexRevision + }); + break; + case 'stale': + writer.markIndexStale( + mutation.key, + mutation.reason, + indexRevision + ); + break; + case 'delete': + writer.deleteIndex(mutation.key, indexRevision); + break; + } + } + } + return wrote; + } + ); for (const [key, clock] of pendingRecordClocks) { this.#recordClocks.set(key, clock); @@ -1990,6 +2038,30 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { return Object.freeze(mutations); } + #directProjectionIndexMutations( + commandId: string, + model: ReplicaModelArtifact, + recordKey: string, + fields: Readonly> + ): readonly DerivedIndexMutation[] { + const change: ReplicaIndexSemanticChange = Object.freeze({ + kind: 'upsert', + model: model.id, + key: recordKey, + fields + }); + const layer: OptimisticLayerView = Object.freeze({ + id: commandId, + sequence: Number.MAX_SAFE_INTEGER, + state: 'accepted', + context: Object.freeze({ + id: commandId, + changes: Object.freeze([change]) + }) + }); + return this.#deriveMaintainedIndexes(this.#engine.extract(), [layer]); + } + #operationProtocol( key: string, operation: string, 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..2e2645da3 100644 --- a/js/tests/replica-command-runtime.test.mjs +++ b/js/tests/replica-command-runtime.test.mjs @@ -1430,6 +1430,54 @@ for (const statusState of ['atomic', 'succeeded_pending_projection']) { }); } +test('terminal exact projection status settles without command-triggered revalidation', async () => { + const replica = new TestReplica(); + let pendingMetadata; + const runtime = createReplicaCommandRuntime( + replica, + { + dispatch(request) { + pendingMetadata = commandMetadata(request, { + actualTitle: 'accepted' + }); + return Promise.resolve( + envelope(request, { command: pendingMetadata }) + ); + }, + status(request) { + const terminalMetadata = Object.freeze({ + ...pendingMetadata, + state: 'atomic', + observations: Object.freeze( + pendingMetadata.expects.map((expectation) => + Object.freeze({ + ...expectation, + causationId: pendingMetadata.causationId + }) + ) + ) + }); + return Promise.resolve( + statusEnvelope(request, terminalMetadata) + ); + } + }, + { change: artifact() }, + { status: STATUS } + ); + const receipt = await runtime.commands.change( + { id: 'todo-1', title: 'preview' }, + { commandId: COMMAND_A } + ); + + assert.equal((await receipt.status()).state, 'atomic'); + assert.equal((await receipt.projected).state, 'atomic'); + assert.deepEqual(replica.revalidations, []); + assert.equal(replica.layer(COMMAND_A), 'accepted'); + assert.equal(replica.record('todo-1').fields.title, 'accepted'); + runtime.dispose(); +}); + test('invalid live progression cannot poison a later valid status transition', async () => { const replica = new TestReplica(); let request; @@ -1558,7 +1606,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 +1616,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 +1817,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/js/tests/replica-protocol.test.mjs b/js/tests/replica-protocol.test.mjs index 52cfbe1ef..11d3a6744 100644 --- a/js/tests/replica-protocol.test.mjs +++ b/js/tests/replica-protocol.test.mjs @@ -195,6 +195,24 @@ const TodosServerOnly = Object.freeze({ ]) }); +const TodosLocallyMaintainable = Object.freeze({ + ...TodosServerOnly, + id: 'query:todos-local', + protocol: Object.freeze({ + ...TodosServerOnly.protocol, + operation: 'query:todos-local' + }), + roots: Object.freeze([ + Object.freeze({ + ...TodosServerOnly.roots[0], + filter: Object.freeze({ + ...TodosServerOnly.roots[0].filter, + rowPolicy: Object.freeze({ kind: 'unrestricted' }) + }) + }) + ]) +}); + const GamesWithOwner = Object.freeze({ id: 'query:games-with-owner', document: 'query GamesWithOwner { games { id owner_id owner { id name } } }', @@ -794,6 +812,83 @@ test('Atomic direct projection does not hold later complete @load membership', ( ]); }); +test('Atomic direct projection commits locally provable collection membership', () => { + const replica = createDistributedReplica(); + write( + replica, + { + operation: TodosLocallyMaintainable.id, + position: '1', + rows: [{ id: 'todo-1', title: 'first' }] + }, + 'network', + TodosLocallyMaintainable + ); + // Render once so the operation's compiler plan owns collection maintenance. + replica.read(TodosLocallyMaintainable, {}); + replica.createOptimisticLayer('cmd-atomic-create', (writer) => { + // Generated partial previews fail closed when the record is not known yet. + writer.writeRecord(Todo, 'todo-2', { + fields: { id: 'todo-2', title: 'preview' }, + ifPresent: true + }); + }); + replica[replicaCommandDirectProjection]('cmd-atomic-create', { + model: Todo, + identity: 'todo-2', + evidence: { + model: Todo.id, + scopeToken: 'record:todo-2', + incarnation: '1', + revision: '2', + tombstone: false + }, + fields: { id: 'todo-2', title: 'canonical', __typename: Todo.id } + }); + + assert.deepEqual(replica.read(TodosLocallyMaintainable, {}).data.todos, [ + { id: 'todo-1', title: 'first' }, + { id: 'todo-2', title: 'canonical' } + ]); +}); + +test('Atomic direct projection keeps unprovable collection membership stale', () => { + const replica = createDistributedReplica(); + write( + replica, + { + operation: TodosServerOnly.id, + position: '1', + rows: [{ id: 'todo-1', title: 'first' }] + }, + 'network', + TodosServerOnly + ); + replica.read(TodosServerOnly, {}); + replica.createOptimisticLayer('cmd-atomic-server-only', (writer) => { + writer.writeRecord(Todo, 'todo-2', { + fields: { id: 'todo-2', title: 'preview' }, + ifPresent: true + }); + }); + replica[replicaCommandDirectProjection]('cmd-atomic-server-only', { + model: Todo, + identity: 'todo-2', + evidence: { + model: Todo.id, + scopeToken: 'record:todo-2', + incarnation: '1', + revision: '2', + tombstone: false + }, + fields: { id: 'todo-2', title: 'canonical', __typename: Todo.id } + }); + + const snapshot = replica.read(TodosServerOnly, {}); + assert.equal(snapshot.stale, true); + assert.deepEqual(snapshot.data.todos, [{ id: 'todo-1', title: 'first' }]); +}); + test('shared non-comparable membership follows request-start order across operations', async () => { const fetches = []; const replica = createDistributedReplica({ 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/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/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/celld-nats-profile/README.md b/tests/e2e-ui/celld-nats-profile/README.md index 529db2190..10e842d91 100644 --- a/tests/e2e-ui/celld-nats-profile/README.md +++ b/tests/e2e-ui/celld-nats-profile/README.md @@ -34,7 +34,7 @@ The playground UI against celld is the sibling example `tests/e2e-celld/` | Path | Where | |---|---| -| GraphQL wait-path mutations | `HttpCommandHost` → celld `POST /todo/{id}/todo.create` (`{ commandId, input }`) | +| GraphQL wait-path mutations | `CelldCommandHost` → celld `POST /todo/{id}/todo.create` (`{ commandId, input }` + internal identity headers) | | Fire-and-forget / events | NATS JetStream `publish` / `subscribe` | | Todo / Chat lists | SQL read models (projectors subscribe on NATS, **not** in cells) | | BlobGames by-id | `ReadStore::CellByKey` GET of the sealed row | 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) }); } 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/e2e/unauth.anon.spec.ts b/tests/e2e-ui/e2e/unauth.anon.spec.ts index 92c93cdb7..8031bb77f 100644 --- a/tests/e2e-ui/e2e/unauth.anon.spec.ts +++ b/tests/e2e-ui/e2e/unauth.anon.spec.ts @@ -25,6 +25,34 @@ test.describe('unauthenticated access', () => { }); }); + test('home soft-navigation installs the anonymous chat client', async ({ page }) => { + await page.goto('/'); + await page.waitForLoadState('networkidle'); + const continuityToken = `anonymous-chat-${Date.now()}`; + await page.evaluate((token) => { + Object.assign(globalThis, { __anonymousChatContinuityToken: token }); + }, continuityToken); + + await page + .getByLabel('main navigation') + .getByRole('link', { name: 'Chat', exact: true }) + .click(); + + await expect(page).toHaveURL(/\/chat(?:[/?#]|$)/); + await expect(page.getByRole('heading', { name: 'Lobby' })).toBeVisible({ + timeout: 20_000 + }); + expect( + await page.evaluate( + () => + (globalThis as typeof globalThis & { + __anonymousChatContinuityToken?: string; + }).__anonymousChatContinuityToken + ), + 'Chat navigation must preserve the current document' + ).toBe(continuityToken); + }); + test('home page is reachable without a session', async ({ page }) => { await page.goto('/'); await expect(page.getByRole('heading', { level: 1 }).first()).toBeVisible({ 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 data); - const guestBootstrap = untrack( - () => - !initialData.session?.user && - initialData.distributed !== undefined && - initialData.distributedAuthority !== undefined - ); + const guestAtMount = untrack(() => !initialData.session?.user); + const guestBootstrap = + guestAtMount && + initialData.distributed !== undefined && + initialData.distributedAuthority !== undefined; const pageData = createPageDataSessionSource(initialData); let appliedHydration: SveltekitReplicaHydration | undefined = guestBootstrap @@ -32,12 +31,16 @@ : undefined; let hydrationTimer: ReturnType | undefined; - const client = guestBootstrap + const client = guestAtMount ? provideDistributed({ session: pageData.session, browser, - hydration: initialData.distributed!, - authority: initialData.distributedAuthority! + ...(guestBootstrap + ? { + hydration: initialData.distributed!, + authority: initialData.distributedAuthority! + } + : {}) }) : null; diff --git a/tests/e2e-ui/ui/src/routes/todos/+page.svelte b/tests/e2e-ui/ui/src/routes/todos/+page.svelte index 87d292db9..5229ad196 100644 --- a/tests/e2e-ui/ui/src/routes/todos/+page.svelte +++ b/tests/e2e-ui/ui/src/routes/todos/+page.svelte @@ -5,6 +5,7 @@ * There is no app cache adapter or manual optimistic recipe here. The * compiler artifact tells the replica how command facts affect this query. */ + import { createReplicaUuidV7 } from '@hops-ops/distributed/replica'; import { Todos, useCommands } from '$distributed'; import { Button } from '$lib/components/shared/ui'; import { @@ -22,6 +23,7 @@ let title = $state(''); let actionError = $state(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}