diff --git a/docs/live-retired-owner-handoff.md b/docs/live-retired-owner-handoff.md new file mode 100644 index 000000000..f0c46f441 --- /dev/null +++ b/docs/live-retired-owner-handoff.md @@ -0,0 +1,21 @@ +# Snapshot stream handoff after owner disposal + +A snapshot live stream can begin while a previous route still owns a shared +relationship index. Its buffered frames must not acquire that graph merely +because the old route later disposes: their start fence predates disposal. +However, retaining that fence forever leaves the new route permanently pending. + +When an incoming snapshot is blocked by a retired owner's index, reopen that +contending operation at a fresh local start fence. Reject the triggering old +frame and all subsequent callbacks from its retired transport. Only a new +server-authorized initial frame may take over the graph. A still-active owner +continues to block; normal same-scope hydration must not restart retained layout +subscriptions. No application polling, forced refresh, or optimistic-state +suppression is part of this recovery. + +Runtime evidence: Forge's creation page received completed/ready rows, but the +prior NewRepository operation owned a shared topology relationship at revision +24, retired at 27, after the new subscription had already started. Entity clocks +advanced while the complete-empty root index remained fenced. The existing +pre-disposal protocol test covers safety; a fresh-receiver continuation covers +liveness after the same boundary. diff --git a/docs/protocol-manifest-reuse.md b/docs/protocol-manifest-reuse.md new file mode 100644 index 000000000..93a610319 --- /dev/null +++ b/docs/protocol-manifest-reuse.md @@ -0,0 +1,44 @@ +# Immutable protocol manifest reuse + +Runtime contract: a selected client surface export owns immutable service, +role/application selection, surface IR and execution limits. Its derived manifest +can be initialized once and shared across clones. Public callers still receive +independently mutable manifest copies. Errors are deterministic for that export +and may be retained; creating another export/engine creates a separate cache. + +The engine must retain the exact exports already validated during protocol +construction, including distinct role and application identities. Request seeds +may borrow their compiled manifests, but must still validate principal, asserted +roles, resolved preset values, authorization generation, visibility surface, +cache-scope HMAC and issuance time for each request. No session, token, result row, +authorization decision or request seed is cached. + +Observed baseline in Forge: authenticated document TTFB around 2.9–4.1 seconds, +simple authenticated GraphQL queries around 1.1 seconds. Native sampling identifies +ProtocolProjectionRequestSeed::new → export.manifest → projection manifest +lowering as repeated CPU work. Validation must cover cache/clone concurrency, +mutable-return independence, role/application/limits isolation and existing +protocol privacy/authority tests, followed by same-route runtime measurements. +No deadline, authentication, SSR or live-subscription behavior may be weakened. + +## Measured local validation + +Same retained Forge stack, same authenticated user and paths, September 21: + +| Warm document (DOM ready) | Before | After | +| --- | ---: | ---: | +| Dashboard | 3140 ms | 116 ms | +| Personal repository, including default-ref redirect | 7445 ms | 291 ms | +| Organization People | 3350 ms | 192 ms | +| ChangeSets | 4123 ms | 97 ms | + +Three direct authenticated GraphQL queries fell from 1143/1135/1072 ms to +22/15/14 ms. These are local observations, not performance assertions or an SLA. +The first repository document after the development server restart still took +18.7 seconds; the warm comparison does not hide that cold-development cost. + +Framework library: 1052 passed, 3 explicit ignored. GraphQL/identity/causal HTTP +integration targets: 43 passed. Added tests exercise concurrent first use, +independent mutable returns, role/application/limit separation, rebuilt-engine +isolation, and actual repeated protocol accumulators retaining/releasing shared +metadata without retaining request authority. Existing auth/privacy checks remain. diff --git a/docs/reloading-event-delivery.md b/docs/reloading-event-delivery.md new file mode 100644 index 000000000..0bd61ce9b --- /dev/null +++ b/docs/reloading-event-delivery.md @@ -0,0 +1,13 @@ +# Event delivery during application reload + +A supervisor generation gate is temporary infrastructure state, not a domain +rejection. Both direct command admission and bus dispatch return typed +`ApplicationReloading` while closed. Bus conversion classifies it as retryable +and retains/stops the receive loop: NAK the exact delivery, then surface the +error to the host's bounded restart policy. It must never Ack, Term or apply +ordinary permanent-failure policy. After activation the same delivery can run. + +Business validation/authorization failures retain their existing permanent +classification and configured settlement policy. No gate bypass or implicit +success is introduced. This does not change general infrastructure NAK delay; +hosts/brokers still own retry timing for other transport outages. diff --git a/js/src/replica/distributed-replica/impl.ts b/js/src/replica/distributed-replica/impl.ts index a18077834..60c6ac38c 100644 --- a/js/src/replica/distributed-replica/impl.ts +++ b/js/src/replica/distributed-replica/impl.ts @@ -1614,6 +1614,12 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { if (source !== 'live' && sourceSwitched) { this.#restartLive(key); } + if (source === 'live' && sharedDisposition.restartAfterRetirement) { + // This receiver began before a shared owner retired. Its buffered + // frame stays fenced, but a fresh receiver starts after that boundary + // and can obtain an authoritative replacement without polling. + this.#restartLive(key); + } this.#trustedPresets = nextTrustedPresets; this.#protocolGeneration = nextProtocolGeneration; this.#resumeLiveWatches(); @@ -2327,6 +2333,7 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { let lower = false; let higher = false; let incomparable = false; + let restartAfterRetirement = false; let equalRevision: string | undefined; let latestOwnerRevision: string | undefined; for (const [key, group] of this.#operationProtocols) { @@ -2348,16 +2355,18 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { !snapshot.indexesComparable && state === group.live && state.retiredAtRevision !== undefined && - liveStart !== undefined && - compareCanonicalDecimalStrings( - liveStart, - state.retiredAtRevision - ) > 0 + liveStart !== undefined ) { // A disposed stream is no longer an owner. Its boundary is // still retained so a stream that started before disposal // cannot win merely because the old transport later closed. - continue; + if (compareCanonicalDecimalStrings(liveStart, state.retiredAtRevision) > 0) { + continue; + } + // Reopen at most once per observed retirement boundary: the + // replacement's allocated start is strictly newer. Active + // siblings never trigger this recovery and remain fenced. + restartAfterRetirement = true; } latestOwnerRevision = latestOwnerRevision === undefined || @@ -2429,7 +2438,11 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { * explicit synchronous ingress retains its caller-defined order. */ if (requestRevision === undefined && source === 'live') { - return { compared: true, disposition: 'lower' }; + return { + compared: true, + disposition: 'lower', + ...(restartAfterRetirement ? { restartAfterRetirement: true } : {}) + }; } if (requestRevision === undefined) return { compared: false }; return latestOwnerRevision !== undefined && diff --git a/js/src/replica/distributed-replica/types.ts b/js/src/replica/distributed-replica/types.ts index 4ae8faade..ebbcbefea 100644 --- a/js/src/replica/distributed-replica/types.ts +++ b/js/src/replica/distributed-replica/types.ts @@ -169,6 +169,7 @@ export type SharedIndexDisposition = { readonly compared: boolean; readonly disposition?: 'equal' | 'higher' | 'lower'; readonly indexRevision?: string; + readonly restartAfterRetirement?: boolean; }; export type CapturedReplicaOptimisticOperation = diff --git a/js/tests/replica-protocol.test.mjs b/js/tests/replica-protocol.test.mjs index 42095e281..7b4c374bd 100644 --- a/js/tests/replica-protocol.test.mjs +++ b/js/tests/replica-protocol.test.mjs @@ -2022,13 +2022,14 @@ test('snapshot live takes over a nested graph from a disposed page subscription' test('snapshot live keeps a stream started before disposal behind the retired owner', () => { const observers = []; + const closed = []; const previousPage = { ...FeaturedGamesWithOwner, live: { id: 'live:featured-owner-before-disposal', document: 'subscription FeaturedOwnerBeforeDisposal { featuredGames { id owner { id name } } }' } }; const replica = createDistributedReplica({ transport: { fetch() { throw new Error('complete snapshot must not force HTTP fallback'); }, - subscribe(_request, observer) { observers.push(observer); return () => {}; } + subscribe(_request, observer) { observers.push(observer); return () => { closed.push(observer); }; } } }); const oldFrame = (ownerName) => gamesFrame({ artifact: previousPage, responseKey: 'featuredGames', operation: previousPage.live.id, @@ -2052,6 +2053,9 @@ test('snapshot live keeps a stream started before disposal behind the retired ow live: { mode: 'snapshot', reset: true, cursors: [] } })); assert.deepEqual(current.get().data.games, []); + const layout = replica.watch(Todos, {}, { live: true }); + observers[2].next(wireFrame({ operation: Todos.live.id, indexesComparable: false, + live: { mode: 'snapshot', reset: true }, rows: [{ id: 'retained', title: 'layout' }] })); oldWatch.destroy(); observers[1].next(gamesFrame({ artifact: GamesWithOwnerLiveOperation, responseKey: 'games', @@ -2060,7 +2064,30 @@ test('snapshot live keeps a stream started before disposal behind the retired ow live: { mode: 'snapshot', reset: true, cursors: [] } })); assert.deepEqual(current.get().data.games, []); + assert.equal(observers.length, 4, 'retired ownership must reopen only the contender after the disposal fence'); + assert.deepEqual(closed, [observers[0], observers[1]]); + assert.equal(layout.get().live, 'active'); + observers[1].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '5', + ownerId: 'user-1', ownerName: 'queued old receiver', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.deepEqual(current.get().data.games, [], 'old receiver stays fenced after reopening'); + observers[3].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '6', + ownerId: 'user-1', ownerName: 'fresh authoritative receiver', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.equal(current.get().data.games[0].owner.name, 'fresh authoritative receiver'); + assert.equal(observers.length, 4, 'the same retirement boundary cannot reopen repeatedly'); + observers[2].next(wireFrame({ operation: Todos.live.id, indexesComparable: false, + live: { mode: 'snapshot', reset: true }, revision: '2', rows: [{ id: 'retained', title: 'continued layout' }] })); + assert.equal(layout.get().data.todos[0].title, 'continued layout'); current.destroy(); + layout.destroy(); + assert.deepEqual(closed, [observers[0], observers[1], observers[3], observers[2]]); }); test('reopening a retired live owner restores its ownership fence', () => { @@ -2195,6 +2222,7 @@ test('two watches retire shared live ownership only after final disposal', () => live: { mode: 'snapshot', reset: true, cursors: [] } })); assert.deepEqual(startedBeforeFinalRelease.get().data.games, []); + assert.equal(observers.length, 3, 'the final release enables one fresh contender receiver'); startedBeforeFinalRelease.destroy(); const afterFinalRelease = replica.watch( @@ -2202,7 +2230,8 @@ test('two watches retire shared live ownership only after final disposal', () => {}, { live: true } ); - observers[2].next(gamesFrame({ + assert.equal(observers.length, 4, 'reopening after disposal creates another receiver'); + observers[3].next(gamesFrame({ artifact: GamesWithOwnerLiveOperation, responseKey: 'games', operation: GamesWithOwnerLiveOperation.live.id, position: '5', ownerId: 'user-1', ownerName: 'after final release', indexesComparable: false, diff --git a/src/bus/runner/tests.rs b/src/bus/runner/tests.rs index a31074c26..5319d8c0b 100644 --- a/src/bus/runner/tests.rs +++ b/src/bus/runner/tests.rs @@ -207,6 +207,54 @@ fn run_with( } // --- tests -------------------------------------------------------------- +#[test] +fn application_reload_retains_exact_delivery_then_succeeds_after_activation() { + use std::sync::atomic::{AtomicBool, Ordering}; + let recorder = Recorder::new(); + let open = Arc::new(AtomicBool::new(false)); + let gate = open.clone(); + let effects = recorder.clone(); + let router = Arc::new( + Handlers::new().on_event("reload", move |message: &Message| { + let admitted = gate.load(Ordering::SeqCst); + let effects = effects.clone(); + let id = message.id().unwrap().to_owned(); + async move { + if !admitted { + return Err(crate::microsvc::HandlerError::ApplicationReloading.into()); + } + effects.push(Event::Handled(id)); + Ok(()) + } + }), + ); + let message = event_message("reload", Some("retained-event")); + let source = || FakeSource { + queue: vec![message.clone()].into_iter().collect(), + recorder: recorder.clone(), + settle_ok: true, + recv_error: false, + decode_error: false, + }; + let error = block_on(run_source( + router.clone(), + source(), + RunOptions::idempotent(), + )) + .unwrap_err(); + assert!(error.is_retryable()); + assert!(error.should_retain_and_stop()); + assert!( + matches!(recorder.events().as_slice(), [Event::Nack(reason)] if reason.contains("reloading")) + ); + open.store(true, Ordering::SeqCst); + block_on(run_source(router, source(), RunOptions::idempotent())).unwrap(); + assert_eq!( + &recorder.events()[1..], + &[Event::Handled("retained-event".into()), Event::Ack] + ); +} + #[test] fn success_dispatches_then_acks_in_order() { let result = run(vec![event_message("ok", None)], RunOptions::idempotent()); diff --git a/src/graphql/client_manifest/export.rs b/src/graphql/client_manifest/export.rs index a7be2ffb9..141c2fed4 100644 --- a/src/graphql/client_manifest/export.rs +++ b/src/graphql/client_manifest/export.rs @@ -6,6 +6,9 @@ pub struct DistributedClientSurfaceExport { identity: ClientSurfaceIdentity, surface: Arc, execution: ClientExecutionLimits, + // All inputs are immutable and private. Clones share only compiled metadata, + // never request authority, preset values, tokens or read results. + manifest: Arc>>, } /// Do not transitively format the selected Surface: it retains a private full @@ -32,6 +35,7 @@ impl DistributedClientSurfaceExport { identity, surface: surface.into(), execution, + manifest: Arc::new(std::sync::OnceLock::new()), } } @@ -116,18 +120,32 @@ impl DistributedClientSurfaceExport { } pub fn manifest(&self) -> Result { - client_manifest_from_surface_with_execution( - &self.service_id, - self.identity.clone(), - &self.surface, - self.execution.clone(), - ) + self.manifest_ref().cloned() + } + + pub(crate) fn manifest_ref(&self) -> Result<&DistributedClientManifest, ClientManifestError> { + self.manifest + .get_or_init(|| { + client_manifest_from_surface_with_execution( + &self.service_id, + self.identity.clone(), + &self.surface, + self.execution.clone(), + ) + }) + .as_ref() + .map_err(Clone::clone) } pub fn service_id(&self) -> &str { &self.service_id } + #[cfg(test)] + pub(crate) fn manifest_cache_owners(&self) -> usize { + Arc::strong_count(&self.manifest) + } + pub fn identity(&self) -> &ClientSurfaceIdentity { &self.identity } diff --git a/src/graphql/client_manifest/tests.rs b/src/graphql/client_manifest/tests.rs index 4f2e876d6..a1b338a1f 100644 --- a/src/graphql/client_manifest/tests.rs +++ b/src/graphql/client_manifest/tests.rs @@ -754,6 +754,75 @@ fn manifest_for_all_models( .expect("client manifest") } +#[test] +fn selected_export_cache_is_shared_concurrently_but_returned_manifests_are_independent() { + let full = full_surface(); + let selected = surface_for_role(&full, "user", &grants()["user"]).unwrap(); + let fresh = client_manifest_from_surface( + "todos-service", + ClientSurfaceIdentity::role("user"), + &selected, + ) + .unwrap(); + let export = DistributedClientSurfaceExport::from_selected("todos-service", selected).unwrap(); + let barrier = Arc::new(std::sync::Barrier::new(8)); + let threads: Vec<_> = (0..8) + .map(|_| { + let export = export.clone(); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + barrier.wait(); + let manifest = export.manifest_ref().unwrap(); + ( + manifest as *const DistributedClientManifest as usize, + manifest.clone(), + ) + }) + }) + .collect(); + let expected_address = + export.manifest_ref().unwrap() as *const DistributedClientManifest as usize; + for thread in threads { + let (address, manifest) = thread.join().unwrap(); + assert_eq!(address, expected_address); + assert_eq!(manifest, fresh); + } + let mut detached = export.manifest().unwrap(); + detached.models.clear(); + detached.projection_programs.clear(); + detached.schema_fingerprint.clear(); + assert_eq!(export.manifest_ref().unwrap(), &fresh); + assert_eq!(export.clone().manifest().unwrap(), fresh); +} + +#[test] +fn selected_export_caches_do_not_alias_role_application_or_execution_limits() { + let full = full_surface(); + let grants = grants(); + let user = surface_for_role(&full, "user", &grants["user"]).unwrap(); + let admin = surface_for_role(&full, "admin", &grants["admin"]).unwrap(); + let application = + surface_for_application(&full, "web", &["user".into()], &["user".into()], &grants).unwrap(); + let user_export = + DistributedClientSurfaceExport::from_selected("todos-service", user.clone()).unwrap(); + let admin_export = + DistributedClientSurfaceExport::from_selected("todos-service", admin).unwrap(); + let app_export = + DistributedClientSurfaceExport::from_selected("todos-service", application).unwrap(); + let mut limits = ClientExecutionLimits::default(); + limits.max_bool_width += 1; + let limited = + DistributedClientSurfaceExport::from_selected_with_execution("todos-service", user, limits) + .unwrap(); + let baseline = user_export.manifest_ref().unwrap(); + for other in [&admin_export, &app_export, &limited] { + let manifest = other.manifest_ref().unwrap(); + assert!(!std::ptr::eq(baseline, manifest)); + assert_ne!(baseline.schema_fingerprint, manifest.schema_fingerprint); + } + assert_ne!(user_export.identity(), app_export.identity()); +} + #[test] fn role_manifest_is_deterministic_and_hides_denied_identity_and_commands() { let full = full_surface(); diff --git a/src/graphql/engine/builder.rs b/src/graphql/engine/builder.rs index 4ccc46b86..aeca975c9 100644 --- a/src/graphql/engine/builder.rs +++ b/src/graphql/engine/builder.rs @@ -836,7 +836,7 @@ impl GraphqlEngineBuilder { .expect("protocol configuration validated a service ID"); let (authorization_fingerprint, claim_keys) = role_authorization_info(role, &self.permissions)?; - let manifest = DistributedClientSurfaceExport::from_selected_with_execution( + let export = DistributedClientSurfaceExport::from_selected_with_execution( service_id, Arc::clone(&role_surface), ClientExecutionLimits::from_runtime( @@ -847,17 +847,22 @@ impl GraphqlEngineBuilder { ) .map_err(|error| GraphqlBuildError(error.to_string()))?, ) - .and_then(|export| export.manifest()) .map_err(|error| { GraphqlBuildError(format!( "failed to derive GraphQL protocol surface for role `{role}`: {error}" )) })?; + let manifest = export.manifest().map_err(|error| { + GraphqlBuildError(format!( + "failed to derive GraphQL protocol surface for role `{role}`: {error}" + )) + })?; let trusted_presets = protocol_trusted_presets(&manifest)?; protocol_roles.insert( role.clone(), ProtocolRoleInfo { surface: ProtocolSurfaceInfo { + export, schema_fingerprint: manifest.schema_fingerprint, protocol_fingerprint: manifest.protocol_fingerprint, trusted_presets, @@ -915,7 +920,7 @@ impl GraphqlEngineBuilder { .service_id .as_deref() .expect("protocol configuration validated a service ID"); - let manifest = DistributedClientSurfaceExport::from_selected_with_execution( + let export = DistributedClientSurfaceExport::from_selected_with_execution( service_id, Arc::clone(&application_surface), ClientExecutionLimits::from_runtime( @@ -926,7 +931,10 @@ impl GraphqlEngineBuilder { ) .map_err(|error| GraphqlBuildError(error.to_string()))?, ) - .and_then(|export| export.manifest()) + .map_err(|error| GraphqlBuildError(format!( + "failed to derive GraphQL protocol surface for application `{application}`: {error}" + )))?; + let manifest = export.manifest() .map_err(|error| { GraphqlBuildError(format!( "failed to derive GraphQL protocol surface for application `{application}`: {error}" @@ -990,6 +998,7 @@ impl GraphqlEngineBuilder { schema_roles: registration.schema_roles.clone(), privilege_key, surface: ProtocolSurfaceInfo { + export, schema_fingerprint: manifest.schema_fingerprint, protocol_fingerprint: manifest.protocol_fingerprint, trusted_presets, diff --git a/src/graphql/engine/core.rs b/src/graphql/engine/core.rs index b453f60ad..f90815d07 100644 --- a/src/graphql/engine/core.rs +++ b/src/graphql/engine/core.rs @@ -170,6 +170,7 @@ pub(crate) struct RoleModelPerm { #[derive(Clone)] pub(crate) struct ProtocolSurfaceInfo { + pub(crate) export: DistributedClientSurfaceExport, pub(crate) schema_fingerprint: String, pub(crate) protocol_fingerprint: String, pub(crate) trusted_presets: Vec, diff --git a/src/graphql/engine/request.rs b/src/graphql/engine/request.rs index e8d3c2e77..bf027f643 100644 --- a/src/graphql/engine/request.rs +++ b/src/graphql/engine/request.rs @@ -298,26 +298,9 @@ impl GraphqlEngine { .get(&authority.privilege_role) .cloned() .ok_or(())?; - let selected_surface = match &surface_identity { - ClientSurfaceIdentity::Role { name } => self.inner.role_surfaces.get(name), - ClientSurfaceIdentity::Application { name, .. } => { - self.inner.application_surfaces.get(name) - } - } - .cloned() - .ok_or(())?; - let export = DistributedClientSurfaceExport::from_selected_with_execution( - &runtime.service_id, - selected_surface, - ClientExecutionLimits::from_runtime( - self.inner.max_depth, - self.inner.max_complexity, - self.inner.max_bool_width, - self.inner.max_in_list, - ) - .map_err(|_| ())?, - ) - .map_err(|_| ())?; + // Selected by the verified authority above; this exact immutable + // export was validated when the engine's protocol surface was built. + let export = surface_info.export.clone(); let issued_at_unix_ms = crate::time::now() .duration_since(std::time::UNIX_EPOCH) .map_err(|_| ())? diff --git a/src/graphql/engine/tests.rs b/src/graphql/engine/tests.rs index 8576815a4..151512583 100644 --- a/src/graphql/engine/tests.rs +++ b/src/graphql/engine/tests.rs @@ -142,6 +142,70 @@ mod client_surface_parity_tests { .unwrap() } + #[cfg(feature = "sqlite")] + #[tokio::test] + async fn protocol_surface_export_cache_is_engine_scoped_and_shared_by_engine_clones() { + let engine = Arc::new(protocol_engine("first-engine")); + let cloned = engine.clone(); + let another = protocol_engine("second-engine"); + let export = &engine.inner.protocol.as_ref().unwrap().roles["user"] + .surface + .export; + let cloned_export = &cloned.inner.protocol.as_ref().unwrap().roles["user"] + .surface + .export; + let other_export = &another.inner.protocol.as_ref().unwrap().roles["user"] + .surface + .export; + assert!(std::ptr::eq( + export.manifest_ref().unwrap(), + cloned_export.manifest_ref().unwrap() + )); + assert!(!std::ptr::eq( + export.manifest_ref().unwrap(), + other_export.manifest_ref().unwrap() + )); + let fresh = engine + .client_surface_for_role("user") + .unwrap() + .manifest() + .unwrap(); + assert_eq!(export.manifest_ref().unwrap(), &fresh); + assert_eq!(other_export.manifest_ref().unwrap(), &fresh); + } + + #[cfg(feature = "sqlite")] + #[tokio::test] + async fn repeated_protocol_accumulators_reuse_metadata_but_release_request_authority() { + use crate::graphql::identity::VerifiedPrincipal; + let engine = protocol_engine("request-cache-test"); + let export = &engine.inner.protocol.as_ref().unwrap().roles["user"] + .surface + .export; + let owners = export.manifest_cache_owners(); + let mut session = Session::new(); + session.set("x-roles", "user"); + let request = Request::new("{ __typename }").data(VerifiedPrincipal::test_oidc( + "https://issuer.example", + "principal-a", + &["orders-service"], + )); + let authority = resolve_execution_authority(&engine.inner, &session, &request).unwrap(); + let first = engine + .protocol_accumulator(&authority, &session, &request) + .unwrap_or_else(|_| panic!("first request authority")) + .unwrap(); + assert_eq!(export.manifest_cache_owners(), owners + 1); + let second = engine + .protocol_accumulator(&authority, &session, &request) + .unwrap_or_else(|_| panic!("second request authority")) + .unwrap(); + assert_eq!(export.manifest_cache_owners(), owners + 2); + drop(first); + drop(second); + assert_eq!(export.manifest_cache_owners(), owners); + } + #[cfg(feature = "sqlite")] fn policy_protocol_engine(namespace: &str, claim_key: &str) -> GraphqlEngine { let pool = sqlx::sqlite::SqlitePoolOptions::new() diff --git a/src/graphql/projection_delta/runtime.rs b/src/graphql/projection_delta/runtime.rs index 8ab93c6bd..4d1f68eb3 100644 --- a/src/graphql/projection_delta/runtime.rs +++ b/src/graphql/projection_delta/runtime.rs @@ -105,9 +105,9 @@ impl ProtocolProjectionRequestSeed { return Err(ProjectionRuntimeAuthorityError::InvalidAuthority); } let manifest = export - .manifest() + .manifest_ref() .map_err(|_| ProjectionRuntimeAuthorityError::InvalidAuthority)?; - let expected = crate::graphql::client_manifest::trusted_preset_descriptors(&manifest) + let expected = crate::graphql::client_manifest::trusted_preset_descriptors(manifest) .map_err(|_| ProjectionRuntimeAuthorityError::InvalidAuthority)?; let mut preset_names = BTreeSet::new(); if trusted_presets.len() != expected.len() diff --git a/src/microsvc/error.rs b/src/microsvc/error.rs index 437eb8e40..e52a2bb23 100644 --- a/src/microsvc/error.rs +++ b/src/microsvc/error.rs @@ -14,6 +14,9 @@ use crate::{repository::RepositoryError, EventRecordError}; #[derive(Debug)] #[non_exhaustive] pub enum HandlerError { + /// Supervisor generation admission is temporarily closed. Retain the exact + /// delivery and stop the receive loop until the host retries after reload. + ApplicationReloading, /// No handler registered for this command name. UnknownCommand(String), /// Payload decode / deserialization failed. @@ -51,6 +54,7 @@ pub enum HandlerError { impl fmt::Display for HandlerError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { + HandlerError::ApplicationReloading => f.write_str("application generation is reloading"), HandlerError::UnknownCommand(name) => write!(f, "unknown command: {}", name), HandlerError::DecodeFailed(msg) => write!(f, "decode failed: {}", msg), HandlerError::Rejected(msg) => write!(f, "rejected: {}", msg), @@ -142,6 +146,7 @@ impl HandlerError { /// Map this error to an HTTP-style status code. pub fn status_code(&self) -> u16 { match self { + HandlerError::ApplicationReloading => 503, HandlerError::UnknownCommand(_) => 404, HandlerError::DecodeFailed(_) => 400, HandlerError::Rejected(_) => 422, @@ -201,7 +206,8 @@ impl HandlerError { TransportErrorKind::Permanent } } - HandlerError::ProjectionRepairPending { .. } + HandlerError::ApplicationReloading + | HandlerError::ProjectionRepairPending { .. } | HandlerError::NotFound(_) | HandlerError::Other(_) => TransportErrorKind::Retryable, HandlerError::ProjectionTerminalRecorded { .. } @@ -228,7 +234,8 @@ impl From for TransportError { let kind = error.transport_error_kind(); let retain_and_stop = matches!( error, - HandlerError::ProjectionTerminalRecorded { .. } + HandlerError::ApplicationReloading + | HandlerError::ProjectionTerminalRecorded { .. } | HandlerError::ProjectionDeliveryHalted { .. } ); let transport = TransportError::new(kind, error.to_string()).with_source(error); @@ -244,6 +251,16 @@ impl From for TransportError { mod tests { use super::*; + #[test] + fn application_reloading_retains_delivery_without_reclassifying_business_errors() { + let reload = TransportError::from(HandlerError::ApplicationReloading); + assert!(reload.is_retryable()); + assert!(reload.should_retain_and_stop()); + let rejection = TransportError::from(HandlerError::Rejected("invalid slug".into())); + assert!(rejection.is_permanent()); + assert!(!rejection.should_retain_and_stop()); + } + #[test] fn transient_handler_errors_are_retryable() { for error in [ diff --git a/src/microsvc/http.rs b/src/microsvc/http.rs index e46fa56a2..b9835faa0 100644 --- a/src/microsvc/http.rs +++ b/src/microsvc/http.rs @@ -197,6 +197,7 @@ async fn command_handler( fn status_for_error(error: &HandlerError) -> StatusCode { match error { + HandlerError::ApplicationReloading => StatusCode::SERVICE_UNAVAILABLE, HandlerError::UnknownCommand(_) | HandlerError::NotFound(_) => StatusCode::NOT_FOUND, HandlerError::DecodeFailed(_) | HandlerError::GuardRejected(_) => StatusCode::BAD_REQUEST, HandlerError::Rejected(_) => StatusCode::UNPROCESSABLE_ENTITY, diff --git a/src/microsvc/service/runtime.rs b/src/microsvc/service/runtime.rs index 688b5a46d..67afbf02b 100644 --- a/src/microsvc/service/runtime.rs +++ b/src/microsvc/service/runtime.rs @@ -40,9 +40,7 @@ fn ensure_lifecycle_mutations_open() -> Result<(), HandlerError> { if crate::microsvc::lifecycle_mutations_open() { Ok(()) } else { - Err(HandlerError::Rejected( - "application generation is reloading".into(), - )) + Err(HandlerError::ApplicationReloading) } } @@ -1064,11 +1062,7 @@ impl Service { message: &Message, ordered: Option<&OrderedDelivery>, ) -> Result { - if !crate::microsvc::lifecycle_mutations_open() { - return Err(HandlerError::Rejected( - "application generation is reloading".into(), - )); - } + ensure_lifecycle_mutations_open()?; if !self.handles_message(message.kind, &message.name) { return Err(HandlerError::UnknownCommand(message.name.clone())); } diff --git a/src/telemetry.rs b/src/telemetry.rs index e538ce152..2a318b932 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -162,7 +162,7 @@ pub(crate) fn handler_error_status(error: &HandlerError) -> &'static str { | HandlerError::ProjectionTerminalRecorded { .. } | HandlerError::ProjectionDeliveryHalted { .. } => dispatch_status::REPOSITORY_ERROR, HandlerError::GuardRejected(_) => dispatch_status::GUARD_REJECTED, - HandlerError::Other(_) => dispatch_status::OTHER_ERROR, + HandlerError::ApplicationReloading | HandlerError::Other(_) => dispatch_status::OTHER_ERROR, } }