diff --git a/apps/api/src/routes/v1/session-replay.http.ts b/apps/api/src/routes/v1/session-replay.http.ts index 179f4d48b..95d190c73 100644 --- a/apps/api/src/routes/v1/session-replay.http.ts +++ b/apps/api/src/routes/v1/session-replay.http.ts @@ -1,7 +1,6 @@ import { HttpApiBuilder } from "effect/unstable/httpapi" import { CurrentTenant, - GetReplayEventsResponse, GetReplayResponse, ListReplaysResponse, ReplaysFacetsResponse, @@ -13,19 +12,17 @@ import { TraceId, UserId, } from "@maple/domain/http" -import { Effect, Layer, Option, Schema } from "effect" +import { Effect, Option, Schema } from "effect" import { CH } from "@maple/query-engine" import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService" -import { ReplayBlobStore, ReplayBlobStoreLive } from "@/platform/ReplayBlobStore" const decodeSessionId = Schema.decodeSync(SessionId) const decodeTraceId = Schema.decodeSync(TraceId) const decodeUserId = Schema.decodeSync(UserId) -const HttpSessionReplaysGroup = HttpApiBuilder.group(MapleApi, "sessionReplays", (handlers) => +export const HttpSessionReplaysLive = HttpApiBuilder.group(MapleApi, "sessionReplays", (handlers) => Effect.gen(function* () { const warehouse = yield* WarehouseQueryService - const blobs = yield* ReplayBlobStore return handlers .handle("listReplays", ({ payload }) => @@ -180,35 +177,6 @@ const HttpSessionReplaysGroup = HttpApiBuilder.group(MapleApi, "sessionReplays", }) }), ) - .handle("getReplayEvents", ({ payload }) => - Effect.gen(function* () { - const tenant = yield* CurrentTenant.Context - yield* Effect.annotateCurrentSpan({ - orgId: tenant.orgId, - "maple.session.id": payload.sessionId, - }) - const compiled = CH.compile( - CH.sessionReplayEventsQuery({ - startTime: payload.windowStart, - endTime: payload.windowEnd, - }), - { - orgId: tenant.orgId, - sessionId: payload.sessionId, - }, - ) - const rows = yield* warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "getReplayEvents", - }) - // Each chunk's `events` is the rrweb array JSON the player parses. - // Rows written since the R2 cutover carry an empty `events` and the - // payload comes from the blob store; older rows carry it inline and - // pass through untouched. - const chunks = yield* blobs.hydrate(tenant.orgId, payload.sessionId, rows) - return new GetReplayEventsResponse({ chunks }) - }), - ) .handle("replaysForTrace", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context @@ -299,10 +267,3 @@ const HttpSessionReplaysGroup = HttpApiBuilder.group(MapleApi, "sessionReplays", ) }), ) - -// Self-contained: the blob store resolves from the worker env (and degrades to -// a no-op hydrator when the R2 binding is absent), so it is provided here -// rather than pushed onto every caller that builds this group — the v2 route -// tests construct their own layer stack and would otherwise have to know about -// a storage detail of one handler. -export const HttpSessionReplaysLive = HttpSessionReplaysGroup.pipe(Layer.provide(ReplayBlobStoreLive)) diff --git a/apps/api/src/routes/v1/session-replay.schema.test.ts b/apps/api/src/routes/v1/session-replay.schema.test.ts index 1a4c05035..e39c85356 100644 --- a/apps/api/src/routes/v1/session-replay.schema.test.ts +++ b/apps/api/src/routes/v1/session-replay.schema.test.ts @@ -134,3 +134,4 @@ describe("SessionTraceSummary.spanCount (ClickHouse UInt64-as-string)", () => { expect(decodeSummary({ ...baseSummary, spanCount: Number("5") }).spanCount).toBe(5) }) }) + diff --git a/apps/api/src/routes/v2/phase1-resources.http.test.ts b/apps/api/src/routes/v2/phase1-resources.http.test.ts index a5817aeae..d1b7aeb91 100644 --- a/apps/api/src/routes/v2/phase1-resources.http.test.ts +++ b/apps/api/src/routes/v2/phase1-resources.http.test.ts @@ -41,6 +41,7 @@ import { UserId, } from "@maple/domain/http" import { MapleApiV2, encodePublicId } from "@maple/domain/http/v2" +import { WarehouseResponseLimitError } from "@maple/query-engine/execution" import { cleanupTestDbs, createTestDb, type TestDb } from "@/platform/test-pglite" import type { WarehouseQueryServiceShape } from "@/services/warehouse/WarehouseQueryService" import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService" @@ -299,6 +300,9 @@ const warehouseStub: WarehouseQueryServiceShape = { sqlQuery: () => Effect.succeed([]), rawSqlQuery: () => Effect.succeed([]), compiledQuery: (_tenant, compiled) => compiled.decodeRows([]).pipe(Effect.orDie), + // Replay payload reads go through the bounded variant (they carry an explicit + // response-byte ceiling), so the stub has to answer it too. + compiledQueryBounded: (_tenant, compiled) => compiled.decodeRows([]).pipe(Effect.orDie), compiledQueryFirst: () => Effect.succeed(Option.none()), ingest: () => Effect.void, asExecutor: () => { @@ -1084,4 +1088,273 @@ describe("v2 session_replays over HTTP", () => { } await harness.dispose() }) + + // A session's rrweb payload is unbounded by construction — ingest accepts up + // to 1 GiB and the p99 is ~594 MB. Reading it all at once buffered the whole + // payload into a 128 MB Worker; the abort was classified as a transient + // warehouse fault, retried twice more, and returned "service unavailable". + // These pin the shape that replaced it. + describe("bounded replay reads", () => { + const chunkRows = Array.from({ length: 40 }, (_, seq) => ({ + chunkSeq: seq, + timestamp: "2026-05-26 08:29:26.243", + durationMs: 5_000, + eventCount: 10, + byteSize: 100_000, + events: "[]", + isCheckpoint: seq % 20 === 0 ? 1 : 0, + })) + + /** Serves rows honouring the ChunkSeq predicates + LIMIT/OFFSET in the SQL. */ + const chunkWarehouse = (): WarehouseQueryServiceShape => { + const serve = (sql: string) => { + const from = Number(/ChunkSeq >= (\d+)/.exec(sql)?.[1] ?? 0) + const to = Number(/ChunkSeq <= (\d+)/.exec(sql)?.[1] ?? Number.MAX_SAFE_INTEGER) + const limit = Number(/LIMIT\s+(\d+)/i.exec(sql)?.[1] ?? chunkRows.length) + const offset = Number(/OFFSET\s+(\d+)/i.exec(sql)?.[1] ?? 0) + return chunkRows + .filter((row) => row.chunkSeq >= from && row.chunkSeq <= to) + .slice(offset, offset + limit) + } + return { + ...warehouseStub, + compiledQuery: (_tenant, compiled) => + compiled.decodeRows(serve(compiled.sql)).pipe(Effect.orDie), + compiledQueryBounded: (_tenant, compiled) => + compiled.decodeRows(serve(compiled.sql)).pipe(Effect.orDie), + } + } + + it("serves a manifest with no payloads, and the caps a client must size against", async () => { + const harness = makeHarness(chunkWarehouse()) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_manifest") + + const response = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/manifest`, + { token: key.secret }, + ) + expect(response.status).toBe(200) + expect(response.body.object).toBe("session_replay.manifest") + expect(response.body.chunk_count).toBe(40) + expect(response.body.total_byte_size).toBe(4_000_000) + expect(response.body.truncated).toBe(false) + // The whole point: the timeline arrives without the payload. + expect(response.body.chunks[0]).not.toHaveProperty("events") + // Playback anchors on the recording's own clock, not the receipt time. + expect(response.body.chunks[0].is_checkpoint).toBe(true) + // Echoed so a client never hardcodes a cap that can drift from ours. + expect(response.body.max_chunks_per_request).toBe(40) + await harness.dispose() + }) + + it("returns only the requested chunk range", async () => { + const harness = makeHarness(chunkWarehouse()) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_ranged") + + const response = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/events?from_chunk_seq=16&to_chunk_seq=31&limit=16`, + { token: key.secret }, + ) + expect(response.status).toBe(200) + expect(response.body.data).toHaveLength(16) + expect(response.body.data[0].chunk_seq).toBe(16) + expect(response.body.data[15].chunk_seq).toBe(31) + await harness.dispose() + }) + + it("clamps a range wider than the per-request cap instead of rejecting it", async () => { + // One chunk over the cap should still return data — only an oversized + // *payload* is refused, and that is the warehouse's call, not arithmetic. + const harness = makeHarness(chunkWarehouse()) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_clamped") + + const response = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/events?from_chunk_seq=0&to_chunk_seq=9999&limit=100`, + { token: key.secret }, + ) + expect(response.status).toBe(200) + expect(response.body.data.length).toBeLessThanOrEqual(40) + expect(response.body.data[0].chunk_seq).toBe(0) + await harness.dispose() + }) + + it("serves a session that never produced a checkpoint", async () => { + // The SDK's over-cap buffer guard drops the batch holding the opening + // snapshot, so some existing recordings have no checkpoint anywhere. + // The manifest must still describe them — the client anchors on chunk 0. + const legacyRows = chunkRows.map((row) => ({ ...row, isCheckpoint: 0 })) + const harness = makeHarness({ + ...warehouseStub, + compiledQuery: (_tenant, compiled) => compiled.decodeRows(legacyRows).pipe(Effect.orDie), + compiledQueryBounded: (_tenant, compiled) => + compiled.decodeRows(legacyRows).pipe(Effect.orDie), + }) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_legacy") + + const response = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/manifest`, + { token: key.secret }, + ) + expect(response.status).toBe(200) + expect(response.body.chunk_count).toBe(40) + expect(response.body.chunks.every((c: { is_checkpoint: boolean }) => !c.is_checkpoint)).toBe(true) + await harness.dispose() + }) + + it("coerces JSON-quoted 64-bit ints from the ClickHouse wire format", async () => { + // Backends that refuse `output_format_json_quote_64bit_integers=0` return + // UInt64s as strings. Schema.Number rejects those, which would surface as + // a bodyless 500 — hence the Number() coercion in both handlers. + const quotedRows = chunkRows.slice(0, 2).map((row) => ({ + ...row, + byteSize: String(row.byteSize) as unknown as number, + eventCount: String(row.eventCount) as unknown as number, + })) + const harness = makeHarness({ + ...warehouseStub, + compiledQuery: (_tenant, compiled) => compiled.decodeRows(quotedRows).pipe(Effect.orDie), + compiledQueryBounded: (_tenant, compiled) => + compiled.decodeRows(quotedRows).pipe(Effect.orDie), + }) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_quoted") + + const manifest = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/manifest`, + { token: key.secret }, + ) + expect(manifest.status).toBe(200) + expect(manifest.body.total_byte_size).toBe(200_000) + + const events = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/events?from_chunk_seq=0&to_chunk_seq=1`, + { token: key.secret }, + ) + expect(events.status).toBe(200) + expect(events.body.data[0].byte_size).toBe(100_000) + await harness.dispose() + }) + + it("keeps pagination honest when the caller asks for more than the cap", async () => { + // `limit=100` is inside the public 1–100 range but above the per-request + // chunk cap. Unclamped, the lookahead asks for 101 rows, gets the SQL + // cap of 41, and concludes 41 <= 100 means "no more pages" — a short + // page reported as complete, silently dropping the rest of the session. + // 60 chunks: more than one capped page, so a page that claims to be the + // last one is provably wrong. + const manyRows = Array.from({ length: 60 }, (_, seq) => ({ ...chunkRows[0]!, chunkSeq: seq })) + const harness = makeHarness({ + ...warehouseStub, + compiledQueryBounded: (_tenant, compiled) => { + const limit = Number(/LIMIT\s+(\d+)/i.exec(compiled.sql)?.[1] ?? manyRows.length) + const offset = Number(/OFFSET\s+(\d+)/i.exec(compiled.sql)?.[1] ?? 0) + return compiled.decodeRows(manyRows.slice(offset, offset + limit)).pipe(Effect.orDie) + }, + }) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_bigpage") + + const response = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/events?limit=100`, + { token: key.secret }, + ) + expect(response.status).toBe(200) + expect(response.body.data.length).toBeLessThanOrEqual(40) + // 60 chunks exist and the page is capped at 40, so there is a next page. + expect(response.body.has_more).toBe(true) + expect(response.body.next_cursor).not.toBeNull() + await harness.dispose() + }) + + it("refuses an over-budget range of blob-backed chunks before hydrating", async () => { + // The seam between this change and the R2 move. Once payloads live in + // the blob store the warehouse response is only an index — `events` is + // "" — so the `responseLimits` ceiling on that read measures almost + // nothing and would wave this through. `byteSize` is the uncompressed + // payload size and is right there in the index, so the range is refused + // without fetching a single object. + const hugeBlobRows = chunkRows.slice(0, 4).map((row) => ({ + ...row, + events: "", + byteSize: 5_000_000, + })) + const harness = makeHarness({ + ...warehouseStub, + compiledQueryBounded: (_tenant, compiled) => + compiled.decodeRows(hugeBlobRows).pipe(Effect.orDie), + }) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_blobs_toobig") + + const response = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/events?from_chunk_seq=0&to_chunk_seq=3`, + { token: key.secret }, + ) + expect(response.status).toBe(413) + expect(response.body.error.code).toBe("range_too_large") + await harness.dispose() + }) + + it("serves a blob-backed range that fits the budget", async () => { + const blobRows = chunkRows.slice(0, 4).map((row) => ({ ...row, events: "", byteSize: 100_000 })) + const harness = makeHarness({ + ...warehouseStub, + compiledQueryBounded: (_tenant, compiled) => compiled.decodeRows(blobRows).pipe(Effect.orDie), + }) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_blobs_ok") + + const response = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/events?from_chunk_seq=0&to_chunk_seq=3`, + { token: key.secret }, + ) + expect(response.status).toBe(200) + expect(response.body.data).toHaveLength(4) + // No R2 binding in tests, so hydration is a no-op passthrough — the + // point here is that the budget guard let the range through. + expect(response.body.data[0].byte_size).toBe(100_000) + await harness.dispose() + }) + + it("refuses an over-budget range with 413 range_too_large, not a 503", async () => { + // The regression that motivated all of this: the old failure claimed the + // database was unavailable and invited a retry that could only fail the + // same way. This one names the cause and says what to change. + const harness = makeHarness({ + ...warehouseStub, + compiledQueryBounded: () => + Effect.fail( + new WarehouseResponseLimitError({ kind: "bytes", message: "response too large" }), + ), + }) + const key = await harness.bootstrapKey() + const sessionId = encodePublicId("srep", "sess_toobig") + + const response = await harness.request( + "GET", + `/v2/session_replays/${sessionId}/events?from_chunk_seq=0&to_chunk_seq=39`, + { token: key.secret }, + ) + expect(response.status).toBe(413) + expect(response.body.error.code).toBe("range_too_large") + expect(response.body.error.param).toBe("to_chunk_seq") + // The message survives the public boundary verbatim — it carries no + // database diagnostics, and it is the only actionable thing here. + expect(response.body.error.message).toContain("narrower chunk range") + await harness.dispose() + }) + }) }) diff --git a/apps/api/src/routes/v2/session-replays.http.ts b/apps/api/src/routes/v2/session-replays.http.ts index e1baba5c0..58c445616 100644 --- a/apps/api/src/routes/v2/session-replays.http.ts +++ b/apps/api/src/routes/v2/session-replays.http.ts @@ -1,19 +1,28 @@ import { HttpApiBuilder } from "effect/unstable/httpapi" import { CurrentTenant, SessionId, TraceId } from "@maple/domain/http" import { + MAX_REPLAY_CHUNKS_PER_REQUEST, + MAX_REPLAY_EVENTS_RESPONSE_BYTES, + MAX_REPLAY_MANIFEST_CHUNKS, + LIST_LIMIT_DEFAULT, MapleApiV2, dependencyUnavailable, invalidRequest, paginateArray, paginateOffsetQuery, + payloadTooLarge, resourceNotFound, timestamp, } from "@maple/domain/http/v2" import type { Timestamp } from "@maple/domain/http/v2" +import type { WarehouseError } from "@maple/domain/http" +import type { WarehouseResponseLimitError } from "@maple/query-engine/execution" import type { V2SessionReplay, V2SessionReplayChunk, + V2SessionReplayChunkMeta, V2SessionReplayListItem, + V2SessionReplayManifest, V2SessionReplayRef, V2SessionTranscriptEvent, } from "@maple/domain/http/v2" @@ -29,6 +38,45 @@ const decodeTraceId = Schema.decodeSync(TraceId) /** Warehouse errors → the proper v2 envelope (400/429/502/503 per tag). */ const mapWarehouseError = warehouseToV2("session_replay_query") +/** + * Refuse a chunk range whose payload would blow the response budget, before a + * byte of it is fetched. + * + * Once payloads live in R2 the warehouse response is only an index — `Events` is + * empty — so the `responseLimits` ceiling on that read stops measuring anything + * that matters, and the memory cost moves to hydration. `ByteSize` is the + * uncompressed payload size and is right there in the index, so the range can be + * rejected without touching the blob store at all. Cheaper and more honest than + * discovering the problem mid-hydration. + */ +const assertRangeFitsBudget = (rows: ReadonlyArray<{ readonly byteSize: number }>) => { + const total = rows.reduce((sum, row) => sum + Number(row.byteSize), 0) + return total <= MAX_REPLAY_EVENTS_RESPONSE_BYTES + ? Effect.void + : Effect.fail( + payloadTooLarge( + "That part of the recording is too large to load in one request. Request a narrower chunk range.", + "to_chunk_seq", + ), + ) +} + +/** + * Warehouse errors → the v2 envelope, plus the bounded-read refusal. + * + * `range_too_large` keeps its message verbatim across the public boundary: it + * carries no database diagnostics — only the range asked for — and unlike the + * warehouse faults it is entirely actionable, so redacting it would strip the + * one useful thing it says. + */ +const mapReplayReadError = (error: WarehouseError | WarehouseResponseLimitError) => + error._tag === "@maple/query-engine/execution/WarehouseResponseLimitError" + ? payloadTooLarge( + "That part of the recording is too large to load in one request. Request a narrower chunk range.", + "to_chunk_seq", + ) + : mapWarehouseError(error) + /** ISO-8601 → Tinybird `YYYY-MM-DD HH:mm:ss` (UTC), validated. */ const toTinybird = (value: string, param: string) => { const ms = Date.parse(value) @@ -211,43 +259,145 @@ const HttpV2SessionReplaysGroup = HttpApiBuilder.group(MapleApiV2, "sessionRepla return replay }), ) - .handle("events", ({ params, query }) => + .handle("manifest", ({ params, query }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context const windowStart = yield* optTinybird(query.window_start, "window_start") const windowEnd = yield* optTinybird(query.window_end, "window_end") const compiled = CH.compile( - CH.sessionReplayEventsQuery({ startTime: windowStart, endTime: windowEnd }), + CH.sessionReplayChunkIndexQuery({ startTime: windowStart, endTime: windowEnd }), { orgId: tenant.orgId, sessionId: params.id }, ) - const stored = yield* warehouse - .compiledQuery(tenant, compiled, { profile: "list", context: "v2GetReplayEvents" }) + // `discovery` is enough — this never reads the `Events` column, and + // with payloads in R2 there is nothing to hydrate here either: the + // manifest is a pure index read whatever the storage backend. + const rows = yield* warehouse + .compiledQuery(tenant, compiled, { + profile: "discovery", + context: "v2GetReplayManifest", + }) .pipe(Effect.mapError(mapWarehouseError)) - if (stored.length === 0) { + if (rows.length === 0) { yield* requireSession(tenant, params.id, windowStart, windowEnd) } - // Paginate before hydrating: the warehouse query has no LIMIT, so - // this is the only place a page's payload cost gets bounded. Doing - // it the other way round would fetch every chunk of the session - // from the blob store to then throw all but `limit` away. - const page = yield* paginateArray(stored, query) - // Blob-backed rows (empty `events`) get their payload from R2; - // pre-cutover rows already carry it inline. - const rows = yield* blobs.hydrate(tenant.orgId, params.id, page.data) - const chunks = rows.map( + const truncated = rows.length > MAX_REPLAY_MANIFEST_CHUNKS + // ClickHouse JSON-quotes 64-bit ints on backends that refuse the + // unquote setting; coerce before Schema.Number validates. + const chunks = rows.slice(0, MAX_REPLAY_MANIFEST_CHUNKS).map( (row) => ({ - object: "session_replay.event_chunk" as const, - chunk_seq: row.chunkSeq, + chunk_seq: Number(row.chunkSeq), timestamp: chToIso(row.timestamp), - duration_ms: row.durationMs, - event_count: row.eventCount, - byte_size: row.byteSize, + duration_ms: Number(row.durationMs), + event_count: Number(row.eventCount), + byte_size: Number(row.byteSize), is_checkpoint: Number(row.isCheckpoint) !== 0, - events: row.events, - }) satisfies V2SessionReplayChunk, + }) satisfies V2SessionReplayChunkMeta, ) - return { object: "list" as const, ...page, data: chunks } + yield* Effect.annotateCurrentSpan({ + "maple.replay.chunk_count": chunks.length, + "maple.replay.truncated": truncated, + }) + return { + object: "session_replay.manifest" as const, + session_id: params.id, + chunks, + chunk_count: chunks.length, + total_byte_size: chunks.reduce((sum, chunk) => sum + chunk.byte_size, 0), + max_chunks_per_request: MAX_REPLAY_CHUNKS_PER_REQUEST, + max_bytes_per_request: MAX_REPLAY_EVENTS_RESPONSE_BYTES, + truncated, + } satisfies V2SessionReplayManifest + }), + ) + .handle("events", ({ params, query }) => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + const windowStart = yield* optTinybird(query.window_start, "window_start") + const windowEnd = yield* optTinybird(query.window_end, "window_end") + // Clamp rather than reject: asking for a chunk past the cap should + // still return data. Only a range whose *payload* blows the byte + // budget is refused, below. + const fromChunkSeq = query.from_chunk_seq + const toChunkSeq = + fromChunkSeq === undefined + ? query.to_chunk_seq + : Math.min( + query.to_chunk_seq ?? Number.POSITIVE_INFINITY, + fromChunkSeq + MAX_REPLAY_CHUNKS_PER_REQUEST - 1, + ) + yield* Effect.annotateCurrentSpan({ + "maple.session.id": params.id, + "maple.replay.chunk_from": fromChunkSeq ?? -1, + "maple.replay.chunk_to": toChunkSeq ?? -1, + }) + // Pagination is pushed into SQL. It used to be applied to an + // already-materialized array, so every page paid for the whole + // session's payload — the read this endpoint could never survive. + // + // The page size is clamped to the same cap the SQL enforces. Left + // unclamped, `limit=100` would ask the lookahead for 101 rows, get + // the SQL cap of 41, and conclude 41 <= 100 means "no more pages" — + // silently returning a short page with `has_more: false` and + // dropping the rest of the recording. + const pageQuery = { + ...query, + limit: Math.min(query.limit ?? LIST_LIMIT_DEFAULT, MAX_REPLAY_CHUNKS_PER_REQUEST), + } + const page = yield* paginateOffsetQuery(pageQuery, ({ limit, offset }) => { + const compiled = CH.compile( + CH.sessionReplayEventsQuery({ + startTime: windowStart, + endTime: windowEnd, + fromChunkSeq, + toChunkSeq: toChunkSeq === Number.POSITIVE_INFINITY ? undefined : toChunkSeq, + limit: Math.min(limit, MAX_REPLAY_CHUNKS_PER_REQUEST + 1), + offset, + }), + { orgId: tenant.orgId, sessionId: params.id }, + ) + // Two ceilings, because there are two places the payload can come + // from. `responseLimits` bounds what the warehouse hands back — + // the only guard for pre-cutover rows, which still carry `events` + // inline. Blob-backed rows make that response nearly empty, so it + // would pass anything; `assertRangeFitsBudget` below bounds the + // hydration instead, using sizes the index already knows. + return warehouse + .compiledQueryBounded(tenant, compiled, { + profile: "list", + context: "v2GetReplayEvents", + responseLimits: { + maxRows: MAX_REPLAY_CHUNKS_PER_REQUEST + 1, + maxBytes: MAX_REPLAY_EVENTS_RESPONSE_BYTES, + }, + }) + .pipe( + Effect.mapError(mapReplayReadError), + Effect.tap((rows) => + rows.length === 0 && offset === 0 + ? requireSession(tenant, params.id, windowStart, windowEnd) + : Effect.void, + ), + Effect.tap(assertRangeFitsBudget), + // Blob-backed rows (empty `events`) get their payload from + // R2; pre-cutover rows already carry it inline. + Effect.flatMap((rows) => blobs.hydrate(tenant.orgId, params.id, rows)), + Effect.map( + (rows): ReadonlyArray => + rows.map((row) => ({ + object: "session_replay.event_chunk" as const, + chunk_seq: Number(row.chunkSeq), + timestamp: chToIso(row.timestamp), + duration_ms: Number(row.durationMs), + event_count: Number(row.eventCount), + byte_size: Number(row.byteSize), + is_checkpoint: Number(row.isCheckpoint) !== 0, + events: row.events, + })), + ), + ) + }) + return { object: "list" as const, ...page } }), ) .handle("transcript", ({ params, query }) => diff --git a/apps/api/src/routes/v2/v2-test-support.ts b/apps/api/src/routes/v2/v2-test-support.ts index 00a36ebe4..65ddfc9dc 100644 --- a/apps/api/src/routes/v2/v2-test-support.ts +++ b/apps/api/src/routes/v2/v2-test-support.ts @@ -148,6 +148,7 @@ export const WarehouseServiceStubLayer = Layer.succeed(WarehouseQueryService, { crossOrgQuery: die, rawSqlQuery: die, compiledQuery: die, + compiledQueryBounded: die, compiledQueryWithCapabilities: die, compiledQueryFirst: die, ingest: die, diff --git a/apps/api/src/services/warehouse/WarehouseQueryService.ts b/apps/api/src/services/warehouse/WarehouseQueryService.ts index ee862d98e..18f5e8748 100644 --- a/apps/api/src/services/warehouse/WarehouseQueryService.ts +++ b/apps/api/src/services/warehouse/WarehouseQueryService.ts @@ -431,6 +431,20 @@ export class WarehouseQueryService extends Context.Service< options?: SqlQueryOptions, ) => this.use((service) => service.compiledQuery(tenant, compiled, options)) + /** + * `compiledQuery` with a hard ceiling on the response we'll materialize. + * Fails with `WarehouseResponseLimitError` past it rather than buffering the + * rest into the Worker heap — for queries whose result size follows user data + * rather than the query shape (session-replay rrweb payloads). + */ + static readonly compiledQueryBounded = ( + tenant: TenantContext, + compiled: CompiledQuery, + options: SqlQueryOptions & { + readonly responseLimits: { readonly maxRows: number; readonly maxBytes: number } + }, + ) => this.use((service) => service.compiledQueryBounded(tenant, compiled, options)) + static readonly compiledQueryFirst = ( tenant: TenantContext, compiled: CompiledQuery, diff --git a/apps/web/src/api/warehouse/effect-utils.ts b/apps/web/src/api/warehouse/effect-utils.ts index a911a223c..fc38533e4 100644 --- a/apps/web/src/api/warehouse/effect-utils.ts +++ b/apps/web/src/api/warehouse/effect-utils.ts @@ -8,7 +8,8 @@ import { } from "@maple/query-engine" import { Effect, Schema } from "effect" import { MapleApiAtomClient } from "@/lib/services/common/atom-client" -import { mapleApiClientLayer } from "@/lib/registry" +import { MapleApiV2AtomClient } from "@/lib/services/common/v2-atom-client" +import { mapleApiClientLayer, mapleApiV2ClientLayer } from "@/lib/registry" export const WarehouseDateTimeString = TinybirdDateTime @@ -110,6 +111,35 @@ export function runWarehouseQuery( ) } +/** + * `runWarehouseQuery` against the v2 client. + * + * Same span + error normalization, different client layer and a wider input + * error type: the v2 endpoints fail with the public envelope union + * (`V2InvalidRequestError`, `V2PayloadTooLargeError`, …) rather than the v1 + * warehouse tags, and all of those collapse into `WarehouseQueryError` here so + * callers keep one error shape. + */ +export function runWarehouseQueryV2( + operation: string, + execute: () => Effect.Effect, +): Effect.Effect { + return Effect.suspend(execute).pipe( + Effect.withSpan(operation), + Effect.provide(mapleApiV2ClientLayer), + Effect.mapError((cause) => { + if (isTaggedBackendError(cause)) { + return cause + } + return new WarehouseQueryError({ + operation, + message: toMessage(cause, `Warehouse query failed for ${operation}`), + cause, + }) + }), + ) +} + export function invalidWarehouseInput( operation: string, message: string, diff --git a/apps/web/src/api/warehouse/replays.ts b/apps/web/src/api/warehouse/replays.ts index eda4eff50..1479a37bb 100644 --- a/apps/web/src/api/warehouse/replays.ts +++ b/apps/web/src/api/warehouse/replays.ts @@ -1,6 +1,5 @@ import { Clock, Effect, Schema } from "effect" import { - GetReplayEventsRequest, GetReplayRequest, ListReplaysRequest, ReplaysFacetsRequest, @@ -11,7 +10,13 @@ import { TraceId, } from "@maple/domain/http" import { MapleApiAtomClient } from "@/lib/services/common/atom-client" -import { WarehouseDateTimeString, decodeInput, runWarehouseQuery } from "@/api/warehouse/effect-utils" +import { MapleApiV2AtomClient } from "@/lib/services/common/v2-atom-client" +import { + WarehouseDateTimeString, + decodeInput, + runWarehouseQuery, + runWarehouseQueryV2, +} from "@/api/warehouse/effect-utils" import { formatWarehouseDateTime } from "@maple/query-engine" // --------------------------------------------------------------------------- @@ -175,13 +180,66 @@ export const getReplay = Effect.fn("SessionReplays.getReplay")(function* ({ }) // --------------------------------------------------------------------------- -// Session event chunks (rrweb payloads inline, from ClickHouse, ordered) +// Session event chunks — manifest first, then bounded ranges (v2) +// +// A session's rrweb payload is unbounded by construction: ingest accepts up to +// 1 GiB and the p99 session is ~594 MB. Fetching all of it in one response +// buffered the whole thing into a 128 MB Worker and surfaced as a 503 that +// blamed the database. So the player pulls the cheap manifest (timeline and +// sizes, no payloads), then pulls payloads a range at a time. +// +// These are the only replay reads on v2 — the v1 group has no payload endpoint +// precisely so the unbounded read cannot come back. // --------------------------------------------------------------------------- +/** Warehouse `YYYY-MM-DD HH:mm:ss` → the ISO-8601 the v2 query params take. */ +const toIsoWindow = (value: string | undefined) => + value === undefined ? undefined : new Date(`${value.replace(" ", "T")}Z`).toISOString() + +const GetReplayManifestInput = Schema.Struct({ + sessionId: SessionId, + windowStart: Schema.optional(WarehouseDateTimeString), + windowEnd: Schema.optional(WarehouseDateTimeString), +}) +export type GetReplayManifestInput = (typeof GetReplayManifestInput)["Encoded"] + +export const getReplayManifest = Effect.fn("SessionReplays.getReplayManifest")(function* ({ + data, +}: { + data: GetReplayManifestInput +}) { + const input = yield* decodeInput(GetReplayManifestInput, data ?? {}, "getReplayManifest") + return yield* runWarehouseQueryV2("getReplayManifest", () => + Effect.gen(function* () { + const client = yield* MapleApiV2AtomClient + // The `srep_…` public-ID codec lives in the client's param encoder, so + // the internal SessionId goes in as-is. + return yield* client.sessionReplays.manifest({ + params: { id: input.sessionId }, + query: { + ...(toIsoWindow(input.windowStart) !== undefined + ? { window_start: toIsoWindow(input.windowStart)! } + : {}), + ...(toIsoWindow(input.windowEnd) !== undefined + ? { window_end: toIsoWindow(input.windowEnd)! } + : {}), + }, + }) + }), + ) +}) + const GetReplayEventsInput = Schema.Struct({ sessionId: SessionId, windowStart: Schema.optional(WarehouseDateTimeString), windowEnd: Schema.optional(WarehouseDateTimeString), + /** + * Inclusive chunk range from the manifest. Required: an optional range would + * leave the unbounded read reachable from the client, and something would + * eventually reach it. + */ + fromChunkSeq: Schema.Number, + toChunkSeq: Schema.Number, }) export type GetReplayEventsInput = (typeof GetReplayEventsInput)["Encoded"] @@ -191,19 +249,29 @@ export const getReplayEvents = Effect.fn("SessionReplays.getReplayEvents")(funct data: GetReplayEventsInput }) { const input = yield* decodeInput(GetReplayEventsInput, data ?? {}, "getReplayEvents") - const result = yield* runWarehouseQuery("getReplayEvents", () => + const result = yield* runWarehouseQueryV2("getReplayEvents", () => Effect.gen(function* () { - const client = yield* MapleApiAtomClient - return yield* client.sessionReplays.getReplayEvents({ - payload: new GetReplayEventsRequest({ - sessionId: input.sessionId, - windowStart: input.windowStart, - windowEnd: input.windowEnd, - }), + const client = yield* MapleApiV2AtomClient + return yield* client.sessionReplays.events({ + params: { id: input.sessionId }, + query: { + from_chunk_seq: input.fromChunkSeq, + to_chunk_seq: input.toChunkSeq, + // One page covers the whole range: ranges are sized by the caller + // against the server's advertised cap, so paging within one would + // only add round-trips. + limit: Math.max(1, input.toChunkSeq - input.fromChunkSeq + 1), + ...(toIsoWindow(input.windowStart) !== undefined + ? { window_start: toIsoWindow(input.windowStart)! } + : {}), + ...(toIsoWindow(input.windowEnd) !== undefined + ? { window_end: toIsoWindow(input.windowEnd)! } + : {}), + }, }) }), ) - return { chunks: result.chunks } + return { chunks: result.data } }) // --------------------------------------------------------------------------- diff --git a/apps/web/src/components/replays/replay-player-context.tsx b/apps/web/src/components/replays/replay-player-context.tsx index 51980b746..336d2d758 100644 --- a/apps/web/src/components/replays/replay-player-context.tsx +++ b/apps/web/src/components/replays/replay-player-context.tsx @@ -2,9 +2,11 @@ import * as React from "react" import { Replayer } from "@rrweb/replay" import { EventType, IncrementalSource, MouseInteractions, ReplayerEvents } from "@rrweb/types" import { Result, useAtomValue } from "@/lib/effect-atom" -import { getReplayEventsResultAtom } from "@/lib/services/atoms/warehouse-query-atoms" +import { getReplayManifestResultAtom } from "@/lib/services/atoms/warehouse-query-atoms" import { normalizeEvents } from "./replay-events" import type { ReplayPartitionWindow } from "./replay-format" +import { manifestDurationMs, type ReplayChunkMeta } from "./replay-range" +import { useReplayChunkLoader } from "./use-replay-chunk-loader" import { buildTimeline, type InactiveInterval, type Timeline } from "./replay-timeline" // --------------------------------------------------------------------------- @@ -17,25 +19,7 @@ import { buildTimeline, type InactiveInterval, type Timeline } from "./replay-ti // `` and `` are both consumers. // --------------------------------------------------------------------------- -interface ReplayChunk { - readonly chunkSeq: number - readonly events: string -} - -/** Sort chunks by sequence, parse each chunk's inline rrweb event JSON, and normalize. */ -function decodeChunks(chunks: ReadonlyArray): unknown[] { - const ordered = [...chunks].sort((a, b) => a.chunkSeq - b.chunkSeq) - const all: unknown[] = [] - for (const chunk of ordered) { - try { - const parsed: unknown = JSON.parse(chunk.events) - if (Array.isArray(parsed)) all.push(...parsed) - } catch { - // Skip a malformed chunk rather than failing the whole replay. - } - } - return normalizeEvents(all) -} +const EMPTY_CHUNKS: ReadonlyArray = [] /** A user interaction worth flagging on the scrubber. */ export type ActionKind = "click" | "input" | "scroll" | "nav" @@ -295,10 +279,25 @@ export function ReplayPlayerProvider({ /** Whether the session is still open (`status === "active"`), i.e. chunks may still arrive. */ sessionActive?: boolean }) { - // Chunks carry their rrweb events inline (read straight from ClickHouse); parse - // + concatenate them in order. Memoized on the (referentially stable) result so - // deriveMeta/engine memos hold across the player's frequent re-renders. - const eventsResult = useAtomValue(getReplayEventsResultAtom({ data: { sessionId, ...window } })) + // The manifest — every chunk's position and size, no payloads. Cheap on any + // session, and the prerequisite for deciding which payload ranges to fetch. + const manifestResult = useAtomValue(getReplayManifestResultAtom({ data: { sessionId, ...window } })) + + const manifestChunks = React.useMemo>( + () => + Result.builder(manifestResult) + .onSuccess((result) => result.chunks as ReadonlyArray) + .orElse(() => EMPTY_CHUNKS), + [manifestResult], + ) + + const loader = useReplayChunkLoader({ + sessionId, + window, + chunks: manifestChunks, + // The preview route supplies events directly and must not hit the network. + enabled: previewEvents === undefined, + }) const { status, error, events } = React.useMemo<{ status: ReplayLoadStatus @@ -320,24 +319,45 @@ export function ReplayPlayerProvider({ events: EMPTY_EVENTS, } } - return Result.builder(eventsResult) + if (loader.loadError !== null) { + return { status: "error" as const, error: loader.loadError, events: EMPTY_EVENTS } + } + return Result.builder(manifestResult) .onInitial(() => ({ status: "loading" as const, error: null, events: EMPTY_EVENTS })) .onError((e) => ({ status: "error" as const, error: e, events: EMPTY_EVENTS })) .onSuccess((result) => { - const chunks = result.chunks as ReadonlyArray - const decoded = decodeChunks(chunks) - if (decoded.length >= 2) { - return { status: "ready" as const, error: null, events: decoded } + // Emptiness is now decided from the manifest — no longer requiring a + // download of the whole session just to discover there's nothing in it. + const chunkCount = result.chunks.length + if (chunkCount === 0) { + return { + status: classifyUnplayable({ recorded, chunkCount, sessionActive }), + error: null, + events: EMPTY_EVENTS, + } } - // Nothing playable — say which kind of nothing. - return { - status: classifyUnplayable({ recorded, chunkCount: chunks.length, sessionActive }), - error: null, - events: EMPTY_EVENTS, + if (loader.seedEvents.length >= 2) { + return { status: "ready" as const, error: null, events: [...loader.seedEvents] } } + // Manifest says there are chunks; the opening window is still in + // flight, or every chunk in it was unparseable. + return loader.seedEvents.length > 0 + ? { + status: classifyUnplayable({ recorded, chunkCount, sessionActive }), + error: null, + events: EMPTY_EVENTS, + } + : { status: "loading" as const, error: null, events: EMPTY_EVENTS } }) .orElse(() => ({ status: "loading" as const, error: null, events: EMPTY_EVENTS })) - }, [eventsResult, previewEvents, recorded, sessionActive]) + }, [manifestResult, previewEvents, recorded, sessionActive, loader.seedEvents, loader.loadError]) + + // Playable length from the manifest — known before any payload is fetched. + const manifestTotalMs = React.useMemo(() => manifestDurationMs(manifestChunks), [manifestChunks]) + const seekTargetMs = loader.seekTargetMs + const { requestSeek, reportProgress } = loader + /** How many trailing events have been handed to the engine already. */ + const fedCountRef = React.useRef(0) const figureRef = React.useRef(null) const surfaceRef = React.useRef(null) @@ -445,6 +465,7 @@ export function ReplayPlayerProvider({ setCurrentMs(0) setFinished(false) setIsPlaying(false) + fedCountRef.current = 0 // rrweb's own transport events are unreliable in @rrweb/replay (Start/Resume // often don't fire); play/pause state is driven from our handlers. We still @@ -469,8 +490,49 @@ export function ReplayPlayerProvider({ replayerRef.current = null mount.innerHTML = "" } + // Keyed on the SEED, and on nothing else that moves. + // + // `events` here is the loader's seed, which only changes on first load or a + // seek that needs a new anchor. Anything else in this list — the manifest + // total, a seek target — rebuilds the engine whenever it moves, and for a + // live session the manifest moves constantly: the iframe flashes and the + // playhead snaps back to 0 every few seconds, which reads as "playback is + // frozen". Trailing events and the total are applied by the effects below. }, [events, status, applyScale]) + // The manifest knows the full length before the payload is loaded, so the + // scrubber shows the real duration from the first frame. `getMetaData()` would + // report only the loaded span and visibly stretch as chunks stream in — but it + // is still the right answer for the preview route, which has no manifest. + React.useEffect(() => { + if (manifestTotalMs > 0) setTotalMs(manifestTotalMs) + }, [manifestTotalMs]) + + // Resume where the user asked after a seek-driven rebuild. Separate from the + // mount effect so that a new seek target never reconstructs the engine. + React.useEffect(() => { + const replayer = replayerRef.current + if (!replayer || status !== "ready" || seekTargetMs === null) return + replayer.pause(seekTargetMs) + setCurrentMs(seekTargetMs) + }, [seekTargetMs, status]) + + // Feed newly-arrived events into the live engine. + // + // Only ever forward: every trailing event postdates the seed, so rrweb + // binary-inserts it and queues it on its own timer. An event that predates + // the baseline would instead be applied synchronously and out of DOM order — + // which is why a backward seek rebuilds (see `requestSeek`) rather than + // arriving here. + React.useEffect(() => { + const replayer = replayerRef.current + if (!replayer || status !== "ready") return + const pending = loader.trailingEvents.slice(fedCountRef.current) + if (pending.length === 0) return + for (const event of pending) replayer.addEvent(event as never) + fedCountRef.current = loader.trailingEvents.length + }, [loader.trailingEvents, status]) + // Recompute scale when entering/leaving fullscreen. React.useEffect(() => { applyScale() @@ -500,12 +562,34 @@ export function ReplayPlayerProvider({ lastJumpedEnd = -1 setCurrentMs(Math.min(Math.max(0, cur), totalMs)) } + // Keep the loader ahead of the playhead — this is what turns the + // opening window into continuous playback. + reportProgress(cur) } raf = requestAnimationFrame(tick) } raf = requestAnimationFrame(tick) return () => cancelAnimationFrame(raf) - }, [isPlaying, totalMs, inactiveIntervals, skipInactive]) + }, [isPlaying, totalMs, inactiveIntervals, skipInactive, reportProgress]) + + // Hold playback when the playhead outruns the loaded events, and release it + // when the next range lands. Without this the engine plays into an empty + // timeline and reports "finished" halfway through the recording. + // Gated on an actual stall having happened: an unconditional `play()` on every + // idle render would re-issue playback the user never paused, and double up on + // the transport controls' own play calls. + const stalledRef = React.useRef(false) + React.useEffect(() => { + const replayer = replayerRef.current + if (!replayer || !isPlaying) return + if (loader.bufferState === "buffering") { + stalledRef.current = true + replayer.pause() + } else if (loader.bufferState === "idle" && stalledRef.current) { + stalledRef.current = false + replayer.play(engineTimeMs(replayer)) + } + }, [loader.bufferState, isPlaying]) const togglePlay = React.useCallback(() => { const replayer = replayerRef.current @@ -551,9 +635,13 @@ export function ReplayPlayerProvider({ const clamped = Math.max(0, Math.min(pending.timeline.toReal(pending.displayMs), pending.totalMs)) setCurrentMs(clamped) if (clamped < pending.totalMs) setFinished(false) + // Seeking outside the loaded span rebuilds the engine from the nearest + // preceding checkpoint (the only thing rrweb can start from). The rebuild + // resumes at this offset, so don't also drive the doomed current engine. + if (requestSeek(clamped)) return if (pending.isPlaying && clamped < pending.totalMs) replayer.play(clamped) else replayer.pause(clamped) - }, []) + }, [requestSeek]) const seekDisplay = React.useCallback( (displayMs: number) => { diff --git a/apps/web/src/components/replays/replay-progressive-load.test.ts b/apps/web/src/components/replays/replay-progressive-load.test.ts new file mode 100644 index 000000000..0799c30ae --- /dev/null +++ b/apps/web/src/components/replays/replay-progressive-load.test.ts @@ -0,0 +1,202 @@ +import { describe, expect, it } from "vitest" +import { + INITIAL_WINDOW_BYTES, + MAX_CHUNKS_PER_RANGE, + type ReplayChunkMeta, + type ReplayRange, + checkpointAtOrBefore, + chunkAtOffset, + initialRanges, + planRanges, + rangeContaining, + rangeKey, + rangesCovering, +} from "./replay-range" + +// --------------------------------------------------------------------------- +// Progressive-load decision logic. +// +// The loader hook itself is a thin shell over these decisions plus React state; +// what actually has to be right is *which ranges get fetched, in what order, and +// when a seek forces the engine to be rebuilt*. Those are exercised here against +// a simulated playthrough, because the browser pane this would otherwise be +// verified in never fires requestAnimationFrame (it renders hidden), so wall-clock +// playback can't be driven there. +// --------------------------------------------------------------------------- + +const START_MS = 1_764_150_000_000 +const CHUNK_MS = 5_000 + +const chunk = (seq: number, overrides: Partial = {}): ReplayChunkMeta => ({ + chunk_seq: seq, + timestamp: new Date(START_MS + seq * CHUNK_MS).toISOString(), + duration_ms: CHUNK_MS, + event_count: 200, + byte_size: 100_000, + is_checkpoint: seq % 20 === 0, + ...overrides, +}) + +/** A 100-chunk session: ~8 minutes, 10 MB of payload, checkpoints every 20. */ +const manifest = Array.from({ length: 100 }, (_, seq) => chunk(seq)) +const plan = planRanges(manifest) + +/** + * Replay a playthrough the way the loader does: start from the initial window, + * then whenever the playhead comes within `PREFETCH_AHEAD` of the loaded end, + * pull the next grid range. + */ +const simulate = (chunks: ReadonlyArray, throughMs: number) => { + const fetched: Array = [...initialRanges(chunks, planRanges(chunks))] + const loadedThrough = () => + Math.max(...fetched.map((r) => r.toChunkSeq), -1) + const lastSeq = chunks[chunks.length - 1]!.chunk_seq + let playhead = 0 + while (playhead <= throughMs) { + const loadedUntilMs = (loadedThrough() + 1) * CHUNK_MS + if (playhead + 20_000 >= loadedUntilMs && loadedThrough() < lastSeq) { + const next = rangeContaining(planRanges(chunks), loadedThrough() + 1) + if (next && !fetched.some((r) => rangeKey(r) === rangeKey(next))) fetched.push(next) + } + playhead += 250 // sample far faster than chunk cadence, like the rAF tick + } + return fetched +} + +describe("opening window", () => { + it("starts on a couple of MB, not the whole session", () => { + const ranges = initialRanges(manifest, plan) + const bytes = ranges.length * MAX_CHUNKS_PER_RANGE * 100_000 + // The session is 10 MB. Before this change the player fetched all of it in + // one response, which is what made long sessions fail. + expect(bytes).toBeLessThan(4 * INITIAL_WINDOW_BYTES) + expect(ranges.length).toBeLessThanOrEqual(2) + }) + + it("is independent of session length", () => { + const long = Array.from({ length: 2000 }, (_, seq) => chunk(seq)) + expect(initialRanges(long, planRanges(long))).toEqual(initialRanges(manifest, plan)) + }) +}) + +describe("playing forward", () => { + it("pulls each range exactly once", () => { + const fetched = simulate(manifest, 100 * CHUNK_MS) + const keys = fetched.map(rangeKey) + expect(new Set(keys).size).toBe(keys.length) + }) + + it("stays ahead of the playhead without loading the whole session at once", () => { + // A third of the way in, the tail must still be unfetched — otherwise this + // is just the old all-at-once read wearing a different shape. + const fetched = simulate(manifest, 33 * CHUNK_MS) + const loadedThrough = Math.max(...fetched.map((r) => r.toChunkSeq)) + expect(loadedThrough).toBeGreaterThanOrEqual(33) + expect(loadedThrough).toBeLessThan(99) + }) + + it("covers the session end to end by the time playback finishes", () => { + const fetched = simulate(manifest, 100 * CHUNK_MS) + const covered = new Set() + for (const r of fetched) { + for (let s = r.fromChunkSeq; s <= r.toChunkSeq; s++) covered.add(s) + } + for (const c of manifest) expect(covered.has(c.chunk_seq)).toBe(true) + }) + + it("does not refetch a range the playhead scrubs back over", () => { + const forward = simulate(manifest, 40 * CHUNK_MS).map(rangeKey) + // Grid alignment means an earlier position maps to an already-fetched key. + const back = rangeContaining(plan, chunkAtOffset(manifest, 12 * CHUNK_MS)!.chunk_seq)! + expect(forward).toContain(rangeKey(back)) + }) +}) + +describe("seeking", () => { + const loadedAfterOpening = () => { + const covered = new Set() + for (const r of initialRanges(manifest, plan)) { + for (let s = r.fromChunkSeq; s <= r.toChunkSeq; s++) covered.add(s) + } + return covered + } + + it("needs a rebuild when the target is outside the loaded span", () => { + const covered = loadedAfterOpening() + const target = chunkAtOffset(manifest, 90 * CHUNK_MS)! + expect(covered.has(target.chunk_seq)).toBe(false) + }) + + it("rebuilds from a checkpoint, never from the bare target chunk", () => { + // rrweb can only start from a full DOM snapshot; seeding on a mid-stream + // chunk would replay mutations against a DOM that was never built. + const target = chunkAtOffset(manifest, 90 * CHUNK_MS)! + const anchor = checkpointAtOrBefore(manifest, target.chunk_seq)! + expect(anchor.is_checkpoint).toBe(true) + expect(anchor.chunk_seq).toBeLessThanOrEqual(target.chunk_seq) + expect(anchor.chunk_seq).toBe(80) + }) + + it("loads a fresh window from the anchor, covering the target", () => { + const target = chunkAtOffset(manifest, 90 * CHUNK_MS)! + const anchor = checkpointAtOrBefore(manifest, target.chunk_seq)! + const ranges = rangesCovering(manifest, plan, anchor.chunk_seq, INITIAL_WINDOW_BYTES) + const covered = new Set() + for (const r of ranges) { + for (let s = r.fromChunkSeq; s <= r.toChunkSeq; s++) covered.add(s) + } + expect(covered.has(anchor.chunk_seq)).toBe(true) + expect(covered.has(target.chunk_seq)).toBe(true) + }) + + it("does not rebuild for a target already loaded", () => { + const covered = loadedAfterOpening() + const target = chunkAtOffset(manifest, 2 * CHUNK_MS)! + expect(covered.has(target.chunk_seq)).toBe(true) + }) + + it("anchors on chunk 0 when the session has no checkpoint at all", () => { + const noCheckpoints = manifest.map((c) => ({ ...c, is_checkpoint: false })) + const target = chunkAtOffset(noCheckpoints, 90 * CHUNK_MS)! + expect(checkpointAtOrBefore(noCheckpoints, target.chunk_seq)?.chunk_seq).toBe(0) + }) +}) + +describe("live sessions", () => { + it("grows only the trailing range as chunks arrive", () => { + // While a recording is still being written its last range is partial, so + // it is the one boundary that moves. Everything before it must not. + const grown = Array.from({ length: 60 }, (_, seq) => chunk(seq)) + const early = planRanges(grown.slice(0, 45)) + const later = planRanges(grown) + expect(later.slice(0, early.length - 1)).toEqual(early.slice(0, early.length - 1)) + expect(later[early.length - 1]).not.toEqual(early[early.length - 1]) + }) + + it("re-requests the grown trailing range under a new key", () => { + // Which is why the loader must drop chunks it already holds when that + // range comes back: without it, the overlap is fed to `addEvent` twice + // and those mutations replay into the live engine a second time. + const grown = Array.from({ length: 60 }, (_, seq) => chunk(seq)) + const early = planRanges(grown.slice(0, 45)) + const later = planRanges(grown) + const tailBefore = early[early.length - 1]! + const tailAfter = rangeContaining(later, tailBefore.toChunkSeq + 1)! + expect(rangeKey(tailAfter)).not.toBe(rangeKey(tailBefore)) + expect(tailAfter.fromChunkSeq).toBe(tailBefore.fromChunkSeq) + // The overlap the loader has to filter out. + expect(tailAfter.toChunkSeq).toBeGreaterThan(tailBefore.toChunkSeq) + }) +}) + +describe("request volume", () => { + it("replaces one unbounded request with a bounded request per range", () => { + // The load-bearing property. Before: 1 request, entire payload, no ceiling. + // After: N requests, each capped at RANGE_SIZE chunks and 8 MB. + const fetched = simulate(manifest, 100 * CHUNK_MS) + expect(fetched.length).toBe(plan.length) + for (const r of fetched) { + expect(r.toChunkSeq - r.fromChunkSeq + 1).toBeLessThanOrEqual(MAX_CHUNKS_PER_RANGE) + } + }) +}) diff --git a/apps/web/src/components/replays/replay-range.test.ts b/apps/web/src/components/replays/replay-range.test.ts new file mode 100644 index 000000000..2c1fff24b --- /dev/null +++ b/apps/web/src/components/replays/replay-range.test.ts @@ -0,0 +1,242 @@ +import { describe, expect, it } from "vitest" +import { + INITIAL_WINDOW_BYTES, + MAX_BYTES_PER_RANGE, + MAX_CHUNKS_PER_RANGE, + type ReplayChunkMeta, + checkpointAtOrBefore, + chunkAtOffset, + chunkStartMs, + initialRanges, + planRanges, + manifestDurationMs, + rangeContaining, + rangesCovering, + replayRangeInput, +} from "./replay-range" + +const START_MS = 1_764_150_000_000 + +/** Chunks every 5s, ~100 KB each — the SDK's steady-state flush cadence. */ +const chunk = (seq: number, overrides: Partial = {}): ReplayChunkMeta => ({ + chunk_seq: seq, + timestamp: new Date(START_MS + seq * 5_000).toISOString(), + duration_ms: 5_000, + event_count: 200, + byte_size: 100_000, + is_checkpoint: seq === 0, + ...overrides, +}) + +const manifest = (count: number, overrides: (seq: number) => Partial = () => ({})) => + Array.from({ length: count }, (_, seq) => chunk(seq, overrides(seq))) + +describe("range planning", () => { + it("keeps every range inside the server's byte budget", () => { + // Real sessions carry ~838 KB chunks. A fixed 16-chunk range of those is + // ~13 MB, which the server refuses with 413 — and the client has no way to + // recover a range it cannot request. + const big = manifest(64, () => ({ byte_size: 838_026 })) + for (const range of planRanges(big)) { + const bytes = big + .filter((c) => c.chunk_seq >= range.fromChunkSeq && c.chunk_seq <= range.toChunkSeq) + .reduce((sum, c) => sum + c.byte_size, 0) + expect(bytes).toBeLessThanOrEqual(MAX_BYTES_PER_RANGE) + } + }) + + it("still caps chunk count when chunks are small", () => { + for (const range of planRanges(manifest(64))) { + expect(range.toChunkSeq - range.fromChunkSeq + 1).toBeLessThanOrEqual(MAX_CHUNKS_PER_RANGE) + } + }) + + it("covers every chunk exactly once", () => { + const chunks = manifest(50, (seq) => ({ byte_size: seq % 7 === 0 ? 900_000 : 100_000 })) + const seen: number[] = [] + for (const r of planRanges(chunks)) { + for (let s = r.fromChunkSeq; s <= r.toChunkSeq; s++) seen.push(s) + } + expect(seen).toEqual(chunks.map((c) => c.chunk_seq)) + }) + + it("keeps boundaries stable as a live session appends chunks", () => { + // The cache-correctness property: the atom family keys on the range, so a + // boundary that shifted when new chunks arrived would orphan everything + // already fetched and refetch the session from the start. + const grown = manifest(60) + const early = planRanges(grown.slice(0, 40)) + const later = planRanges(grown) + expect(later.slice(0, early.length - 1)).toEqual(early.slice(0, early.length - 1)) + }) + + it("resolves any chunk in a range to that same range", () => { + const chunks = manifest(64) + const p = planRanges(chunks) + const first = p[0]! + for (const seq of [first.fromChunkSeq, first.toChunkSeq]) { + expect(rangeContaining(p, seq)).toEqual(first) + } + }) + + it("produces one identical atom key for every position inside a range", () => { + const chunks = manifest(64) + const p = planRanges(chunks) + const window = { windowStart: "2026-08-04 10:00:00", windowEnd: "2026-08-04 11:00:00" } + const keys = new Set( + [p[0]!.fromChunkSeq, p[0]!.fromChunkSeq + 1, p[0]!.toChunkSeq].map((seq) => + JSON.stringify(replayRangeInput("sess_1", window, rangeContaining(p, seq)!)), + ), + ) + expect(keys.size).toBe(1) + }) + + it("returns nothing for a chunk past the end of the plan", () => { + expect(rangeContaining(planRanges(manifest(4)), 999)).toBeUndefined() + }) + + it("keeps ranges within the server's per-request chunk cap", () => { + expect(MAX_CHUNKS_PER_RANGE).toBeLessThanOrEqual(40) + }) +}) + +describe("initialRanges", () => { + it("stops on the byte budget, not a chunk count", () => { + // 100 KB chunks: ~20 chunks fit in the 2 MB budget, so more than one + // 16-chunk range. + const ranges = initialRanges(manifest(64), planRanges(manifest(64))) + const bytes = ranges.length * MAX_CHUNKS_PER_RANGE * 100_000 + expect(bytes).toBeGreaterThanOrEqual(INITIAL_WINDOW_BYTES) + expect(ranges.length).toBeLessThanOrEqual(2) + }) + + it("loads exactly one range when chunks are large", () => { + // A 4 MB chunk (full snapshot of a dense DOM) blows the budget alone. The + // count-based version of this would have pulled 8 of them — 32 MB. + const ranges = initialRanges(manifest(64, () => ({ byte_size: 4_000_000 })), planRanges(manifest(64, () => ({ byte_size: 4_000_000 })))) + expect(ranges).toHaveLength(1) + }) + + it("seeds from the first checkpoint, not chunk 0", () => { + // Chunks before the first snapshot are mutations against a DOM that was + // never built — loading them buys a blank screen. + const chunks = manifest(64, (seq) => ({ is_checkpoint: seq === 20 })) + expect(initialRanges(chunks, planRanges(chunks))[0]).toEqual(rangeContaining(planRanges(chunks), 20)) + }) + + it("returns nothing for an empty manifest", () => { + expect(initialRanges([], planRanges([]))).toEqual([]) + }) + + it("never runs past the end of a short session", () => { + // 3 chunks = 300 KB, well under the budget: it must stop at the manifest + // end rather than looping toward the budget forever. + expect(initialRanges(manifest(3), planRanges(manifest(3)))).toEqual([rangeContaining(planRanges(manifest(3)), 0)]) + }) +}) + +describe("rangesCovering", () => { + it("always yields at least one range, even past the budget", () => { + expect(rangesCovering(manifest(4), planRanges(manifest(4)), 0, 0)).toHaveLength(1) + }) + + it("returns nothing when there is nothing to cover", () => { + expect(rangesCovering([], planRanges([]), 0, INITIAL_WINDOW_BYTES)).toEqual([]) + }) +}) + +describe("checkpointAtOrBefore", () => { + it("finds the nearest preceding snapshot — the only valid seek anchor", () => { + const chunks = manifest(64, (seq) => ({ is_checkpoint: seq % 20 === 0 })) + expect(checkpointAtOrBefore(chunks, 45)?.chunk_seq).toBe(40) + }) + + it("never returns a checkpoint after the target", () => { + const chunks = manifest(64, (seq) => ({ is_checkpoint: seq % 20 === 0 })) + expect(checkpointAtOrBefore(chunks, 39)?.chunk_seq).toBe(20) + }) + + it("falls back to the first chunk when no checkpoint precedes the target", () => { + // Sessions whose opening snapshot was dropped by the SDK's over-cap buffer + // guard have no checkpoint at all — seeking must still do something. + const chunks = manifest(8, () => ({ is_checkpoint: false })) + expect(checkpointAtOrBefore(chunks, 5)?.chunk_seq).toBe(0) + }) +}) + +describe("playback clock", () => { + it("positions a chunk by its ingest timestamp", () => { + expect(chunkStartMs(chunk(4))).toBe(START_MS + 20_000) + }) + + it("derives total duration from the manifest, not from loaded events", () => { + // getMetaData() reads live context, so a player-derived total would grow + // as chunks stream in and the scrubber would stretch mid-playback. + expect(manifestDurationMs(manifest(10))).toBe(9 * 5_000 + 5_000) + }) + + it("reports zero duration for an empty manifest", () => { + expect(manifestDurationMs([])).toBe(0) + }) +}) + +// Every session in the 30-day retention predates this change, and none of them +// carry anything the old read path didn't already write. These pin that the new +// planning logic works off columns that have always been populated — so an +// existing recording plays exactly like one captured after the change. +describe("sessions recorded before this change", () => { + it("needs nothing beyond chunk_seq, timestamp, duration, size and checkpoint", () => { + // The full input contract. If a future field creeps in here, older + // sessions start failing to plan. + expect(Object.keys(chunk(0)).sort()).toEqual([ + "byte_size", + "chunk_seq", + "duration_ms", + "event_count", + "is_checkpoint", + "timestamp", + ]) + }) + + it("plays a session that never produced a checkpoint at all", () => { + // The SDK's over-cap buffer guard drops the batch containing the opening + // snapshot; such sessions have no checkpoint anywhere. Playback must still + // pick an anchor rather than refusing. + const chunks = manifest(8, () => ({ is_checkpoint: false })) + expect(checkpointAtOrBefore(chunks, 5)?.chunk_seq).toBe(0) + expect(initialRanges(chunks, planRanges(chunks))[0]).toEqual(rangeContaining(planRanges(chunks), 0)) + }) + + it("handles a single-chunk session", () => { + const chunks = manifest(1) + expect(initialRanges(chunks, planRanges(chunks))).toEqual([{ fromChunkSeq: 0, toChunkSeq: 0 }]) + expect(chunkAtOffset(chunks, 0)?.chunk_seq).toBe(0) + expect(manifestDurationMs(chunks)).toBe(5_000) + }) + + it("handles a session whose chunk_seq does not start at 0", () => { + // nextChunkSeq() is monotonic across reloads and persisted on the session + // record, so a session resumed after a refresh starts mid-sequence. + const chunks = manifest(20).map((c) => ({ ...c, chunk_seq: c.chunk_seq + 137 })) + expect(initialRanges(chunks, planRanges(chunks))[0]).toEqual(rangeContaining(planRanges(chunks), 137)) + expect(checkpointAtOrBefore(chunks, 150)?.chunk_seq).toBe(137) + }) +}) + +describe("chunkAtOffset", () => { + it("resolves a playback offset to its covering chunk", () => { + expect(chunkAtOffset(manifest(10), 22_000)?.chunk_seq).toBe(4) + }) + + it("clamps a target before the recording to the first chunk", () => { + expect(chunkAtOffset(manifest(10), -5_000)?.chunk_seq).toBe(0) + }) + + it("clamps a target past the end to the last chunk", () => { + expect(chunkAtOffset(manifest(10), 10_000_000)?.chunk_seq).toBe(9) + }) + + it("returns nothing for an empty manifest", () => { + expect(chunkAtOffset([], 0)).toBeUndefined() + }) +}) diff --git a/apps/web/src/components/replays/replay-range.ts b/apps/web/src/components/replays/replay-range.ts new file mode 100644 index 000000000..0d8b597ef --- /dev/null +++ b/apps/web/src/components/replays/replay-range.ts @@ -0,0 +1,260 @@ +import type { ReplayPartitionWindow } from "./replay-format" + +/** + * Chunk-range planning for progressive replay playback. + * + * A session's rrweb payload can run to hundreds of megabytes, so the player + * fetches it a range at a time rather than all at once. These helpers decide + * *which* ranges — and, just as importantly, keep those ranges stable so the + * atom cache can actually hit. + */ + +/** + * Upper bound on chunks per fetched range. + * + * A range can be shorter — `planRanges` closes one early when its payload would + * exceed the server's byte budget. Must stay <= the server's + * `max_chunks_per_request` (40); the manifest echoes the real cap so a mismatch + * is detectable rather than silent. + */ +export const MAX_CHUNKS_PER_RANGE = 16 + +/** + * Fallback byte budget for one range, used before the manifest reports the + * server's own `max_bytes_per_request`. Matches the server default. + */ +export const MAX_BYTES_PER_RANGE = 8_000_000 + +/** + * Bytes to load before starting playback. + * + * Byte-driven, not count-driven: a chunk flushes at ~100 KB but may be up to + * 4 MB (a full snapshot of a dense DOM), so "8 chunks" can mean 800 KB or 32 MB. + */ +export const INITIAL_WINDOW_BYTES = 2_000_000 + +/** Playback time to keep buffered ahead of the playhead. */ +export const PREFETCH_AHEAD_MS = 20_000 + +/** One chunk's manifest entry — the subset of the v2 manifest the player needs. */ +export interface ReplayChunkMeta { + readonly chunk_seq: number + readonly timestamp: string + readonly duration_ms: number + readonly event_count: number + readonly byte_size: number + readonly is_checkpoint: boolean +} + +export interface ReplayRange { + readonly fromChunkSeq: number + readonly toChunkSeq: number +} + +/** + * Cut the session into the ranges playback will fetch. + * + * Two properties matter, and a fixed-size grid only delivers one of them: + * + * 1. **Every range fits the server's byte budget.** Chunk sizes vary by more + * than an order of magnitude — a flush is ~100 KB but a full DOM snapshot + * runs to several hundred KB or more, and real sessions carry 838 KB chunks. + * A fixed 16-chunk slot of those is ~13 MB, well past the 8 MB ceiling, so + * the request comes back 413 and that stretch of the recording is unplayable. + * 2. **Boundaries are stable**, because the atom family keys on the range. A + * range recomputed from the playhead would mint a new key every tick and + * nothing would ever be cached. + * + * Walking greedily from the first chunk gives both: the cut points depend only + * on chunks at or before them, so a live session appending chunks extends the + * plan without moving any boundary already in it. + * + * A single chunk always gets a range even if it alone exceeds the budget — the + * SDK caps a chunk at 4 MB, so under an 8 MB budget that can't happen, but + * emitting nothing would stall playback rather than fail it. + */ +export const planRanges = ( + chunks: ReadonlyArray, + maxBytes: number = MAX_BYTES_PER_RANGE, + maxChunks: number = MAX_CHUNKS_PER_RANGE, +): ReadonlyArray => { + const ranges: Array = [] + let start: number | undefined + let end = 0 + let bytes = 0 + let count = 0 + for (const chunk of chunks) { + if (start !== undefined && (bytes + chunk.byte_size > maxBytes || count >= maxChunks)) { + ranges.push({ fromChunkSeq: start, toChunkSeq: end }) + start = undefined + bytes = 0 + count = 0 + } + if (start === undefined) start = chunk.chunk_seq + end = chunk.chunk_seq + bytes += chunk.byte_size + count++ + } + if (start !== undefined) ranges.push({ fromChunkSeq: start, toChunkSeq: end }) + return ranges +} + +/** The planned range containing `chunkSeq`, or undefined if it is past the end. */ +export const rangeContaining = ( + plan: ReadonlyArray, + chunkSeq: number, +): ReplayRange | undefined => + plan.find((range) => chunkSeq >= range.fromChunkSeq && chunkSeq <= range.toChunkSeq) + +/** Stable key for a range — must match `replayRangeInput`'s identity exactly. */ +export const rangeKey = (range: ReplayRange) => `${range.fromChunkSeq}:${range.toChunkSeq}` + +/** + * Build the atom-family input for a range. + * + * `makeQueryAtomFamily` keys on `JSON.stringify(input)`, so key identity is + * property-ORDER sensitive. Every call site goes through this one builder so an + * innocently reordered object literal can't silently split the cache. + */ +export const replayRangeInput = ( + sessionId: string, + window: ReplayPartitionWindow | undefined, + range: ReplayRange, +) => ({ + sessionId, + ...window, + fromChunkSeq: range.fromChunkSeq, + toChunkSeq: range.toChunkSeq, +}) + +/** + * Where a chunk sits on the playback timeline. + * + * `timestamp` is the ingest gateway's receipt time, so it trails the recording's + * own clock by the upload latency — well inside one chunk's duration. That makes + * it precise enough to resolve a seek to the right *chunk*, which is all this is + * used for; the exact offset within that chunk comes from its rrweb events once + * loaded. + * + * Deliberately not a stored first-event timestamp: adding one meant a new + * warehouse column, and a column the deployed cluster doesn't have yet fails + * every read with schema drift (and every insert), so it would have broken all + * replay until an out-of-band migration landed. Not worth sub-second precision + * on a chunk picker. + */ +export const chunkStartMs = (chunk: ReplayChunkMeta): number => Date.parse(chunk.timestamp) + +/** Epoch-ms the recording starts at, or 0 for an empty manifest. */ +export const manifestStartMs = (chunks: ReadonlyArray): number => { + const first = chunks[0] + return first === undefined ? 0 : chunkStartMs(first) +} + +/** + * Total playable length. + * + * Derived from the manifest rather than the player's `getMetaData()`, which + * reads live context and would therefore *grow* as chunks stream in — the + * scrubber would stretch while you watched. + */ +export const manifestDurationMs = (chunks: ReadonlyArray): number => { + const last = chunks[chunks.length - 1] + if (last === undefined) return 0 + return Math.max(0, chunkStartMs(last) + last.duration_ms - manifestStartMs(chunks)) +} + +/** + * The chunk covering `offsetMs` into the recording. + * + * Returns the last chunk that starts at or before the target; falls back to the + * first chunk for a target before the recording begins. + */ +export const chunkAtOffset = ( + chunks: ReadonlyArray, + offsetMs: number, +): ReplayChunkMeta | undefined => { + if (chunks.length === 0) return undefined + const targetMs = manifestStartMs(chunks) + offsetMs + let match = chunks[0] + for (const chunk of chunks) { + if (chunkStartMs(chunk) > targetMs) break + match = chunk + } + return match +} + +/** + * The seek anchor for a target chunk: the nearest checkpoint at or before it. + * + * rrweb can only start from a full DOM snapshot, so seeking into an unloaded + * region means loading from here — not from the target chunk, which on its own + * is a stream of mutations against a DOM that was never built. + * + * Falls back to the first chunk when no checkpoint precedes the target, which + * happens on sessions whose opening snapshot was dropped (see the over-cap + * buffer path in the SDK recorder). + */ +export const checkpointAtOrBefore = ( + chunks: ReadonlyArray, + chunkSeq: number, +): ReplayChunkMeta | undefined => { + let anchor: ReplayChunkMeta | undefined + for (const chunk of chunks) { + if (chunk.chunk_seq > chunkSeq) break + if (chunk.is_checkpoint) anchor = chunk + } + return anchor ?? chunks[0] +} + +/** + * Ranges to load for the opening frame: from the first checkpoint until the + * byte budget is met. + * + * Starting at the first checkpoint rather than chunk 0 matters for sessions + * whose first chunks are pre-snapshot noise — without a snapshot there is + * nothing to render, so those bytes would buy a blank screen. + */ +export const initialRanges = ( + chunks: ReadonlyArray, + plan: ReadonlyArray, +): ReadonlyArray => { + if (chunks.length === 0) return [] + const seed = chunks.find((chunk) => chunk.is_checkpoint) ?? chunks[0]! + return rangesCovering(chunks, plan, seed.chunk_seq, INITIAL_WINDOW_BYTES) +} + +/** + * Planned ranges from the one containing `fromChunkSeq`, extended until + * `byteBudget` is met. + * + * Always returns at least one range, so a chunk larger than the whole budget + * still loads instead of stalling playback forever. + */ +export const rangesCovering = ( + chunks: ReadonlyArray, + plan: ReadonlyArray, + fromChunkSeq: number, + byteBudget: number, +): ReadonlyArray => { + const startIndex = plan.findIndex( + (range) => fromChunkSeq >= range.fromChunkSeq && fromChunkSeq <= range.toChunkSeq, + ) + if (startIndex < 0) return [] + const bytesIn = (range: ReplayRange) => + chunks.reduce( + (sum, chunk) => + chunk.chunk_seq >= range.fromChunkSeq && chunk.chunk_seq <= range.toChunkSeq + ? sum + chunk.byte_size + : sum, + 0, + ) + const ranges: Array = [] + let bytes = 0 + for (let i = startIndex; i < plan.length; i++) { + const range = plan[i]! + ranges.push(range) + bytes += bytesIn(range) + if (bytes >= byteBudget) break + } + return ranges +} diff --git a/apps/web/src/components/replays/use-replay-chunk-loader.ts b/apps/web/src/components/replays/use-replay-chunk-loader.ts new file mode 100644 index 000000000..abbc64d44 --- /dev/null +++ b/apps/web/src/components/replays/use-replay-chunk-loader.ts @@ -0,0 +1,317 @@ +import * as React from "react" +import { Result, useAtomValue } from "@/lib/effect-atom" +import { getReplayEventsResultAtom } from "@/lib/services/atoms/warehouse-query-atoms" +import { normalizeEvents } from "./replay-events" +import type { ReplayPartitionWindow } from "./replay-format" +import { + INITIAL_WINDOW_BYTES, + MAX_BYTES_PER_RANGE, + MAX_CHUNKS_PER_RANGE, + PREFETCH_AHEAD_MS, + type ReplayChunkMeta, + type ReplayRange, + chunkAtOffset, + chunkStartMs, + checkpointAtOrBefore, + initialRanges, + planRanges, + manifestStartMs, + rangeContaining, + rangeKey, + rangesCovering, + replayRangeInput, +} from "./replay-range" + +// --------------------------------------------------------------------------- +// Progressive chunk loading +// +// The player used to fetch a session's entire rrweb payload before rendering a +// frame. Ingest accepts up to 1 GiB per session and the p99 is ~594 MB, so on a +// long session that read died in the Worker and surfaced as a 503 blaming the +// database. Now: seed from a checkpoint, play, and pull the rest as it's needed. +// +// Ranges load one at a time. Sequential is not a limitation here — playback +// consumes chunks in order, so a second in-flight range would only ever be the +// one after the range already being fetched. One subscription also keeps this a +// plain hook instead of an imperative registry subscription per range. +// --------------------------------------------------------------------------- + +/** What the player needs to know about loading, beyond the events themselves. */ +export type ReplayBufferState = "idle" | "buffering" | "seeking" + +interface LoadedRange { + readonly key: string + readonly range: ReplayRange + readonly events: ReadonlyArray + /** Highest chunk_seq actually returned — the real end of loaded playback. */ + readonly lastChunkSeq: number +} + +interface ChunkLoaderArgs { + readonly sessionId: string + readonly window: ReplayPartitionWindow | undefined + readonly chunks: ReadonlyArray + /** False while the manifest is still loading, or for the preview route. */ + readonly enabled: boolean +} + +export interface ChunkLoaderState { + /** + * Events the engine is constructed from. Changes only when playback needs a + * different anchor (first load, or a seek into an unloaded region), so a + * prefetch arriving mid-playback never tears the player down. + */ + readonly seedEvents: ReadonlyArray + /** Events loaded after the seed, in arrival order, for incremental feeding. */ + readonly trailingEvents: ReadonlyArray + readonly bufferState: ReplayBufferState + readonly loadError: unknown + /** Playback offset (ms from recording start) the engine should resume at. */ + readonly seekTargetMs: number | null + /** Ask for a playback offset; rebuilds from a checkpoint if it isn't loaded. */ + readonly requestSeek: (offsetMs: number) => boolean + /** Report the playhead so the loader can stay ahead of it. */ + readonly reportProgress: (offsetMs: number) => void +} + +const EMPTY: ReadonlyArray = [] + +/** Parse one range response into a flat, normalized rrweb event array. */ +const decodeRange = (chunks: ReadonlyArray<{ chunk_seq: number; events: string }>): unknown[] => { + const ordered = [...chunks].sort((a, b) => a.chunk_seq - b.chunk_seq) + const all: unknown[] = [] + for (const chunk of ordered) { + try { + const parsed: unknown = JSON.parse(chunk.events) + if (Array.isArray(parsed)) all.push(...parsed) + } catch { + // Skip a malformed chunk rather than failing the whole replay. + } + } + // Normalized per range, not across the session: chunks can upload out of + // order, and this is what makes each range independently well-formed. + return normalizeEvents(all) +} + +export function useReplayChunkLoader({ + sessionId, + window, + chunks, + enabled, +}: ChunkLoaderArgs): ChunkLoaderState { + // Ranges belonging to the current seed, in load order. Reset on a rebuild. + const [loaded, setLoaded] = React.useState>([]) + // How many of `loaded` make up the seed. The rest are fed incrementally. + const [seedCount, setSeedCount] = React.useState(0) + const [queue, setQueue] = React.useState>([]) + const [seekTargetMs, setSeekTargetMs] = React.useState(null) + const [isSeeking, setIsSeeking] = React.useState(false) + const playheadRef = React.useRef(0) + + const startMs = React.useMemo(() => manifestStartMs(chunks), [chunks]) + // Where the session gets cut into fetchable ranges. Derived from the manifest + // so no range can exceed the server's byte ceiling — chunk sizes vary by more + // than an order of magnitude, and a fixed chunk count would 413 on a session + // whose snapshots are large. + const plan = React.useMemo( + () => planRanges(chunks, MAX_BYTES_PER_RANGE, MAX_CHUNKS_PER_RANGE), + [chunks], + ) + + // Plan the opening window once the manifest lands — and only once. + // + // A live session's manifest grows on every refetch, so re-planning whenever + // `chunks` changes identity would discard the seed and rebuild the player + // every few seconds. Growth is picked up by `enqueueNext` instead; a genuinely + // different session is what warrants starting over. + const plannedForRef = React.useRef(null) + React.useEffect(() => { + if (!enabled || chunks.length === 0) return + if (plannedForRef.current === sessionId) return + plannedForRef.current = sessionId + setLoaded([]) + setSeedCount(0) + setIsSeeking(false) + setSeekTargetMs(null) + setQueue(initialRanges(chunks, plan)) + }, [enabled, chunks, plan, sessionId]) + + const activeRange = queue[0] + const loadedKeys = React.useMemo(() => new Set(loaded.map((entry) => entry.key)), [loaded]) + + // One subscription, always mounted so the hook order never changes. With no + // range to fetch it re-reads the last one, which the 10-minute atom TTL serves + // from cache — chunk rows are immutable, so a repeat read is free. + // + // Before the manifest lands there is no range at all. Subscribing to an + // invented window would fire a request for chunks we have no reason to think + // exist, so hold at the plan's first range once there is one — that is where + // playback starts for most sessions, making it a prefetch rather than waste. + const subscribedRange = + activeRange ?? loaded[loaded.length - 1]?.range ?? plan[0] ?? { fromChunkSeq: 0, toChunkSeq: 0 } + const rangeResult = useAtomValue( + getReplayEventsResultAtom({ data: replayRangeInput(sessionId, window, subscribedRange) }), + ) + + const loadError = React.useMemo( + () => + activeRange === undefined + ? null + : Result.builder(rangeResult) + .onError((error) => error) + .orElse(() => null), + [rangeResult, activeRange], + ) + + // Commit a completed range and advance the queue. + React.useEffect(() => { + if (activeRange === undefined) return + const key = rangeKey(activeRange) + if (loadedKeys.has(key)) { + setQueue((current) => current.slice(1)) + return + } + const committed = Result.builder(rangeResult) + .onSuccess((result) => result) + .orElse(() => null) + if (committed === null) return + const all = committed.chunks as ReadonlyArray<{ chunk_seq: number; events: string }> + // Take only chunks this range is responsible for and that aren't already + // loaded. + // + // The overlap is real on a LIVE session: the plan's last range is partial + // while the recording is still being written, so a range loaded as 40–45 + // becomes 40–55 once more chunks land. That is a different key, so it is + // fetched again — and without this filter chunks 40–45 would be handed to + // `addEvent` a second time, replaying those mutations into the live engine. + // + // The range check also means a result that somehow belongs to a different + // range is dropped rather than committed under this one's key. + const alreadyThrough = loaded.reduce((max, entry) => Math.max(max, entry.lastChunkSeq), -1) + const rows = all.filter( + (row) => + row.chunk_seq >= activeRange.fromChunkSeq && + row.chunk_seq <= activeRange.toChunkSeq && + row.chunk_seq > alreadyThrough, + ) + const events = decodeRange(rows) + const lastChunkSeq = rows.reduce((max, row) => Math.max(max, row.chunk_seq), alreadyThrough) + setLoaded((current) => [...current, { key, range: activeRange, events, lastChunkSeq }]) + setQueue((current) => current.slice(1)) + }, [rangeResult, activeRange, loadedKeys, loaded]) + + // The seed is everything loaded before the queue first drains; after that, + // arrivals are trailing and get fed to the live engine. + React.useEffect(() => { + if (queue.length === 0 && loaded.length > 0 && seedCount === 0) { + setSeedCount(loaded.length) + setIsSeeking(false) + } + }, [queue.length, loaded.length, seedCount]) + + const seedEvents = React.useMemo(() => { + if (seedCount === 0) return EMPTY + return loaded.slice(0, seedCount).flatMap((entry) => entry.events) + }, [loaded, seedCount]) + + const trailingEvents = React.useMemo(() => { + if (seedCount === 0) return EMPTY + return loaded.slice(seedCount).flatMap((entry) => entry.events) + }, [loaded, seedCount]) + + /** Highest chunk sequence currently loaded, or -1. */ + const loadedThroughSeq = React.useMemo( + () => loaded.reduce((max, entry) => Math.max(max, entry.lastChunkSeq), -1), + [loaded], + ) + + /** Playback offset at which loaded events run out. */ + const loadedUntilMs = React.useMemo(() => { + if (loadedThroughSeq < 0) return 0 + const last = chunks.find((chunk) => chunk.chunk_seq === loadedThroughSeq) + if (last === undefined) return 0 + return chunkStartMs(last) + last.duration_ms - startMs + }, [chunks, loadedThroughSeq, startMs]) + + const hasMoreChunks = React.useMemo(() => { + const lastSeq = chunks[chunks.length - 1]?.chunk_seq + return lastSeq !== undefined && loadedThroughSeq >= 0 && loadedThroughSeq < lastSeq + }, [chunks, loadedThroughSeq]) + + /** Queue the next contiguous range, unless one is already in flight. */ + const enqueueNext = React.useCallback(() => { + if (!hasMoreChunks) return + setQueue((current) => { + if (current.length > 0) return current + const next = rangeContaining(plan, loadedThroughSeq + 1) + if (next === undefined || loadedKeys.has(rangeKey(next))) return current + return [next] + }) + }, [hasMoreChunks, loadedThroughSeq, loadedKeys, plan]) + + const reportProgress = React.useCallback( + (offsetMs: number) => { + playheadRef.current = offsetMs + if (seedCount === 0) return + if (offsetMs + PREFETCH_AHEAD_MS >= loadedUntilMs) enqueueNext() + }, + [seedCount, loadedUntilMs, enqueueNext], + ) + + /** + * Ask to jump to a playback offset. + * + * Returns true when the engine must be rebuilt: the target is outside the + * loaded span, so playback has to restart from the nearest preceding + * checkpoint. rrweb applies an event whose timestamp precedes its baseline + * *synchronously and out of DOM order*, so feeding backwards through + * `addEvent` would corrupt the frame — a rebuild is the only correct move. + */ + const requestSeek = React.useCallback( + (offsetMs: number): boolean => { + playheadRef.current = offsetMs + if (chunks.length === 0) return false + const target = chunkAtOffset(chunks, offsetMs) + if (target === undefined) return false + const withinLoaded = + target.chunk_seq <= loadedThroughSeq && + loaded.some( + (entry) => + target.chunk_seq >= entry.range.fromChunkSeq && + target.chunk_seq <= entry.range.toChunkSeq, + ) + if (withinLoaded) { + // Already have it — the engine seeks itself, no transport needed. + return false + } + const anchor = checkpointAtOrBefore(chunks, target.chunk_seq) + if (anchor === undefined) return false + setLoaded([]) + setSeedCount(0) + setIsSeeking(true) + setSeekTargetMs(offsetMs) + setQueue(rangesCovering(chunks, plan, anchor.chunk_seq, INITIAL_WINDOW_BYTES)) + return true + }, + [chunks, plan, loaded, loadedThroughSeq], + ) + + const bufferState: ReplayBufferState = isSeeking + ? "seeking" + : // Starved: the playhead has reached the end of what's loaded and more is + // still coming. Distinct from "finished", which is loadedThroughSeq at the + // manifest's end. + queue.length > 0 && seedCount > 0 && playheadRef.current + 1_000 >= loadedUntilMs && hasMoreChunks + ? "buffering" + : "idle" + + return { + seedEvents, + trailingEvents, + bufferState, + loadError, + seekTargetMs, + requestSeek, + reportProgress, + } +} diff --git a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts index b20d5b5d7..30b34c0b2 100644 --- a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts +++ b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts @@ -104,6 +104,7 @@ import { getQueryBuilderBreakdown } from "@/api/warehouse/query-builder-breakdow import { getReplay, getReplayEvents, + getReplayManifest, getReplaysFacets, getReplaysForTrace, getSessionTranscript, @@ -248,14 +249,22 @@ export const getSessionTraceSummariesResultAtom = makeQueryAtomFamily(getSession staleTime: 60_000, }) -// Idle TTL keeps the chunks (and their inline rrweb events) stable across the -// player's frequent re-renders so the decode memo in the player context isn't -// thrown away and re-run. The API serves event arrays inline whether they came -// from ClickHouse or R2, so there is no client-side fetch/gunzip either way. -export const getReplayEventsResultAtom = makeQueryAtomFamily(getReplayEvents, { +// The session's chunk timeline without payloads. Cheap enough to fetch on every +// replay open, and the prerequisite for every payload range. +export const getReplayManifestResultAtom = makeQueryAtomFamily(getReplayManifest, { staleTime: 240_000, }) +// One entry per chunk range. Held far longer than a normal list query because a +// chunk row is immutable once written (plain MergeTree, 30-day TTL) — so +// scrubbing back over an already-played stretch must never refetch it. +// +// Idle TTL also keeps the chunks stable across the player's frequent +// re-renders, so the decode memo in the player context isn't thrown away. +export const getReplayEventsResultAtom = makeQueryAtomFamily(getReplayEvents, { + staleTime: 600_000, +}) + // Distilled session transcript (console/network/error/nav/click) for the panels. export const getSessionTranscriptResultAtom = makeQueryAtomFamily(getSessionTranscript, { staleTime: 60_000, diff --git a/apps/web/src/routes/replays/$sessionId.tsx b/apps/web/src/routes/replays/$sessionId.tsx index fb4e77d1a..577a0ab0c 100644 --- a/apps/web/src/routes/replays/$sessionId.tsx +++ b/apps/web/src/routes/replays/$sessionId.tsx @@ -6,7 +6,7 @@ import { DashboardLayout } from "@/components/layout/dashboard-layout" import { ReplayStudio } from "@/components/replays/replay-studio" import { Result, useAtomValue } from "@/lib/effect-atom" import { - getReplayEventsResultAtom, + getReplayManifestResultAtom, getReplayResultAtom, getSessionTranscriptResultAtom, } from "@/lib/services/atoms/warehouse-query-atoms" @@ -29,7 +29,11 @@ export const Route = createFileRoute("/replays/$sessionId")({ const window = replayPartitionWindow(typeof deps.t === "string" ? deps.t : undefined) const data = { sessionId: params.sessionId, ...window } context.effectRegistry.mount(getReplayResultAtom({ data })) - context.effectRegistry.mount(getReplayEventsResultAtom({ data })) + // The manifest, not the payload: which payload range to fetch depends on + // where the first checkpoint is, which the manifest is what tells us. That + // costs one extra round-trip on a cold load and saves fetching a session + // that can run to hundreds of megabytes. + context.effectRegistry.mount(getReplayManifestResultAtom({ data })) context.effectRegistry.mount(getSessionTranscriptResultAtom({ data })) }, }) diff --git a/packages/domain/src/http/session-replay.ts b/packages/domain/src/http/session-replay.ts index ee70337e3..99b705775 100644 --- a/packages/domain/src/http/session-replay.ts +++ b/packages/domain/src/http/session-replay.ts @@ -185,31 +185,7 @@ export class GetReplayResponse extends Schema.Class("GetRepla ), }) {} -// --- Events (rrweb chunks, payload inline) --- - -export class GetReplayEventsRequest extends Schema.Class("GetReplayEventsRequest")({ - sessionId: SessionId, - // See GetReplayRequest — optional partition-pruning window. - windowStart: Schema.optional(TinybirdDateTime), - windowEnd: Schema.optional(TinybirdDateTime), -}) {} - -export const SessionReplayChunk = Schema.Struct({ - chunkSeq: Schema.Number, - timestamp: Schema.String, - durationMs: Schema.Number, - eventCount: Schema.Number, - byteSize: Schema.Number, - isCheckpoint: Schema.Number, - /** The rrweb event array for this chunk, serialized as a JSON string. */ - events: Schema.String, -}) - -export class GetReplayEventsResponse extends Schema.Class("GetReplayEventsResponse")( - { - chunks: Schema.Array(SessionReplayChunk), - }, -) {} +// Replay chunk payloads are not served here — see the API group below. // --- Reverse correlation (trace → sessions) --- @@ -332,13 +308,11 @@ export class SessionReplaysApiGroup extends HttpApiGroup.make("sessionReplays") error: sessionReplayEndpointErrors, }), ) - .add( - HttpApiEndpoint.post("getReplayEvents", "/events", { - payload: GetReplayEventsRequest, - success: GetReplayEventsResponse, - error: sessionReplayEndpointErrors, - }), - ) + // Replay payload reads live on v2 only: `GET /v2/session_replays/:id/manifest` + // then `GET /v2/session_replays/:id/events?from_chunk_seq=…`. There is no v1 + // equivalent on purpose — the v1 endpoint fetched a whole session's rrweb + // payload in one unbounded response, which is the bug, and keeping a second + // surface would have kept a way to reintroduce it. .add( HttpApiEndpoint.post("replaysForTrace", "/for-trace", { payload: ReplaysForTraceRequest, diff --git a/packages/domain/src/http/v2/errors.ts b/packages/domain/src/http/v2/errors.ts index 7d2dd9533..4dc682064 100644 --- a/packages/domain/src/http/v2/errors.ts +++ b/packages/domain/src/http/v2/errors.ts @@ -138,6 +138,35 @@ export class V2ConflictError extends Schema.ErrorClass("@maple/ }, ) {} +/** + * The request asked for more data than one response may carry. + * + * Reuses the `invalid_request_error` type — the closed enum stays closed — but + * keeps 413 so the distinction from an ordinary 400 survives: nothing about the + * request is malformed, the window is simply too wide. It is a statement about + * the size of the answer, so retrying it unchanged can only fail identically; + * the message says what to narrow. + */ +export class V2PayloadTooLargeError extends Schema.ErrorClass( + "@maple/http/v2/PayloadTooLargeError", +)( + { + error: errorBody("invalid_request_error", { + code: "range_too_large", + message: + "That part of the recording is too large to load in one request. Request a narrower chunk range.", + param: "to_chunk_seq", + }), + }, + { + httpApiStatus: 413, + identifier: "PayloadTooLargeError", + title: "Payload too large error", + description: + "The requested range would exceed the endpoint's response budget. Narrow the range and retry. HTTP 413.", + }, +) {} + export class V2RateLimitError extends Schema.ErrorClass("@maple/http/v2/RateLimitError")( { error: errorBody("rate_limit_error", { @@ -247,6 +276,22 @@ export const resourceNotFound = (resource: string, message: string, param = "id" export const conflict = (code: string, message: string) => new V2ConflictError({ error: { type: "conflict_error", code, message } }) +/** + * The message crosses the public boundary verbatim: unlike the warehouse + * errors, this one carries no database diagnostics — only the range the caller + * asked for and the caps it exceeded — and it is the one error here where + * telling the user exactly what to do is the whole value. + */ +export const payloadTooLarge = (message: string, param?: string) => + new V2PayloadTooLargeError({ + error: { + type: "invalid_request_error", + code: "range_too_large", + message, + ...(param !== undefined ? { param } : {}), + }, + }) + export const rateLimited = () => new V2RateLimitError({ error: { diff --git a/packages/domain/src/http/v2/openapi.test.ts b/packages/domain/src/http/v2/openapi.test.ts index e99949676..f56ce22cb 100644 --- a/packages/domain/src/http/v2/openapi.test.ts +++ b/packages/domain/src/http/v2/openapi.test.ts @@ -113,6 +113,7 @@ describe("MapleApiV2 OpenAPI", () => { "GET /v2/services/{name}", "GET /v2/session_replays/{id}", "GET /v2/session_replays/{id}/events", + "GET /v2/session_replays/{id}/manifest", "GET /v2/session_replays/{id}/transcript", "GET /v2/traces/{trace_id}", "GET /v2/traces/{trace_id}/spans/{span_id}", diff --git a/packages/domain/src/http/v2/session-replays.ts b/packages/domain/src/http/v2/session-replays.ts index 6a9d59e8b..b436602a8 100644 --- a/packages/domain/src/http/v2/session-replays.ts +++ b/packages/domain/src/http/v2/session-replays.ts @@ -6,6 +6,7 @@ import { ListOf, ListQuery, Timestamp } from "./envelopes" import { V2InvalidRequestError, V2NotFoundError, + V2PayloadTooLargeError, V2RateLimitError, V2ServiceUnavailableError, V2UpstreamError, @@ -126,18 +127,70 @@ export const V2SessionReplay = Schema.Struct({ }) export type V2SessionReplay = Schema.Schema.Type -export const V2SessionReplayChunk = Schema.Struct({ - object: Schema.Literal("session_replay.event_chunk").annotate({ - description: 'The object type — always `"session_replay.event_chunk"`.', - }), +/** + * A chunk's metadata without its payload. + * + * A session's rrweb payload is unbounded by construction, so it cannot be + * fetched in one response. The manifest carries every chunk's position and size + * cheaply, which is what lets a client decide which payload ranges to request — + * and lets a player start on a couple of MB regardless of session length. + */ +export const V2SessionReplayChunkMeta = Schema.Struct({ chunk_seq: Schema.Number.annotate({ description: "Ordinal of the chunk within the session." }), - timestamp: Timestamp.annotate({ description: "When the chunk's events start." }), + timestamp: Timestamp.annotate({ + description: + "When the ingest gateway received the chunk — the chunk's position on the playback timeline. It trails the recording's own clock by the upload latency, which is well inside one chunk's duration, so it resolves a seek to the right chunk; the exact offset within that chunk comes from its rrweb events.", + }), duration_ms: Schema.Number.annotate({ description: "Duration covered by the chunk in ms." }), event_count: Schema.Number.annotate({ description: "Number of rrweb events in the chunk." }), byte_size: Schema.Number.annotate({ description: "Serialized size of the chunk in bytes." }), is_checkpoint: Schema.Boolean.annotate({ - description: "Whether the chunk is a full-snapshot checkpoint.", + description: + "Whether the chunk contains a full DOM snapshot. Only a checkpoint can seed a player, so seeking means loading the nearest checkpoint at or before the target.", + }), +}).annotate({ + identifier: "SessionReplayChunkMeta", + title: "Session replay chunk metadata", + description: "One chunk's position and size, without its payload.", +}) +export type V2SessionReplayChunkMeta = Schema.Schema.Type + +export const V2SessionReplayManifest = Schema.Struct({ + object: Schema.Literal("session_replay.manifest").annotate({ + description: 'The object type — always `"session_replay.manifest"`.', + }), + session_id: SessionReplayPublicId, + chunks: Schema.Array(V2SessionReplayChunkMeta).annotate({ + description: "Every chunk in the session, ordered by `chunk_seq`.", + }), + chunk_count: Schema.Number.annotate({ description: "Number of chunks in the session." }), + total_byte_size: Schema.Number.annotate({ + description: "Sum of every chunk's `byte_size` — the session's full payload size.", + }), + max_chunks_per_request: Schema.Number.annotate({ + description: "Server cap on how many chunks one events request may return.", }), + max_bytes_per_request: Schema.Number.annotate({ + description: + "Server cap on the encoded size of one events response. A range whose payload exceeds this is refused with `range_too_large`.", + }), + truncated: Schema.Boolean.annotate({ + description: + "Whether the chunk list was cut off at the manifest ceiling. `false` for any session recordable under the current ingest limits.", + }), +}).annotate({ + identifier: "SessionReplayManifest", + title: "Session replay manifest", + description: + "The session's chunk timeline without payloads — fetch this first, then request payload ranges against it.", +}) +export type V2SessionReplayManifest = Schema.Schema.Type + +export const V2SessionReplayChunk = Schema.Struct({ + object: Schema.Literal("session_replay.event_chunk").annotate({ + description: 'The object type — always `"session_replay.event_chunk"`.', + }), + ...V2SessionReplayChunkMeta.fields, events: Schema.String.annotate({ description: "The rrweb event array for this chunk, serialized as a JSON string.", }), @@ -281,6 +334,46 @@ export const V2SessionReplayCollectionQuery = Schema.Struct({ description: "Pagination plus an optional time window for replay child collections.", }) +/** Chunks one events request may return. */ +export const MAX_REPLAY_CHUNKS_PER_REQUEST = 40 + +/** + * Encoded-response ceiling for one events request. + * + * Sits far below both the Worker heap and the platform's own body limit, so an + * over-wide range is refused by us — a `range_too_large` naming the range — and + * never dies as a platform abort, which the transient-error classifier reads as + * a flaky warehouse and retries before reporting the service as unavailable. + */ +export const MAX_REPLAY_EVENTS_RESPONSE_BYTES = 8_000_000 + +/** + * Hard ceiling on manifest rows. Well beyond reach: ingest caps a session at + * 1 GiB and chunks flush at ~100 KB, so a session tops out around 10k chunks. + */ +export const MAX_REPLAY_MANIFEST_CHUNKS = 20_000 + +export const V2SessionReplayEventsQuery = Schema.Struct({ + ...V2SessionReplayCollectionQuery.fields, + from_chunk_seq: Schema.optionalKey( + Schema.Number.check(Schema.isInt(), Schema.isGreaterThanOrEqualTo(0)).annotate({ + description: + "Only return chunks at or after this `chunk_seq`. Take it from the manifest — chunk sequences are stable identities, unlike offsets, which shift if a chunk uploads out of order.", + }), + ), + to_chunk_seq: Schema.optionalKey( + Schema.Number.check(Schema.isInt(), Schema.isGreaterThanOrEqualTo(0)).annotate({ + description: "Only return chunks at or before this `chunk_seq` (inclusive).", + }), + ), +}).annotate({ + identifier: "SessionReplayEventsQuery", + title: "Session replay events query", + description: + "Pagination, an optional time window, and an optional chunk range. The range selects which part of the recording to load; pagination then pages within it.", +}) +export type V2SessionReplayEventsQuery = Schema.Schema.Type + export const V2SessionReplaysForTraceParams = Schema.Struct({ trace_id: TraceId.annotate({ description: "The trace ID to find sessions for." }), start_time: Timestamp.annotate({ description: "Window start (ISO-8601)." }), @@ -362,18 +455,33 @@ export class V2SessionReplaysApiGroup extends HttpApiGroup.make("sessionReplays" }), ), ) + .add( + HttpApiEndpoint.get("manifest", "/:id/manifest", { + params: { id: SessionReplayPublicId }, + query: V2SessionReplayWindowQuery, + success: V2SessionReplayManifest, + error: [...commonErrors, V2NotFoundError], + }).annotateMerge( + OpenApi.annotations({ + identifier: "getSessionReplayManifest", + summary: "Retrieve a session replay manifest", + description: + "Returns every chunk's position and size without payloads — the session's timeline in one cheap read. Fetch this before `/events`: a session's payload can run to hundreds of megabytes, so it must be pulled a range at a time, and the manifest is what tells you which ranges exist and where the seekable checkpoints are. Requires the `session_replays:read` scope.", + }), + ), + ) .add( HttpApiEndpoint.get("events", "/:id/events", { params: { id: SessionReplayPublicId }, - query: V2SessionReplayCollectionQuery, + query: V2SessionReplayEventsQuery, success: SessionReplayChunkList, - error: [...commonErrors, V2NotFoundError], + error: [...commonErrors, V2NotFoundError, V2PayloadTooLargeError], }).annotateMerge( OpenApi.annotations({ identifier: "getSessionReplayEvents", summary: "List session replay events", description: - "Returns the session's rrweb event chunks (player payload) in order. Cursor-paginated. Requires the `session_replays:read` scope.", + "Returns the session's rrweb event chunks (player payload) in order. Scope the read with `from_chunk_seq`/`to_chunk_seq` from the manifest; the response is capped, and a range whose payload exceeds the cap is refused with `range_too_large` rather than truncated silently. Cursor-paginated within the range. Requires the `session_replays:read` scope.", }), ), ) diff --git a/packages/domain/src/tinybird/datasources.ts b/packages/domain/src/tinybird/datasources.ts index 2f42ecfb0..da22e7eef 100644 --- a/packages/domain/src/tinybird/datasources.ts +++ b/packages/domain/src/tinybird/datasources.ts @@ -1830,6 +1830,9 @@ export const sessionReplayEvents = defineDatasource("session_replay_events", { OrgId: column(t.string().lowCardinality(), { jsonPath: "$.org_id" }), SessionId: column(t.string(), { jsonPath: "$.session_id" }), ChunkSeq: column(t.uint32(), { jsonPath: "$.chunk_seq" }), + // Gateway receipt time. Drives partitioning and the TTL, and doubles as the + // chunk index's playback anchor: it trails the recording's own clock by the + // upload latency, which is well inside a single chunk's duration. Timestamp: column(t.dateTime64(9), { jsonPath: "$.timestamp" }), DurationMs: column(t.uint32().default(0), { jsonPath: "$.duration_ms" }), EventCount: column(t.uint32().default(0), { jsonPath: "$.event_count" }), diff --git a/packages/query-engine/src/__sql_baseline__/catalog.sql b/packages/query-engine/src/__sql_baseline__/catalog.sql index 730253c3e..f1af98480 100644 --- a/packages/query-engine/src/__sql_baseline__/catalog.sql +++ b/packages/query-engine/src/__sql_baseline__/catalog.sql @@ -1462,6 +1462,22 @@ SELECT LIMIT 1 FORMAT JSON +-- builder:session-replays:sessionReplayChunkIndexQuery:default [2e322b20] +SELECT + ChunkSeq AS chunkSeq, + Timestamp AS timestamp, + DurationMs AS durationMs, + EventCount AS eventCount, + ByteSize AS byteSize, + IsCheckpoint AS isCheckpoint + FROM session_replay_events + WHERE OrgId = 'org_sql_catalog' + AND SessionId = 'sess_0af7651916cd43dd' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + ORDER BY chunkSeq ASC + FORMAT JSON + -- builder:session-replays:sessionReplayEventsQuery:default [42cce486] SELECT ChunkSeq AS chunkSeq, @@ -1479,6 +1495,26 @@ SELECT ORDER BY chunkSeq ASC FORMAT JSON +-- builder:session-replays:sessionReplayEventsQuery:ranged [5bdfa930] +SELECT + ChunkSeq AS chunkSeq, + Timestamp AS timestamp, + DurationMs AS durationMs, + EventCount AS eventCount, + ByteSize AS byteSize, + Events AS events, + IsCheckpoint AS isCheckpoint + FROM session_replay_events + WHERE OrgId = 'org_sql_catalog' + AND SessionId = 'sess_0af7651916cd43dd' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND ChunkSeq >= 16 + AND ChunkSeq <= 31 + ORDER BY chunkSeq ASC + LIMIT 40 + FORMAT JSON + -- builder:session-replays:sessionReplaysFacetsQuery:default [896379cc] SELECT ServiceName AS name, diff --git a/packages/query-engine/src/ch/builder-fixtures.ts b/packages/query-engine/src/ch/builder-fixtures.ts index 9d6ddf0af..b77d87d28 100644 --- a/packages/query-engine/src/ch/builder-fixtures.ts +++ b/packages/query-engine/src/ch/builder-fixtures.ts @@ -97,6 +97,16 @@ export const builderFixtures: ReadonlyArray = [ sessionId: SESSION_ID, }), }, + { + module: "session-replays", + name: "sessionReplayChunkIndexQuery", + label: "default", + compile: () => + CH.compile(CH.sessionReplayChunkIndexQuery({ startTime: START_TIME, endTime: END_TIME }), { + orgId: ORG_ID, + sessionId: SESSION_ID, + }), + }, { module: "session-replays", name: "sessionReplayEventsQuery", @@ -107,6 +117,24 @@ export const builderFixtures: ReadonlyArray = [ sessionId: SESSION_ID, }), }, + { + // The shape playback actually emits: a bounded chunk window. Distinct SQL + // from "default" (ChunkSeq predicates + LIMIT), so it needs its own sweep. + module: "session-replays", + name: "sessionReplayEventsQuery", + label: "ranged", + compile: () => + CH.compile( + CH.sessionReplayEventsQuery({ + startTime: START_TIME, + endTime: END_TIME, + fromChunkSeq: 16, + toChunkSeq: 31, + limit: 40, + }), + { orgId: ORG_ID, sessionId: SESSION_ID }, + ), + }, { module: "session-replays", name: "sessionsForTraceQuery", diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts index 746c357bd..aa6408b59 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -119,6 +119,7 @@ export { sessionReplaysListQuery, sessionReplaysFacetsQuery, getSessionReplayQuery, + sessionReplayChunkIndexQuery, sessionReplayEventsQuery, sessionsForTraceQuery, sessionTraceSummariesQuery, @@ -127,6 +128,9 @@ export { type SessionReplaysFacetsOpts, type SessionReplaysFacetsOutput, type SessionReplayDetailOutput, + type SessionReplayChunkIndexOpts, + type SessionReplayChunkIndexOutput, + type SessionReplayEventsOpts, type SessionReplayEventsOutput, type SessionsForTraceOpts, type SessionsForTraceOutput, diff --git a/packages/query-engine/src/ch/queries/session-replays.test.ts b/packages/query-engine/src/ch/queries/session-replays.test.ts index 97dd520fa..45b734935 100644 --- a/packages/query-engine/src/ch/queries/session-replays.test.ts +++ b/packages/query-engine/src/ch/queries/session-replays.test.ts @@ -4,6 +4,7 @@ import { getSessionReplayQuery, sessionReplaysFacetsQuery, sessionReplaysListQuery, + sessionReplayChunkIndexQuery, sessionReplayEventsQuery, sessionsForTraceQuery, sessionTraceSummariesQuery, @@ -71,6 +72,56 @@ describe("sessionReplayEventsQuery", () => { const { sql } = compileCH(q, sessionParams) expect(sql).not.toContain("Timestamp >=") }) + + it("bounds the read to a chunk range so a session is fetched in slices", () => { + const q = sessionReplayEventsQuery({ ...WINDOW, fromChunkSeq: 16, toChunkSeq: 31, limit: 40 }) + const { sql } = compileCH(q, sessionParams) + expect(sql).toContain("ChunkSeq >= 16") + expect(sql).toContain("ChunkSeq <= 31") + expect(sql).toContain("LIMIT 40") + }) + + it("omits the chunk range when absent, so existing callers keep their SQL", () => { + const { sql } = compileCH(sessionReplayEventsQuery(WINDOW), sessionParams) + expect(sql).not.toContain("ChunkSeq >=") + expect(sql).not.toContain("ChunkSeq <=") + expect(sql).not.toContain("LIMIT") + }) + + it("keeps chunk 0 as a real lower bound rather than a falsy no-op", () => { + // `if (opts.fromChunkSeq)` would silently drop this — and chunk 0 is the + // single most-requested range (the initial playback window). + const { sql } = compileCH(sessionReplayEventsQuery({ fromChunkSeq: 0, toChunkSeq: 15 }), sessionParams) + expect(sql).toContain("ChunkSeq >= 0") + }) +}) + +describe("sessionReplayChunkIndexQuery", () => { + it("never reads the payload column — that is the whole point of the index", () => { + const { sql } = compileCH(sessionReplayChunkIndexQuery(WINDOW), sessionParams) + expect(sql).toContain("FROM session_replay_events") + expect(sql).not.toContain("Events") + expect(sql).toContain("AS byteSize") + expect(sql).toContain("AS isCheckpoint") + }) + + it("projects only columns every existing recording already has", () => { + // Deliberately no stored first-event timestamp: a column the deployed + // cluster lacks fails every read with schema drift (and every insert), so + // adding one would have broken all replay until a migration landed. The + // ingest timestamp positions a chunk closely enough to pick it. + const { sql } = compileCH(sessionReplayChunkIndexQuery(WINDOW), sessionParams) + expect(sql).toContain("Timestamp AS timestamp") + expect(sql).toContain("DurationMs AS durationMs") + expect(sql).not.toContain("FirstEventMs") + }) + + it("orders by chunk sequence and scopes to org + session", () => { + const { sql } = compileCH(sessionReplayChunkIndexQuery(WINDOW), sessionParams) + expect(sql).toContain("OrgId = 'org_1'") + expect(sql).toContain("SessionId = 'sess_1'") + expect(sql).toContain("ORDER BY chunkSeq ASC") + }) }) describe("sessionsForTraceQuery", () => { diff --git a/packages/query-engine/src/ch/queries/session-replays.ts b/packages/query-engine/src/ch/queries/session-replays.ts index c83b52190..02ecbf347 100644 --- a/packages/query-engine/src/ch/queries/session-replays.ts +++ b/packages/query-engine/src/ch/queries/session-replays.ts @@ -630,7 +630,13 @@ export function getSessionReplayQuery(opts: SessionReplayDetailOpts = {}) { } // --------------------------------------------------------------------------- -// Chunk index for one session (ordered for playback) +// Chunk reads for one session (ordered for playback) +// +// Two builders, deliberately split: `sessionReplayChunkIndexQuery` returns the +// timeline without payloads, and `sessionReplayEventsQuery` returns payloads for +// a bounded chunk range. Playback fetches the index once, then pulls ranges on +// demand. Reading every chunk's `Events` in one go is what made large sessions +// fail — see the range opts below. // // session_replay_events is a plain MergeTree — each chunk is written exactly // once, so no dedup is needed. Sorted by (OrgId, SessionId, ChunkSeq) so the @@ -643,10 +649,73 @@ export function getSessionReplayQuery(opts: SessionReplayDetailOpts = {}) { // the session's time window) prune to the 1-2 partitions the session spans. // --------------------------------------------------------------------------- +export interface SessionReplayChunkIndexOpts { + /** Optional session time window — prunes daily partitions. Omit to scan all. */ + startTime?: string + endTime?: string +} + +export interface SessionReplayChunkIndexOutput { + readonly chunkSeq: number + /** + * Gateway receipt time — the chunk's position on the playback timeline. + * + * It trails the recording's own clock by the upload latency, which is well + * inside one chunk's duration, so it resolves a seek to the right chunk. The + * exact offset within that chunk comes from its rrweb events once loaded. + */ + readonly timestamp: string + readonly durationMs: number + readonly eventCount: number + readonly byteSize: number + readonly isCheckpoint: number +} + +/** + * Every chunk of a session EXCEPT its payload — the playback timeline and byte + * budget in one cheap read. + * + * This is what makes bounded playback possible: the player learns how many + * chunks exist, how big each one is, where the checkpoints (seek anchors) are, + * and which chunk covers a given moment — all without touching `Events`. On a + * MergeTree, omitting the wide column means its granules are never read, so + * this stays milliseconds even on a session whose payload is hundreds of MB. + */ +export function sessionReplayChunkIndexQuery(opts: SessionReplayChunkIndexOpts = {}) { + return from(SessionReplayEvents) + .select(($) => ({ + chunkSeq: $.ChunkSeq, + timestamp: $.Timestamp, + durationMs: $.DurationMs, + eventCount: $.EventCount, + byteSize: $.ByteSize, + isCheckpoint: $.IsCheckpoint, + })) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.SessionId.eq(param.string("sessionId")), + CH.when(opts.startTime, (v: string) => $.Timestamp.gte(v)), + CH.when(opts.endTime, (v: string) => $.Timestamp.lte(v)), + ]) + .orderBy(["chunkSeq", "asc"]) + .format("JSON") +} + export interface SessionReplayEventsOpts { /** Optional session time window — prunes daily partitions. Omit to scan all. */ startTime?: string endTime?: string + /** + * Inclusive chunk-sequence window. Callers get these from the chunk index and + * fetch a session in bounded slices — selecting `Events` for a whole session + * buffers the entire payload (p99 ~594 MB) into a 128 MB Worker. + */ + fromChunkSeq?: number + toChunkSeq?: number + /** Row cap — the last line of defence if the range is miscomputed. */ + limit?: number + /** Page within the range, for the public cursor-paginated surface. */ + offset?: number } export interface SessionReplayEventsOutput { @@ -661,7 +730,7 @@ export interface SessionReplayEventsOutput { } export function sessionReplayEventsQuery(opts: SessionReplayEventsOpts = {}) { - return from(SessionReplayEvents) + const query = from(SessionReplayEvents) .select(($) => ({ chunkSeq: $.ChunkSeq, timestamp: $.Timestamp, @@ -676,9 +745,15 @@ export function sessionReplayEventsQuery(opts: SessionReplayEventsOpts = {}) { $.SessionId.eq(param.string("sessionId")), CH.when(opts.startTime, (v: string) => $.Timestamp.gte(v)), CH.when(opts.endTime, (v: string) => $.Timestamp.lte(v)), + opts.fromChunkSeq === undefined ? undefined : $.ChunkSeq.gte(opts.fromChunkSeq), + opts.toChunkSeq === undefined ? undefined : $.ChunkSeq.lte(opts.toChunkSeq), ]) .orderBy(["chunkSeq", "asc"]) - .format("JSON") + // `ORDER BY chunkSeq ASC` makes the truncation deterministic: a clipped range + // loses the tail, never a hole in the middle. + if (opts.limit === undefined) return query.format("JSON") + const limited = query.limit(opts.limit) + return (opts.offset === undefined ? limited : limited.offset(opts.offset)).format("JSON") } // --------------------------------------------------------------------------- diff --git a/packages/query-engine/src/ch/tables.ts b/packages/query-engine/src/ch/tables.ts index a3fe4e225..3bcb94e01 100644 --- a/packages/query-engine/src/ch/tables.ts +++ b/packages/query-engine/src/ch/tables.ts @@ -570,6 +570,8 @@ export const SessionReplayEvents = table("session_replay_events", { OrgId: T.string, SessionId: T.string, ChunkSeq: T.uint32, + // Gateway receipt time — partitioning, TTL, and the anchor the chunk index + // resolves a seek target against. Timestamp: T.dateTime64, DurationMs: T.uint32, EventCount: T.uint32, diff --git a/packages/query-engine/src/execution/executor.test.ts b/packages/query-engine/src/execution/executor.test.ts index 01239fab9..bc8600147 100644 --- a/packages/query-engine/src/execution/executor.test.ts +++ b/packages/query-engine/src/execution/executor.test.ts @@ -806,6 +806,86 @@ describe("makeWarehouseExecutor raw response limits", () => { assert.strictEqual(error.code, "ResourceLimit") }), ) + + it.effect("propagates the limit error unchanged for a trusted bounded query", () => + Effect.gen(function* () { + // A trusted query's oversized response is OUR budget being exceeded, not + // the caller writing bad SQL — so it must not be restated as a + // validation error the way the raw path does above. + const executor = makeWarehouseExecutor({ + ...makeDeps([]), + createClient: () => ({ + sql: async () => { + throw new WarehouseResponseLimitError({ + kind: "bytes", + message: "response too large", + }) + }, + insert: async () => {}, + }), + }) + const error = yield* Effect.flip( + executor.compiledQueryBounded(tenant, compiled, { + context: "test", + responseLimits: { maxRows: 40, maxBytes: 8_000_000 }, + }), + ) + assert.instanceOf(error, WarehouseResponseLimitError) + }), + ) + + it.effect("runs an oversized bounded read exactly once instead of retrying it", () => + Effect.gen(function* () { + // The regression this whole change exists for: an over-budget response + // used to surface as an aborted subrequest, which the transient + // classifier read as a flaky upstream and retried — re-running a read + // that had already exhausted the heap, twice more, before 503ing. + let attempts = 0 + const executor = makeWarehouseExecutor({ + ...makeDeps([]), + createClient: () => ({ + sql: async () => { + attempts++ + throw new WarehouseResponseLimitError({ + kind: "bytes", + message: "response too large", + }) + }, + insert: async () => {}, + }), + }) + yield* Effect.flip( + executor.compiledQueryBounded(tenant, compiled, { + context: "test", + responseLimits: { maxRows: 40, maxBytes: 8_000_000 }, + }), + ) + assert.strictEqual(attempts, 1) + }), + ) + + it.effect("forwards the caller's budget to the driver", () => + Effect.gen(function* () { + let seen: unknown + const executor = makeWarehouseExecutor({ + ...makeDeps([]), + createClient: () => ({ + sql: async (_sql: string, options?: unknown) => { + seen = options + return { data: [] } + }, + insert: async () => {}, + }), + }) + yield* executor.compiledQueryBounded(tenant, compiled, { + context: "test", + responseLimits: { maxRows: 40, maxBytes: 8_000_000 }, + }) + assert.deepStrictEqual(seen, { + responseLimits: { maxRows: 40, maxBytes: 8_000_000 }, + }) + }), + ) }) describe("makeWarehouseExecutor client cache partitions", () => { diff --git a/packages/query-engine/src/execution/executor.ts b/packages/query-engine/src/execution/executor.ts index 8bd24fbb4..7295fd908 100644 --- a/packages/query-engine/src/execution/executor.ts +++ b/packages/query-engine/src/execution/executor.ts @@ -407,18 +407,27 @@ WHERE name = 'enable_full_text_index'`, ) const attemptTimeoutMs = clientTimeoutMs(options?.profile, settings?.maxExecutionTime) const retryAttempts = yield* Ref.make(0) + // A caller-supplied budget wins: a trusted query that knows its own response + // can blow the Worker heap (session replay's rrweb payloads) opts in + // explicitly. Raw SQL keeps its standing caps. const responseLimits = - execution === "raw" + options?.responseLimits ?? + (execution === "raw" ? { maxRows: MAX_RAW_SQL_RESULT_ROWS, maxBytes: MAX_RAW_SQL_RESULT_BYTES } - : undefined + : undefined) const queryAttempt = Effect.tryPromise({ try: () => client.sql(finalSql, responseLimits === undefined ? undefined : { responseLimits }), catch: (error) => error instanceof WarehouseResponseLimitError - ? new RawSqlValidationError({ - code: "ResourceLimit", - message: error.message, - }) + ? // Only raw SQL restates this as a validation error — there the + // oversized result IS the caller's query problem. For a trusted + // query the limit is ours, so it propagates unchanged and the + // caller maps it to a domain error. Either way it never becomes a + // WarehouseUpstreamError, so the transient retry loop skips it + // instead of re-running a read that already exhausted the heap. + execution === "raw" + ? new RawSqlValidationError({ code: "ResourceLimit", message: error.message }) + : error : // `execution` decides authorship: raw SQL comes from the caller (the // raw_sql widget, the `run_sql` MCP tool), so an analyzer complaint // about it is their typo and must keep the database's own message. @@ -529,6 +538,31 @@ WHERE name = 'enable_full_text_index'`, "maple.query.plan.full_text_setting": capabilities.fullTextSearchSetting, }) + // --- Response-limit narrowing ------------------------------------------ + // + // `executeSql` can raise WarehouseResponseLimitError, but only when a caller + // passed `responseLimits`. `compiledQueryBounded` is the one entry point that + // does; every other one strips the option (`withoutResponseLimits`) and then + // narrows the error away (`unbounded`). Types can't see that the strip makes + // the error unreachable, hence the explicit pair — and hence `Effect.die` + // rather than a mapping: if it ever fires, a caller reached the limit path + // without declaring it, which is a bug here and not a condition to handle. + + const withoutResponseLimits = (options?: SqlQueryOptions): SqlQueryOptions | undefined => { + if (options?.responseLimits === undefined) return options + const { responseLimits: _optedOut, ...rest } = options + return rest + } + + const unbounded = ( + effect: Effect.Effect, + ): Effect.Effect => + Effect.catchIf( + effect, + (error): error is WarehouseResponseLimitError => error instanceof WarehouseResponseLimitError, + (error) => Effect.die(error), + ) + const query = Effect.fn("WarehouseQueryService.query")(function* ( tenant: ExecutionTenant, payload: WarehouseQueryRequest, @@ -696,13 +730,29 @@ WHERE name = 'enable_full_text_index'`, tenant: ExecutionTenant, compiled: CompiledQuery | ((capabilities: WarehouseCapabilities) => CompiledQuery), options?: SqlQueryOptions, + ) => unbounded(executeCompiledQuery(tenant, compiled, withoutResponseLimits(options))) + + /** + * Read with an explicit ceiling on the response we're willing to materialize. + * + * The limit error propagates to the caller instead of being retried: it is a + * statement about the size of the answer, so re-running the query produces + * the same oversized response. Callers map it to a domain error that tells + * the user to ask for less. + */ + const compiledQueryBounded = ( + tenant: ExecutionTenant, + compiled: CompiledQuery, + options: SqlQueryOptions & { + readonly responseLimits: { readonly maxRows: number; readonly maxBytes: number } + }, ) => executeCompiledQuery(tenant, compiled, options) const compiledQueryWithCapabilities = ( tenant: ExecutionTenant, compile: (capabilities: WarehouseCapabilities) => CompiledQuery, options?: SqlQueryOptions, - ) => executeCompiledQuery(tenant, compile, options) + ) => unbounded(executeCompiledQuery(tenant, compile, withoutResponseLimits(options))) /** * Deliberately read across every tenant. @@ -882,25 +932,34 @@ WHERE name = 'enable_full_text_index'`, const asExecutor = (tenant: ExecutionTenant): WarehouseExecutorShape => ({ orgId: tenant.orgId, query: (pipe: WarehouseQueryName, params: Record, options?: SqlQueryOptions) => - query(tenant, { pipeName: pipe, params }, { context: `pipe:${pipe}`, ...options }).pipe( - Effect.map((response) => ({ data: response.data as unknown as ReadonlyArray })), - ), + unbounded( + query( + tenant, + { pipeName: pipe, params }, + { context: `pipe:${pipe}`, ...withoutResponseLimits(options) }, + ), + ).pipe(Effect.map((response) => ({ data: response.data as unknown as ReadonlyArray }))), compiledQuery: (compiled: CompiledQuery, options?: SqlQueryOptions) => compiledQuery(tenant, compiled, { context: "warehouseExecutor.compiledQuery", ...options }), compiledQueryFirst: (compiled: CompiledQuery, options?: SqlQueryOptions) => - compiledQueryFirst(tenant, compiled, { - context: "warehouseExecutor.compiledQueryFirst", - ...options, - }), + unbounded( + compiledQueryFirst(tenant, compiled, { + context: "warehouseExecutor.compiledQueryFirst", + ...withoutResponseLimits(options), + }), + ), }) return { - query, - crossOrgQuery, - rawSqlQuery, + query: (tenant, payload, options) => unbounded(query(tenant, payload, withoutResponseLimits(options))), + crossOrgQuery: (tenant, compiled, options) => + unbounded(crossOrgQuery(tenant, compiled, { ...withoutResponseLimits(options), justification: options.justification })), + rawSqlQuery: (tenant, sql, options) => unbounded(rawSqlQuery(tenant, sql, options)), compiledQuery, + compiledQueryBounded, compiledQueryWithCapabilities, - compiledQueryFirst, + compiledQueryFirst: (tenant, compiled, options) => + unbounded(compiledQueryFirst(tenant, compiled, withoutResponseLimits(options))), ingest, asExecutor, } satisfies WarehouseQueryServiceShape diff --git a/packages/query-engine/src/execution/ports.ts b/packages/query-engine/src/execution/ports.ts index a5d1725d0..a856081cb 100644 --- a/packages/query-engine/src/execution/ports.ts +++ b/packages/query-engine/src/execution/ports.ts @@ -13,6 +13,7 @@ import type { WarehouseCapabilities } from "../capabilities" import type { WarehouseExecutorShape } from "../observability" import type { SqlQueryOptions } from "../profiles" import type { WarehouseSqlError } from "./errors" +import type { WarehouseResponseLimitError } from "./response-limits" /** The minimal tenant surface the executor reads (org scope + identity for spans). */ export interface ExecutionTenant { @@ -115,6 +116,24 @@ export interface WarehouseQueryServiceShape { compiled: CompiledQuery | ((capabilities: WarehouseCapabilities) => CompiledQuery), options?: SqlQueryOptions, ) => Effect.Effect, WarehouseSqlError | WarehouseValidationError> + /** + * `compiledQuery` with an explicit ceiling on how much of the response we are + * willing to materialize, failing with `WarehouseResponseLimitError` past it. + * + * Separate from `compiledQuery` on purpose: the extra failure mode belongs in + * the signature of the handful of call sites that can actually hit it, not in + * the error union of the ~30 endpoints that cannot. + */ + readonly compiledQueryBounded: ( + tenant: ExecutionTenant, + compiled: CompiledQuery, + options: SqlQueryOptions & { + readonly responseLimits: { readonly maxRows: number; readonly maxBytes: number } + }, + ) => Effect.Effect< + ReadonlyArray, + WarehouseSqlError | WarehouseValidationError | WarehouseResponseLimitError + > readonly compiledQueryWithCapabilities: ( tenant: ExecutionTenant, compile: (capabilities: WarehouseCapabilities) => CompiledQuery, diff --git a/packages/query-engine/src/profiles/query-profile.ts b/packages/query-engine/src/profiles/query-profile.ts index 09eda34c2..d48c0a481 100644 --- a/packages/query-engine/src/profiles/query-profile.ts +++ b/packages/query-engine/src/profiles/query-profile.ts @@ -84,6 +84,18 @@ export type SqlQueryOptions = WarehouseQueryOptions & { * reads of gateway-written data gated on write-readiness). */ route?: "ingest" + /** + * Abort the read once the encoded response crosses these bounds, failing with + * `WarehouseResponseLimitError` instead of buffering the rest. + * + * ClickHouse settings cap what the *warehouse* spends; this caps what *we* + * are willing to materialize in a 128 MB Worker. Set it for queries whose + * result size is driven by user data rather than by the query shape — without + * it an oversized response dies as a platform abort, which the transient + * classifier reads as a flaky upstream and retries. Reach for it via + * `compiledQueryBounded`, which surfaces the error in its signature. + */ + responseLimits?: { readonly maxRows: number; readonly maxBytes: number } } /**