diff --git a/src/agent/hosted/executor-session.test.ts b/src/agent/hosted/executor-session.test.ts index 853c461276..146912de74 100644 --- a/src/agent/hosted/executor-session.test.ts +++ b/src/agent/hosted/executor-session.test.ts @@ -639,7 +639,10 @@ describe("hosted executor session", () => { } it("observes preparation without attaching until the allocator supplies a ready endpoint", async () => { - const f = fixture(); + const attachments: Parameters< + NonNullable + >[0][] = []; + const f = fixture({ observeAttachment: (receipt) => void attachments.push(receipt) }); const endpoint = f.returned.endpoint!; f.returned.phase = "preparing"; delete f.returned.endpoint; @@ -650,10 +653,20 @@ describe("hosted executor session", () => { f.time.advanceBy(100); await tick(); assertEquals(f.calls, ["allocate", "observe"]); + assertEquals(attachments, []); f.returned.phase = "ready"; f.returned.endpoint = endpoint; f.time.advanceBy(100); await session.ready; + assertEquals(attachments, [{ + allocationId: binding.allocationId, + generation: binding.generation, + invocationId: binding.invocationId, + podUid: endpoint.podUid, + nodeName: endpoint.nodeName, + declaredImage: image, + }]); + assert(Object.isFrozen(attachments[0])); assertEquals(f.calls.filter((call) => call === "allocate").length, 1); assertEquals(f.calls.filter((call) => call === "connect").length, 1); await session.close(); @@ -758,7 +771,12 @@ describe("hosted executor session", () => { }); it("does not repeat a failed authenticated attachment or expose its error details", async () => { - const f = fixture(); + let attachments = 0; + const f = fixture({ + observeAttachment: () => { + attachments++; + }, + }); let attempts = 0; f.options.connectTransport = () => { attempts++; @@ -770,6 +788,7 @@ describe("hosted executor session", () => { f.time.advanceBy(1000); await tick(); assertEquals(attempts, 1); + assertEquals(attachments, 0); assertEquals(f.calls.includes("operations"), false); assertEquals(f.allocatedKey, new Uint8Array(32)); assertEquals(f.time.pendingWaitCount, 0); diff --git a/src/agent/hosted/executor-session.ts b/src/agent/hosted/executor-session.ts index 3f85588f25..5ea7b0ab83 100644 --- a/src/agent/hosted/executor-session.ts +++ b/src/agent/hosted/executor-session.ts @@ -59,6 +59,17 @@ export interface HostedExecutorSessionOptions { allocator: HostedExecutorAllocatorClient; /** Authenticate one connection. Snapshot the allocation key before awaiting. */ connectTransport(options: ConnectExecutorTransportOptions): Promise; + /** Trusted synchronous receipt sink, invoked after channel authentication and before readiness. */ + observeAttachment?( + receipt: Readonly<{ + allocationId: string; + generation: number; + invocationId: string; + podUid: string; + nodeName: string; + declaredImage: string; + }>, + ): void; /** Factory must release its own partial resources if construction throws. */ createOperations(binding: Readonly, signal: AbortSignal): { operations: ReadonlyMap; @@ -140,6 +151,7 @@ class Session implements HostedExecutorSession { readonly #expectedImage: string; readonly #allocator: HostedExecutorAllocatorClient; readonly #connect: HostedExecutorSessionOptions["connectTransport"]; + readonly #observeAttachment: HostedExecutorSessionOptions["observeAttachment"]; readonly #createOperations: HostedExecutorSessionOptions["createOperations"]; readonly #clock: HostedExecutorSessionClock; readonly #pollInterval: number; @@ -212,6 +224,7 @@ class Session implements HostedExecutorSession { release: options.allocator.release.bind(options.allocator), }; this.#connect = options.connectTransport; + this.#observeAttachment = options.observeAttachment; this.#createOperations = options.createOperations; this.#clock = options.clock ?? createHostedExecutorSessionClock(Date.now()); this.#pollInterval = positiveLimit(options.pollIntervalMs ?? 250, 30_000); @@ -518,6 +531,14 @@ class Session implements HostedExecutorSession { void this.#channel.closed.then(() => this.#stop("channel-closed")); await awaitAbortable(this.#channel.ready, this.signal); this.#assertActive(); + const endpoint = this.#endpoint!; + this.#observeAttachment?.(Object.freeze({ + ...channelBinding, + podUid: endpoint.podUid, + nodeName: endpoint.nodeName, + declaredImage: endpoint.image, + })); + this.#assertActive(); this.#cancelTimer("attachment"); this.#isReady = true; this.#ready.resolve(this.#channel); diff --git a/src/server/isolated-http/generation-executor.ts b/src/server/isolated-http/generation-executor.ts index c49e61ba06..43235ea439 100644 --- a/src/server/isolated-http/generation-executor.ts +++ b/src/server/isolated-http/generation-executor.ts @@ -3,7 +3,10 @@ import { getExecutorHttpInstallSchema, parseExecutorInstallation, } from "#veryfront/agent/hosted/executor-runtime-install-schema.ts"; -import type { HostedExecutorSession } from "#veryfront/agent/hosted/executor-session.ts"; +import type { + HostedExecutorSession, + HostedExecutorSessionOptions, +} from "#veryfront/agent/hosted/executor-session.ts"; import type { HostedExecutorSessionPool } from "#veryfront/agent/hosted/executor-session-pool.ts"; import { getHostedExecutorAllocationRequestSchema, @@ -13,6 +16,10 @@ import { sameHostedExecutorOwner, } from "#veryfront/agent/hosted/executor-session-schema.ts"; import { verifyHostedRuntimeSourceBinding } from "#veryfront/agent/hosted/runtime-source-binding.ts"; +import { + formatTraceparent, + parseTraceparent, +} from "#veryfront/observability/tracing/traceparent.ts"; import { normalizeDynamicDenyHeaders } from "#veryfront/security/http/application-request.ts"; import { completeOnResponseBodyConsumption } from "#veryfront/platform/compat/http/response-lifecycle.ts"; import type { RenderGenerationExecutor } from "#veryfront/rendering/render-generation.ts"; @@ -112,6 +119,17 @@ export interface HostedHttpGenerationExecutorOptions { sessions: Pick; /** Look up the broker's data for an admitted request. */ context(request: Request): HostedHttpGenerationRequest | undefined; + /** Trusted synchronous observer of a request about to use an authenticated attachment. */ + observeDispatch?( + receipt: Readonly<{ + attachment: Parameters>[0]; + brokerInstanceId: string; + projectId: string; + environmentId: string; + releaseId: string; + traceparent?: string; + }>, + ): void; /** Stop admitting requests after this long without one in flight. */ idleTimeoutMs: number; /** @@ -206,6 +224,10 @@ export function createHostedHttpGenerationExecutor( let released = false; let stopping: Promise | undefined; let observation: Promise | undefined; + const observeDispatch = options.observeDispatch; + let attachment: + | Parameters>[0] + | undefined; function retire(): void { if (retired) return; @@ -224,6 +246,12 @@ export function createHostedHttpGenerationExecutor( } const started = sessions.start({ ...input.session, + ...(observeDispatch === undefined ? {} : { + observeAttachment(receipt) { + attachment = receipt; + input.session.observeAttachment?.(receipt); + }, + }), request: allocation, // The session is shared by every request of this generation. Only the generation // owns its lifetime; a caller's owner signal bounds that caller's request alone. @@ -363,7 +391,26 @@ export function createHostedHttpGenerationExecutor( const current = session!; const signal = AbortSignal.any([request.signal, current.signal, lifetime.signal]); const owned = new Request(request, { signal }); - const response = await current.runOwned(() => client!.fetch(owned, context.tracing)); + const response = await current.runOwned(() => { + signal.throwIfAborted(); + if (observeDispatch) { + if (!attachment) throw new Error("HTTP executor attachment receipt is unavailable"); + const { owner, source, environmentId } = input.installation; + if (owner.scopeKind !== "project") throw new Error("HTTP executor owner is unverified"); + observeDispatch( + Object.freeze({ + attachment, + brokerInstanceId: input.session.expectedBrokerInstanceId, + projectId: owner.projectId, + environmentId, + releaseId: source.releaseId, + traceparent: formatTraceparent(parseTraceparent(context.tracing?.traceparent)), + }), + ); + signal.throwIfAborted(); + } + return client!.fetch(owned, context.tracing); + }); return completeOnResponseBodyConsumption( response, finish, diff --git a/src/server/isolated-http/hosted-http-broker.test.ts b/src/server/isolated-http/hosted-http-broker.test.ts index 862f25cf10..850eb8005e 100644 --- a/src/server/isolated-http/hosted-http-broker.test.ts +++ b/src/server/isolated-http/hosted-http-broker.test.ts @@ -53,6 +53,52 @@ describe("hosted HTTP executor broker", () => { } }); + it("joins each warm dispatch to the same authenticated attachment and its own host trace", async () => { + const f = fixture(() => new Response("app")); + Object.assign(f.input.installation.source, { type: "environment", environmentName: "staging" }); + const records: unknown[] = []; + const broker = createHostedHttpBroker({ + maxActive: 1, + observeDispatch: (receipt) => void records.push(receipt), + }); + const traces = [ + "00-11111111111111111111111111111111-2222222222222222-01", + "00-33333333333333333333333333333333-4444444444444444-01", + ]; + try { + for (const [index, traceparent] of traces.entries()) { + const response = await broker.fetch( + new Request("https://app.example/api", { headers: { traceparent: "untrusted" } }), + { + ...f.input, + tracing: { + traceparent: index === 0 ? traceparent : traceparent + "-private", + onRecords() {}, + }, + }, + ); + assertEquals(await response.text(), "app"); + } + assertEquals(f.calls.filter((call) => call === "allocate").length, 1); + const attachment = (records[0] as { attachment: unknown }).attachment; + assertExists(attachment); + assertEquals( + records, + traces.map((traceparent) => ({ + attachment, + brokerInstanceId: "broker-a", + projectId: "project-a", + environmentId: "environment-a", + releaseId: "release-a", + traceparent, + })), + ); + } finally { + await broker.shutdown(); + await broker.settled; + } + }); + it("delivers only the matched application snapshot through the allocation channel", async () => { let appValue: string | undefined; const setup = fixture(() => new Response(appValue), async (peer, installed) => { diff --git a/src/server/isolated-http/hosted-http-broker.ts b/src/server/isolated-http/hosted-http-broker.ts index e7b80bb36f..4d00f558ec 100644 --- a/src/server/isolated-http/hosted-http-broker.ts +++ b/src/server/isolated-http/hosted-http-broker.ts @@ -64,6 +64,8 @@ export interface HostedHttpInput { /** Admission and lifetime limits of the hosted HTTP broker's warm executors. */ export interface HostedHttpBrokerOptions extends HostedExecutorSessionPoolOptions { + observeDispatch?: + import("./generation-executor.ts").HostedHttpGenerationExecutorOptions["observeDispatch"]; /** Requests one executor serves at once. Default 8, maximum 16. */ maxConcurrentRequestsPerExecutor?: number; /** Retire an executor after this long without a request in flight. Default 60 seconds, maximum 1 hour. */ @@ -176,6 +178,7 @@ function bounded(value: number, minimum: number, maximum: number, name: string): * executor with the same identity is refused, never rebound. */ export function createHostedHttpBroker(options: HostedHttpBrokerOptions) { + const observeDispatch = options.observeDispatch; const sessions = createHostedExecutorSessionPool(options); const perExecutor = bounded( options.maxConcurrentRequestsPerExecutor ?? 8, @@ -530,6 +533,7 @@ export function createHostedHttpBroker(options: HostedHttpBrokerOptions) { input, sessions, context: (value) => contexts.get(value), + ...(observeDispatch === undefined ? {} : { observeDispatch }), idleTimeoutMs, retireBeforeDeadlineMs: drainTimeoutMs, }); diff --git a/src/server/isolated-http/installed-http-broker.test.ts b/src/server/isolated-http/installed-http-broker.test.ts index e2d37478a0..5b55608711 100644 --- a/src/server/isolated-http/installed-http-broker.test.ts +++ b/src/server/isolated-http/installed-http-broker.test.ts @@ -66,11 +66,16 @@ describe("installed HTTP broker", () => { }); it("uses the configured bounded lifetime for long requests and streams", async () => { const f = createHostedHttpFixture(() => new Response("A")); + let attachments = 0; + const observeAttachment = () => { + attachments++; + }; const options = { installations: [{ image: f.input.session.expectedImage, configuration }], expectedBrokerInstanceId: "broker-a", allocator: f.input.session.allocator, connectTransport: f.input.session.connectTransport, + observeAttachment, maxActive: 1, authorize: async () => {}, requestLifetimeMs: 180_000, @@ -79,6 +84,8 @@ describe("installed HTTP broker", () => { try { const startedAt = Date.now(); const input = await b.hostedHttp.resolve(authority, new AbortController().signal); + assertEquals(input.session.observeAttachment, observeAttachment); + assertEquals(attachments, 0); const finishedAt = Date.now(); const { hardDeadlineAt, prepareDeadlineAt } = input.session.request; // Lifetimes start after authorization, which can cross a clock tick. diff --git a/src/server/isolated-http/installed-http-broker.ts b/src/server/isolated-http/installed-http-broker.ts index 9478d5ba3d..03c0e6964c 100644 --- a/src/server/isolated-http/installed-http-broker.ts +++ b/src/server/isolated-http/installed-http-broker.ts @@ -11,7 +11,7 @@ import { type ExecutorHttpApplicationConfiguration, snapshotExecutorHttpApplicationConfiguration, } from "./application-configuration.ts"; -import { createHostedHttpBroker } from "./hosted-http-broker.ts"; +import { createHostedHttpBroker, type HostedHttpBrokerOptions } from "./hosted-http-broker.ts"; import type { HostedHttpIngressOptions, HostedHttpRequestAuthority, @@ -26,6 +26,10 @@ export interface InstalledHttpBrokerOptions { expectedBrokerInstanceId: string; allocator: HostedExecutorSessionOptions["allocator"]; connectTransport: HostedExecutorSessionOptions["connectTransport"]; + /** Trusted observer for a newly authenticated executor, once per attachment. */ + observeAttachment?: HostedExecutorSessionOptions["observeAttachment"]; + /** Trusted observer for each dispatched request, including warm reuse. */ + observeDispatch?: HostedHttpBrokerOptions["observeDispatch"]; /** Check current project/environment/release authority against the source API on every request. */ authorize: { (authority: HostedHttpRequestAuthority, signal: AbortSignal): Promise; @@ -112,8 +116,13 @@ export function createInstalledHttpBroker(options: InstalledHttpBrokerOptions) { const authorize = options.authorize.bind(options); const allocator = options.allocator; const connectTransport = options.connectTransport; + const observeAttachment = options.observeAttachment; const owner = new AbortController(); - const broker = createHostedHttpBroker({ maxActive: options.maxActive, signal: owner.signal }); + const broker = createHostedHttpBroker({ + maxActive: options.maxActive, + signal: owner.signal, + ...(options.observeDispatch === undefined ? {} : { observeDispatch: options.observeDispatch }), + }); let pendingAuthorizations = 0; const authorizationDrain = Promise.withResolvers(); const finishAuthorization = () => { @@ -173,6 +182,7 @@ export function createInstalledHttpBroker(options: InstalledHttpBrokerOptions) { expectedImage: selected.image, allocator, connectTransport, + ...(observeAttachment === undefined ? {} : { observeAttachment }), ownerSignal: owner.signal, }, installation: {