Cache protocol manifests and recover retired-page live streams

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.

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.

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.

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 &&

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 =

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,14 +2222,16 @@ 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(
 		GamesWithOwnerLiveOperation,
 		{},
 		{ 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,

src/bus/runner/tests.rs

@@ -207,6 +207,54 @@ fn run_with<I: Send>(
 }
 
 // --- 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());

src/graphql/client_manifest/export.rs

@@ -6,6 +6,9 @@ pub struct DistributedClientSurfaceExport {
     identity: ClientSurfaceIdentity,
     surface: Arc<Surface>,
     execution: ClientExecutionLimits,
+    // All inputs are immutable and private. Clones share only compiled metadata,
+    // never request authority, preset values, tokens or read results.
+    manifest: Arc<std::sync::OnceLock<Result<DistributedClientManifest, ClientManifestError>>>,
 }
 
 /// 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<DistributedClientManifest, ClientManifestError> {
-        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
     }