diff --git a/.server-changes/streams-version-s2-guard.md b/.server-changes/streams-version-s2-guard.md new file mode 100644 index 0000000000..0fbfcb1514 --- /dev/null +++ b/.server-changes/streams-version-s2-guard.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Realtime streams written inside a chat session run now use the same backend as the session itself, and runs are no longer created against a backend that cannot serve them. diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts b/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts index 2bd7bc5650..904bc9d4a9 100644 --- a/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts +++ b/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts @@ -97,7 +97,8 @@ const { action } = createActionApiRoute( traceContext, spanParentAsLink: spanParentAsLink === 1, realtimeStreamsVersion: determineRealtimeStreamsVersion( - realtimeStreamsVersion ?? undefined + realtimeStreamsVersion ?? undefined, + authentication.environment.organization.streamBasinName ), }); diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts b/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts index 6165049330..c0de0c59f7 100644 --- a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts +++ b/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts @@ -144,7 +144,8 @@ const { action, loader } = createActionApiRoute( spanParentAsLink: spanParentAsLink === 1, oneTimeUseToken, realtimeStreamsVersion: determineRealtimeStreamsVersion( - realtimeStreamsVersion ?? undefined + realtimeStreamsVersion ?? undefined, + authentication.environment.organization.streamBasinName ), triggerSource: isFromWorker ? "sdk" diff --git a/apps/webapp/app/routes/api.v1.tasks.batch.ts b/apps/webapp/app/routes/api.v1.tasks.batch.ts index 5c9202d6fe..2ea5ebb3b2 100644 --- a/apps/webapp/app/routes/api.v1.tasks.batch.ts +++ b/apps/webapp/app/routes/api.v1.tasks.batch.ts @@ -116,7 +116,8 @@ const { action, loader } = createActionApiRoute( spanParentAsLink: spanParentAsLink === 1, oneTimeUseToken, realtimeStreamsVersion: determineRealtimeStreamsVersion( - realtimeStreamsVersion ?? undefined + realtimeStreamsVersion ?? undefined, + authentication.environment.organization.streamBasinName ), triggerSource: isFromWorker ? "sdk" : (sanitizeTriggerSource(triggerSourceHeader) ?? "api"), triggerAction: "trigger", diff --git a/apps/webapp/app/routes/api.v2.tasks.batch.ts b/apps/webapp/app/routes/api.v2.tasks.batch.ts index 5dcbf13e0f..8bdfdb569d 100644 --- a/apps/webapp/app/routes/api.v2.tasks.batch.ts +++ b/apps/webapp/app/routes/api.v2.tasks.batch.ts @@ -143,7 +143,8 @@ const { action, loader } = createActionApiRoute( spanParentAsLink: spanParentAsLink === 1, oneTimeUseToken, realtimeStreamsVersion: determineRealtimeStreamsVersion( - realtimeStreamsVersion ?? undefined + realtimeStreamsVersion ?? undefined, + authentication.environment.organization.streamBasinName ), triggerSource: isFromWorker ? "sdk" : (sanitizeTriggerSource(triggerSourceHeader) ?? "api"), triggerAction: "trigger", diff --git a/apps/webapp/app/routes/api.v3.batches.ts b/apps/webapp/app/routes/api.v3.batches.ts index 071bb783b8..ce301718f0 100644 --- a/apps/webapp/app/routes/api.v3.batches.ts +++ b/apps/webapp/app/routes/api.v3.batches.ts @@ -167,7 +167,8 @@ const { action, loader } = createActionApiRoute( spanParentAsLink: spanParentAsLink === 1, oneTimeUseToken, realtimeStreamsVersion: determineRealtimeStreamsVersion( - realtimeStreamsVersion ?? undefined + realtimeStreamsVersion ?? undefined, + authentication.environment.organization.streamBasinName ), triggerSource: isFromWorker ? "sdk" : (sanitizeTriggerSource(triggerSourceHeader) ?? "api"), }); diff --git a/apps/webapp/app/services/realtime/realtimeStreamsVersion.ts b/apps/webapp/app/services/realtime/realtimeStreamsVersion.ts new file mode 100644 index 0000000000..a7d207c713 --- /dev/null +++ b/apps/webapp/app/services/realtime/realtimeStreamsVersion.ts @@ -0,0 +1,44 @@ +/** + * Pure realtime-streams version resolution. Deliberately free of `env` and of + * any module-scope singletons so it can be tested with injected values, the + * same split as `nativeRealtimeClient` and `nativeRealtimeClientInstance`. + * The env-bound wrapper is `determineRealtimeStreamsVersion` in + * `v1StreamsGlobal.server.ts`. + */ + +export type RealtimeStreamsVersionConfig = { + defaultVersion: "v1" | "v2"; + /** A basin that will actually resolve at read/write time, or undefined if none will. */ + basin?: string; + accessToken?: string; + skipAccessTokens: boolean; +}; + +/** + * Resolve the streams version to stamp on a run, falling back to the + * deployment default when the caller expresses no preference. + * + * v2 is only ever returned when S2 can actually serve it. A run stamped v2 on a + * deployment without S2 is unusable: `getRealtimeStreamInstance` throws for the + * life of the run, and no read or write against its streams can succeed. v1 is + * a working backend, so an unsatisfiable v2 degrades to it. + * + * The basin must be one that will actually resolve later. Enabling per-org + * basins is not enough on its own: provisioning is out of band, so an + * unprovisioned organization has no basin and a global setting may not exist + * to fall back to. + */ +export function resolveRealtimeStreamsVersion( + streamVersion: string | undefined, + config: RealtimeStreamsVersionConfig +): "v1" | "v2" { + const requested = streamVersion ?? config.defaultVersion; + + if (requested !== "v2") { + return "v1"; + } + + const hasCredentials = Boolean(config.accessToken) || config.skipAccessTokens; + + return hasCredentials && Boolean(config.basin) ? "v2" : "v1"; +} diff --git a/apps/webapp/app/services/realtime/sessionRunManager.server.ts b/apps/webapp/app/services/realtime/sessionRunManager.server.ts index 10cd966d95..53d436e1e5 100644 --- a/apps/webapp/app/services/realtime/sessionRunManager.server.ts +++ b/apps/webapp/app/services/realtime/sessionRunManager.server.ts @@ -8,6 +8,7 @@ import { logger } from "~/services/logger.server"; import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server"; import { TriggerTaskService } from "~/v3/services/triggerTask.server"; import { isFinalRunStatus } from "~/v3/taskStatus"; +import { determineRealtimeStreamsVersion } from "./v1StreamsGlobal.server"; /** * Schema for `Session.triggerConfig` (stored as JSONB). The wire-format @@ -275,6 +276,12 @@ export async function ensureRunForSession( * Trigger a single run for a session. Builds `TriggerTaskRequestBody` * by shallow-merging `payloadOverrides` over `config.basePayload` and * threading `config`'s machine/queue/tags through the trigger options. + * + * A session's own channels are always v2, so the run is stamped to match + * rather than inheriting the `realtimeStreamsVersion` column default. Without + * this, run-scoped `streams.*` calls inside a session run resolve to v1 while + * the session it belongs to is on v2. `determineRealtimeStreamsVersion` + * degrades to v1 where v2 streams are not configured. */ async function triggerSessionRun(params: { session: Pick; @@ -310,6 +317,10 @@ async function triggerSessionRun(params: { const result = await service.call(session.taskIdentifier, environment, body, { triggerSource: "session", triggerAction: "trigger", + realtimeStreamsVersion: determineRealtimeStreamsVersion( + "v2", + environment.organization.streamBasinName + ), }); if (!result) { diff --git a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts index a6801b8a1b..27305f676e 100644 --- a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts +++ b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts @@ -10,6 +10,10 @@ import { singleton } from "~/utils/singleton"; import type { AuthenticatedEnvironment } from "../apiAuth.server"; import { RedisRealtimeStreams } from "./redisRealtimeStreams.server"; import { S2RealtimeStreams } from "./s2realtimeStreams.server"; +import { + resolveRealtimeStreamsVersion, + type RealtimeStreamsVersionConfig, +} from "./realtimeStreamsVersion"; import type { StreamIngestor, StreamResponder } from "./types"; function initializeRedisRealtimeStreams() { @@ -96,20 +100,24 @@ function streamPrefixFor(environment: AuthenticatedEnvironment, basin: string): return segments.join("/"); } -export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" { - if (!streamVersion) { - return env.REALTIME_STREAMS_DEFAULT_VERSION; - } - - if ( - streamVersion === "v2" && - env.REALTIME_STREAMS_S2_BASIN && - (env.REALTIME_STREAMS_S2_ACCESS_TOKEN || env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true") - ) { - return "v2"; - } +export type { RealtimeStreamsVersionConfig }; - return "v1"; +/** + * Pass `organizationBasinName` wherever the caller has it. It mirrors the + * organization step of {@link resolveStreamBasin}, and is what lets a + * per-org-basin deployment with no global setting resolve v2 for a + * provisioned organization while an unprovisioned one still degrades to v1. + */ +export function determineRealtimeStreamsVersion( + streamVersion?: string, + organizationBasinName?: string | null +): "v1" | "v2" { + return resolveRealtimeStreamsVersion(streamVersion, { + defaultVersion: env.REALTIME_STREAMS_DEFAULT_VERSION, + basin: organizationBasinName ?? env.REALTIME_STREAMS_S2_BASIN, + accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN, + skipAccessTokens: env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true", + }); } const s2RealtimeStreamsCache = singleton( diff --git a/apps/webapp/app/v3/services/replayTaskRun.server.ts b/apps/webapp/app/v3/services/replayTaskRun.server.ts index d626ec2ae8..750427fb32 100644 --- a/apps/webapp/app/v3/services/replayTaskRun.server.ts +++ b/apps/webapp/app/v3/services/replayTaskRun.server.ts @@ -163,7 +163,8 @@ export class ReplayTaskRunService extends BaseService { traceparent: `00-${existingTaskRun.traceId}-${existingTaskRun.spanId}-01`, }, realtimeStreamsVersion: determineRealtimeStreamsVersion( - existingTaskRun.realtimeStreamsVersion + existingTaskRun.realtimeStreamsVersion, + authenticatedEnvironment.organization.streamBasinName ), triggerSource: overrideOptions.triggerSource ?? "api", triggerAction: "replay", diff --git a/apps/webapp/test/realtimeServices.replicaLag.test.ts b/apps/webapp/test/realtimeServices.replicaLag.test.ts index 02bb605afe..6a302dcfd9 100644 --- a/apps/webapp/test/realtimeServices.replicaLag.test.ts +++ b/apps/webapp/test/realtimeServices.replicaLag.test.ts @@ -39,6 +39,15 @@ const replicaHolder = vi.hoisted(() => ({ client: undefined as any })); const storeHolder = vi.hoisted(() => ({ store: undefined as any })); // Records every TriggerTaskService.call so read 3 can assert NO double-trigger and read 4 can assert // which previousRunId the resolveRunFriendlyId fallback forwarded. +const versionCalls = vi.hoisted(() => [] as Array<{ requested?: string; basin?: string | null }>); + +vi.mock("~/services/realtime/v1StreamsGlobal.server", () => ({ + determineRealtimeStreamsVersion: (requested?: string, basin?: string | null) => { + versionCalls.push({ requested, basin }); + return "v2"; + }, +})); + const triggerState = vi.hoisted(() => ({ calls: [] as Array<{ taskIdentifier: string; body: any; options: any }>, result: { run: { id: "", friendlyId: "" } } as { run: { id: string; friendlyId: string } }, @@ -331,7 +340,10 @@ describe("realtime-svc — replica-lag guards", () => { const result = await ensureRunForSession({ session, - environment: { id: seed.environment.id } as unknown as AuthenticatedEnvironment, + environment: { + id: seed.environment.id, + organization: { streamBasinName: null }, + } as unknown as AuthenticatedEnvironment, reason: "manual", }); @@ -386,6 +398,7 @@ describe("realtime-svc — replica-lag guards", () => { triggerConfig: { basePayload: {} }, currentRunId: callingRunId, currentRunVersion: 0, + streamBasinName: "session-pinned-basin", }, }); @@ -394,6 +407,7 @@ describe("realtime-svc — replica-lag guards", () => { replicaHolder.client = replica.client; storeHolder.store = writerStore; triggerState.calls.length = 0; + versionCalls.length = 0; const newRunId = cuidRunId(`sn${seq}`); const newFriendlyId = `run_${suffix}_new`; triggerState.result = { run: { id: newRunId, friendlyId: newFriendlyId } }; @@ -401,7 +415,10 @@ describe("realtime-svc — replica-lag guards", () => { const result = await swapSessionRun({ session: sessionRow, callingRunId, - environment: { id: seed.environment.id } as unknown as AuthenticatedEnvironment, + environment: { + id: seed.environment.id, + organization: { streamBasinName: null }, + } as unknown as AuthenticatedEnvironment, reason: "upgrade", }); @@ -412,6 +429,7 @@ describe("realtime-svc — replica-lag guards", () => { // previousRunId forwarded to the triggered run is the calling run's cuid (documented fallback). expect(triggerState.calls).toHaveLength(1); expect(triggerState.calls[0]!.body.payload.previousRunId).toBe(callingRunId); + expect(versionCalls.at(-1)).toEqual({ requested: "v2", basin: null }); expect(replica.wasHit("taskRun")).toBe(true); // Proof the null was lag-induced: the primary holds the resolvable friendlyId (≠ the cuid). diff --git a/apps/webapp/test/realtimeStreamsVersion.test.ts b/apps/webapp/test/realtimeStreamsVersion.test.ts new file mode 100644 index 0000000000..4ffa973d21 --- /dev/null +++ b/apps/webapp/test/realtimeStreamsVersion.test.ts @@ -0,0 +1,116 @@ +import { describe, expect, it } from "vitest"; +import { + resolveRealtimeStreamsVersion, + type RealtimeStreamsVersionConfig, +} from "~/services/realtime/realtimeStreamsVersion"; + +const NO_S2: RealtimeStreamsVersionConfig = { + defaultVersion: "v1", + basin: undefined, + accessToken: undefined, + skipAccessTokens: false, +}; + +const GLOBAL_BASIN: RealtimeStreamsVersionConfig = { + ...NO_S2, + basin: "a-basin", + accessToken: "a-token", +}; + +const ORG_BASIN: RealtimeStreamsVersionConfig = { + ...NO_S2, + basin: "an-org-basin", + accessToken: "a-token", +}; + +describe("resolveRealtimeStreamsVersion", () => { + it("honours an explicit v2 when a global basin is configured", () => { + expect(resolveRealtimeStreamsVersion("v2", GLOBAL_BASIN)).toBe("v2"); + }); + + it("honours an explicit v2 when only an org basin is resolvable", () => { + expect(resolveRealtimeStreamsVersion("v2", ORG_BASIN)).toBe("v2"); + }); + + it("accepts a skip-tokens deployment as credentialed", () => { + expect( + resolveRealtimeStreamsVersion("v2", { + ...NO_S2, + basin: "a-basin", + skipAccessTokens: true, + }) + ).toBe("v2"); + }); + + it("degrades an explicit v2 to v1 when S2 is not configured", () => { + expect(resolveRealtimeStreamsVersion("v2", NO_S2)).toBe("v1"); + }); + + it("falls back to the default version when the caller expresses no preference", () => { + expect( + resolveRealtimeStreamsVersion(undefined, { ...GLOBAL_BASIN, defaultVersion: "v2" }) + ).toBe("v2"); + }); + + it("degrades a v2 default to v1 when S2 is not configured", () => { + expect(resolveRealtimeStreamsVersion(undefined, { ...NO_S2, defaultVersion: "v2" })).toBe("v1"); + }); + + it("keeps a v2 default on v2 when only an org basin is resolvable", () => { + expect(resolveRealtimeStreamsVersion(undefined, { ...ORG_BASIN, defaultVersion: "v2" })).toBe( + "v2" + ); + }); + + it("requires credentials, not just a basin", () => { + const basinOnly = { ...NO_S2, basin: "a-basin", defaultVersion: "v2" as const }; + expect(resolveRealtimeStreamsVersion(undefined, basinOnly)).toBe("v1"); + expect(resolveRealtimeStreamsVersion("v2", basinOnly)).toBe("v1"); + }); + + it("requires a basin, not just credentials", () => { + const tokenOnly = { ...NO_S2, accessToken: "a-token", defaultVersion: "v2" as const }; + expect(resolveRealtimeStreamsVersion(undefined, tokenOnly)).toBe("v1"); + expect(resolveRealtimeStreamsVersion("v2", tokenOnly)).toBe("v1"); + }); + + it("keeps an explicit v1 on v1 even where S2 is available", () => { + expect(resolveRealtimeStreamsVersion("v1", { ...GLOBAL_BASIN, defaultVersion: "v2" })).toBe( + "v1" + ); + }); + + it("treats an unrecognised version as v1", () => { + expect(resolveRealtimeStreamsVersion("v3", GLOBAL_BASIN)).toBe("v1"); + }); +}); + +describe("resolveRealtimeStreamsVersion invariant", () => { + const BASINS = [undefined, "", "a-basin"]; + const TOKENS = [undefined, "a-token"]; + const SKIPS = [false, true]; + const DEFAULTS: Array<"v1" | "v2"> = ["v1", "v2"]; + const REQUESTED = [undefined, "v1", "v2", "v3"]; + + it("only returns v2 when a basin and credentials are both present, for every configuration", () => { + const counterexamples: string[] = []; + + for (const basin of BASINS) { + for (const accessToken of TOKENS) { + for (const skipAccessTokens of SKIPS) { + for (const defaultVersion of DEFAULTS) { + for (const requested of REQUESTED) { + const config = { defaultVersion, basin, accessToken, skipAccessTokens }; + const usable = Boolean(basin) && (Boolean(accessToken) || skipAccessTokens); + if (resolveRealtimeStreamsVersion(requested, config) === "v2" && !usable) { + counterexamples.push(JSON.stringify({ requested, ...config })); + } + } + } + } + } + } + + expect(counterexamples).toEqual([]); + }); +}); diff --git a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts new file mode 100644 index 0000000000..e500a3eb2b --- /dev/null +++ b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts @@ -0,0 +1,147 @@ +/** + * Full-stack e2e for which realtime streams backend a Session's run lands on. + * + * Boots the real webapp + Postgres + Redis + s2-lite (via + * startSessionStreamTestServer), creates a Session through the public API so + * the run is triggered by the real `sessionRunManager` path, then appends to a + * run-scoped stream exactly as `streams.append()` does and checks where the + * bytes actually went. + * + * The harness starts the webapp with `REALTIME_STREAMS_DEFAULT_VERSION: "v2"` + * and a live S2, so a run landing on v1 here is not a configuration gap. It + * means the trigger path never asked, and fell through to the + * `realtimeStreamsVersion` column default. + * + * Requires a pre-built webapp: pnpm run build --filter webapp + */ +import { randomBytes } from "crypto"; +import Redis from "ioredis"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import type { SessionStreamTestServer } from "@internal/testcontainers/webapp"; +import { startSessionStreamTestServer } from "@internal/testcontainers/webapp"; +import { seedTestEnvironment } from "./helpers/seedTestEnvironment"; + +vi.setConfig({ testTimeout: 120_000, hookTimeout: 180_000 }); + +let server: SessionStreamTestServer; + +beforeAll(async () => { + server = await startSessionStreamTestServer(); +}, 180_000); + +afterAll(async () => { + await server?.stop(); +}, 120_000); + +const STREAM_ID = "frames"; +const PART_ID = "part"; +const FRAME_BYTES = 250 * 1024; +const FRAME_COUNT = 8; + +/** Mirrors `S2RealtimeStreams.toStreamName` on the shared-basin prefix. */ +function runStreamName(p: { + orgId: string; + envSlug: string; + envId: string; + runId: string; + streamId: string; +}): string { + return `org/${p.orgId}/env/${p.envSlug}/${p.envId}/runs/${p.runId}/${p.streamId}`; +} + +/** Mirrors the `keyPrefix` + key shape in `v1StreamsGlobal` / `RedisRealtimeStreams`. */ +function redisStreamKey(runId: string, streamId: string): string { + return `tr:realtime:streams:stream:${runId}:${streamId}`; +} + +function framesFound(body: string): number { + return Array.from({ length: FRAME_COUNT }, (_, i) => `${PART_ID}-${i}`).filter((id) => + body.includes(id) + ).length; +} + +async function s2Body(streamName: string): Promise { + const qs = new URLSearchParams({ seq_num: "0", clamp: "true", wait: "0" }); + const res = await fetch( + `${server.s2.endpoint}/v1/streams/${encodeURIComponent(streamName)}/records?${qs}`, + { + headers: { + Authorization: "Bearer ignored", + Accept: "text/event-stream", + "S2-Format": "raw", + "S2-Basin": server.s2.basin, + }, + } + ); + + if (res.status === 404) return ""; + expect(res.ok).toBe(true); + + return res.text(); +} + +describe("session runs and the realtime streams backend", () => { + it("stamps the run v2 and routes a run-scoped stream to S2, not Redis", async () => { + const { organization, environment, apiKey } = await seedTestEnvironment(server.prisma); + + const createRes = await fetch(`${server.webapp.baseUrl}/api/v1/sessions`, { + method: "POST", + headers: { Authorization: `Bearer ${apiKey}`, "Content-Type": "application/json" }, + body: JSON.stringify({ + type: "chat.agent", + externalId: `e2e-${randomBytes(6).toString("hex")}`, + taskIdentifier: "e2e-browser-agent", + triggerConfig: { basePayload: {} }, + }), + }); + + expect(createRes.ok).toBe(true); + const created = (await createRes.json()) as { runId: string }; + expect(created.runId).toBeTruthy(); + + const run = await server.prisma.taskRun.findFirstOrThrow({ + where: { friendlyId: created.runId }, + select: { realtimeStreamsVersion: true }, + }); + + const appendStatuses: number[] = []; + for (let i = 0; i < FRAME_COUNT; i++) { + const res = await fetch( + `${server.webapp.baseUrl}/realtime/v1/streams/${created.runId}/self/${STREAM_ID}/append`, + { + method: "POST", + headers: { + Authorization: `Bearer ${apiKey}`, + "Content-Type": "text/plain", + "X-Part-Id": `${PART_ID}-${i}`, + }, + body: JSON.stringify({ i, frame: "a".repeat(FRAME_BYTES) }), + } + ); + appendStatuses.push(res.status); + } + + expect(appendStatuses).toEqual(Array.from({ length: FRAME_COUNT }, () => 200)); + + const streamName = runStreamName({ + orgId: organization.id, + envSlug: environment.slug, + envId: environment.id, + runId: created.runId, + streamId: STREAM_ID, + }); + const redis = new Redis({ host: server.redis.host, port: server.redis.port }); + let observed: { version: string; framesInS2: number; keyInRedis: boolean }; + try { + observed = { + version: run.realtimeStreamsVersion, + framesInS2: framesFound(await s2Body(streamName)), + keyInRedis: (await redis.exists(redisStreamKey(created.runId, STREAM_ID))) === 1, + }; + } finally { + redis.disconnect(); + } + + expect(observed).toEqual({ version: "v2", framesInS2: FRAME_COUNT, keyInRedis: false }); + }); +}); diff --git a/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts b/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts new file mode 100644 index 0000000000..5e7527c901 --- /dev/null +++ b/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts @@ -0,0 +1,177 @@ +/** + * Full-stack e2e for the per-org-basin configuration: S2 credentials present, + * no global basin, so whether a run can use v2 depends entirely on whether its + * organization has been provisioned one. + * + * The sibling `sessionRunStreamsBackend` e2e runs with a global basin set, + * which makes every basin value work and hides this whole class of bug. Here a + * run stamped v2 without a resolvable basin is not a degraded experience, it + * throws on every stream operation for the life of the run, so both directions + * are asserted: a provisioned organization reaches S2, and an unprovisioned one + * degrades to v1 and keeps working on Redis. + * + * Scope: both cases assert run-scoped streams only. Neither drives a session + * channel, so neither says anything about `.in`/`.out`. That matters for the + * unprovisioned case, where the session's own channels cannot resolve a basin + * at all and fail: the run degrading to v1 is what keeps working there, not the + * session. Do not read these as evidence that a session is healthy. + * + * Which basin the trigger path reads is pinned separately, by the swap case in + * `realtimeServices.replicaLag.test.ts`, which asserts the organization's basin + * reaches the resolver even when the session row carries one of its own. + * + * Requires a pre-built webapp: pnpm run build --filter webapp + */ +import { randomBytes } from "crypto"; +import Redis from "ioredis"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import type { SessionStreamTestServer } from "@internal/testcontainers/webapp"; +import { startSessionStreamTestServer } from "@internal/testcontainers/webapp"; +import { seedTestEnvironment } from "./helpers/seedTestEnvironment"; + +vi.setConfig({ testTimeout: 120_000, hookTimeout: 180_000 }); + +let server: SessionStreamTestServer; + +beforeAll(async () => { + server = await startSessionStreamTestServer({ + extraEnv: { + REALTIME_STREAMS_S2_BASIN: "", + REALTIME_STREAMS_PER_ORG_BASINS_ENABLED: "true", + }, + }); +}, 180_000); + +afterAll(async () => { + await server?.stop(); +}, 120_000); + +const STREAM_ID = "frames"; + +/** Per-org basins drop the `org/{id}` segment; see `streamPrefixFor`. */ +function perOrgStreamName(p: { envSlug: string; envId: string; runId: string }): string { + return `env/${p.envSlug}/${p.envId}/runs/${p.runId}/${STREAM_ID}`; +} + +function redisStreamKey(runId: string): string { + return `tr:realtime:streams:stream:${runId}:${STREAM_ID}`; +} + +async function s2HasRecords(basin: string, streamName: string): Promise { + const qs = new URLSearchParams({ seq_num: "0", clamp: "true", wait: "0" }); + const res = await fetch( + `${server.s2.endpoint}/v1/streams/${encodeURIComponent(streamName)}/records?${qs}`, + { + headers: { + Authorization: "Bearer ignored", + Accept: "text/event-stream", + "S2-Format": "raw", + "S2-Basin": basin, + }, + } + ); + if (!res.ok) return false; + return (await res.text()).includes(STREAM_ID); +} + +async function createSessionRun(apiKey: string, taskIdentifier: string): Promise { + const res = await fetch(`${server.webapp.baseUrl}/api/v1/sessions`, { + method: "POST", + headers: { Authorization: `Bearer ${apiKey}`, "Content-Type": "application/json" }, + body: JSON.stringify({ + type: "chat.agent", + externalId: `e2e-${randomBytes(6).toString("hex")}`, + taskIdentifier, + triggerConfig: { basePayload: {} }, + }), + }); + expect(res.ok).toBe(true); + return ((await res.json()) as { runId: string }).runId; +} + +async function appendFrame(apiKey: string, runId: string): Promise { + const res = await fetch( + `${server.webapp.baseUrl}/realtime/v1/streams/${runId}/self/${STREAM_ID}/append`, + { + method: "POST", + headers: { + Authorization: `Bearer ${apiKey}`, + "Content-Type": "text/plain", + "X-Part-Id": STREAM_ID, + }, + body: JSON.stringify({ frame: "a".repeat(1024) }), + } + ); + return res.status; +} + +describe("session runs with per-org basins and no global basin", () => { + it("reaches S2 for a provisioned organization", async () => { + const { organization, environment, apiKey } = await seedTestEnvironment(server.prisma); + const basin = server.s2.basin; + + await server.prisma.organization.update({ + where: { id: organization.id }, + data: { streamBasinName: basin }, + }); + + const runId = await createSessionRun(apiKey, "e2e-per-org-provisioned"); + + const run = await server.prisma.taskRun.findFirstOrThrow({ + where: { friendlyId: runId }, + select: { realtimeStreamsVersion: true, streamBasinName: true }, + }); + + expect(await appendFrame(apiKey, runId)).toBe(200); + + const redis = new Redis({ host: server.redis.host, port: server.redis.port }); + let observed; + try { + observed = { + version: run.realtimeStreamsVersion, + runBasin: run.streamBasinName, + inS2: await s2HasRecords( + basin, + perOrgStreamName({ envSlug: environment.slug, envId: environment.id, runId }) + ), + keyInRedis: (await redis.exists(redisStreamKey(runId))) === 1, + }; + } finally { + redis.disconnect(); + } + + expect(observed).toEqual({ version: "v2", runBasin: basin, inS2: true, keyInRedis: false }); + }); + + it("degrades to v1 for an unprovisioned organization, keeping its run-scoped streams usable", async () => { + const { organization, apiKey } = await seedTestEnvironment(server.prisma); + + await server.prisma.organization.update({ + where: { id: organization.id }, + data: { streamBasinName: null }, + }); + + const runId = await createSessionRun(apiKey, "e2e-per-org-unprovisioned"); + + const run = await server.prisma.taskRun.findFirstOrThrow({ + where: { friendlyId: runId }, + select: { realtimeStreamsVersion: true, streamBasinName: true }, + }); + + expect(await appendFrame(apiKey, runId)).toBe(200); + + const redis = new Redis({ host: server.redis.host, port: server.redis.port }); + let observed; + try { + observed = { + version: run.realtimeStreamsVersion, + runBasin: run.streamBasinName, + keyInRedis: (await redis.exists(redisStreamKey(runId))) === 1, + }; + } finally { + redis.disconnect(); + } + + expect(observed).toEqual({ version: "v1", runBasin: null, keyInRedis: true }); + }); +}); diff --git a/internal-packages/testcontainers/src/webapp.ts b/internal-packages/testcontainers/src/webapp.ts index 22d3485f8f..2ff73b1b52 100644 --- a/internal-packages/testcontainers/src/webapp.ts +++ b/internal-packages/testcontainers/src/webapp.ts @@ -272,6 +272,12 @@ export type { StartedS2Container } from "./s2"; export interface SessionStreamTestServer extends TestServer { s2: StartedS2Container; minio: StartedMinIOContainer; + /** + * Mapped connection for the same Redis the webapp under test uses. Lets a + * test assert which backend a stream actually landed on, rather than + * inferring it from the absence of records in S2. + */ + redis: { host: string; port: number }; } /** @@ -280,7 +286,9 @@ export interface SessionStreamTestServer extends TestServer { * process reaching every container over its mapped port, so the S2 endpoint is * the mapped localhost URL (the docker-network alias is unusable from the host). */ -export async function startSessionStreamTestServer(): Promise { +export async function startSessionStreamTestServer( + options: StartWebappOptions = {} +): Promise { const network = await new Network().start(); let pgContainer: Awaited>["container"] | undefined; @@ -322,6 +330,7 @@ export async function startSessionStreamTestServer(): Promise console.error("network.stop failed:", err)); }; - return { webapp, prisma: prisma!, databaseUrl: pgUrl!, s2: s2!, minio: minio!, stop }; + return { + webapp, + prisma: prisma!, + databaseUrl: pgUrl!, + s2: s2!, + minio: minio!, + redis: { host: redisContainer!.getHost(), port: redisContainer!.getPort() }, + stop, + }; }