From b37326939b078101773f47c78aedeba281452e5a Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 4 Aug 2026 18:07:20 +0200 Subject: [PATCH 1/3] feat(replays): manifest + bounded chunk ranges for playback Opening a replay often failed with 503 "Database is temporarily unavailable". Nothing was unavailable: getReplayEvents selected Events for every chunk of a session with no LIMIT and buffered it into a 128 MB Worker. Ingest allows 1 GiB per session (p99 ~594 MB), so the abort matched the transient-upstream rule and was retried 3x before failing. Prod spans ran 428ms to 12.6s against a 15s cap. Playback now reads the way other replay products do: a cheap manifest of chunk positions and sizes, then payloads a bounded range at a time, seeded from a checkpoint. Both endpoints are v2-only; the v1 payload endpoint is deleted so the unbounded read cannot come back. --- apps/api/src/routes/v1/session-replay.http.ts | 43 +-- .../routes/v1/session-replay.schema.test.ts | 1 + .../routes/v2/phase1-resources.http.test.ts | 189 ++++++++++++ .../api/src/routes/v2/session-replays.http.ts | 183 +++++++++-- apps/api/src/routes/v2/v2-test-support.ts | 1 + .../warehouse/WarehouseQueryService.ts | 14 + apps/web/src/api/warehouse/effect-utils.ts | 32 +- apps/web/src/api/warehouse/replays.ts | 92 +++++- .../replays/replay-player-context.tsx | 162 +++++++--- .../replays/replay-progressive-load.test.ts | 173 +++++++++++ .../components/replays/replay-range.test.ts | 206 +++++++++++++ .../src/components/replays/replay-range.ts | 203 ++++++++++++ .../replays/use-replay-chunk-loader.ts | 289 ++++++++++++++++++ .../services/atoms/warehouse-query-atoms.ts | 19 +- apps/web/src/routes/replays/$sessionId.tsx | 8 +- packages/domain/src/http/session-replay.ts | 38 +-- packages/domain/src/http/v2/errors.ts | 45 +++ packages/domain/src/http/v2/openapi.test.ts | 1 + .../domain/src/http/v2/session-replays.ts | 126 +++++++- packages/domain/src/tinybird/datasources.ts | 3 + .../src/__sql_baseline__/catalog.sql | 36 +++ .../query-engine/src/ch/builder-fixtures.ts | 28 ++ packages/query-engine/src/ch/index.ts | 4 + .../src/ch/queries/session-replays.test.ts | 51 ++++ .../src/ch/queries/session-replays.ts | 81 ++++- packages/query-engine/src/ch/tables.ts | 2 + .../src/execution/executor.test.ts | 80 +++++ .../query-engine/src/execution/executor.ts | 95 ++++-- packages/query-engine/src/execution/ports.ts | 19 ++ .../src/profiles/query-profile.ts | 12 + 30 files changed, 2054 insertions(+), 182 deletions(-) create mode 100644 apps/web/src/components/replays/replay-progressive-load.test.ts create mode 100644 apps/web/src/components/replays/replay-range.test.ts create mode 100644 apps/web/src/components/replays/replay-range.ts create mode 100644 apps/web/src/components/replays/use-replay-chunk-loader.ts 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..c55453ec1 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,189 @@ 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("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..35f5f0cc1 100644 --- a/apps/api/src/routes/v2/session-replays.http.ts +++ b/apps/api/src/routes/v2/session-replays.http.ts @@ -1,19 +1,27 @@ 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, 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 +37,45 @@ const decodeTraceId = Schema.decodeSync(TraceId) /** Warehouse errors → the proper v2 envelope (400/429/502/503 per tag). */ const mapWarehouseError = warehouseToV2("session_replay_query") +/** + * As above, 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. + */ +/** + * 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", + ), + ) +} + +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 +258,135 @@ 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. + const page = yield* paginateOffsetQuery(query, ({ 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..9781b7187 --- /dev/null +++ b/apps/web/src/components/replays/replay-progressive-load.test.ts @@ -0,0 +1,173 @@ +import { describe, expect, it } from "vitest" +import { + INITIAL_WINDOW_BYTES, + RANGE_SIZE, + type ReplayChunkMeta, + type ReplayRange, + checkpointAtOrBefore, + chunkAtOffset, + initialRanges, + 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)) + +/** + * 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)] + 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(loadedThrough() + 1) + if (!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) + const bytes = ranges.length * RANGE_SIZE * 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)).toEqual(initialRanges(manifest)) + }) +}) + +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(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)) { + 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, 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("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(Math.ceil(manifest.length / RANGE_SIZE)) + for (const r of fetched) { + expect(r.toChunkSeq - r.fromChunkSeq + 1).toBe(RANGE_SIZE) + } + }) +}) 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..4152d1425 --- /dev/null +++ b/apps/web/src/components/replays/replay-range.test.ts @@ -0,0 +1,206 @@ +import { describe, expect, it } from "vitest" +import { + INITIAL_WINDOW_BYTES, + RANGE_SIZE, + type ReplayChunkMeta, + alignRangeStart, + checkpointAtOrBefore, + chunkAtOffset, + chunkStartMs, + initialRanges, + 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 grid alignment", () => { + it("snaps any chunk in a slot to the same range", () => { + // The cache-correctness property: without it the playhead mints a new atom + // key every tick and scrubbing back refetches everything. + const ranges = [0, 1, 7, 15].map((seq) => rangeContaining(seq)) + for (const range of ranges) { + expect(range).toEqual({ fromChunkSeq: 0, toChunkSeq: RANGE_SIZE - 1 }) + } + expect(rangeContaining(RANGE_SIZE)).toEqual({ + fromChunkSeq: RANGE_SIZE, + toChunkSeq: RANGE_SIZE * 2 - 1, + }) + }) + + it("produces one identical atom key for every position inside a slot", () => { + // Keys are JSON.stringify of the input, so this also pins property order. + const window = { windowStart: "2026-08-04 10:00:00", windowEnd: "2026-08-04 11:00:00" } + const keys = new Set( + [0, 3, 9, 15].map((seq) => + JSON.stringify(replayRangeInput("sess_1", window, rangeContaining(seq))), + ), + ) + expect(keys.size).toBe(1) + }) + + it("keeps ranges within the server's per-request chunk cap", () => { + expect(RANGE_SIZE).toBeLessThanOrEqual(40) + }) + + it("aligns down, never up — chunk 0 must not land in a negative slot", () => { + expect(alignRangeStart(0)).toBe(0) + }) +}) + +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)) + const bytes = ranges.length * RANGE_SIZE * 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 }))) + 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)[0]).toEqual(rangeContaining(20)) + }) + + it("returns nothing for an empty manifest", () => { + expect(initialRanges([])).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))).toEqual([rangeContaining(0)]) + }) +}) + +describe("rangesCovering", () => { + it("always yields at least one range, even past the budget", () => { + expect(rangesCovering(manifest(4), 0, 0)).toHaveLength(1) + }) + + it("returns nothing when there is nothing to cover", () => { + expect(rangesCovering([], 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)[0]).toEqual(rangeContaining(0)) + }) + + it("handles a single-chunk session", () => { + const chunks = manifest(1) + expect(initialRanges(chunks)).toEqual([rangeContaining(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)[0]).toEqual(rangeContaining(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..29779b855 --- /dev/null +++ b/apps/web/src/components/replays/replay-range.ts @@ -0,0 +1,203 @@ +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. + */ + +/** + * Chunks per fetched range. + * + * Ranges are aligned to a fixed grid of this size, which is the single load- + * bearing decision here: an unaligned range starting at the playhead would mint + * a new atom-family key on every tick, so nothing would ever be served from + * cache and scrubbing back would refetch. Aligned, the same stretch of the + * recording always resolves to the same key. + * + * 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 RANGE_SIZE = 16 + +/** + * 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 +} + +/** Snap a chunk sequence down to its grid slot. */ +export const alignRangeStart = (chunkSeq: number) => Math.floor(chunkSeq / RANGE_SIZE) * RANGE_SIZE + +/** The grid range containing `chunkSeq`. */ +export const rangeContaining = (chunkSeq: number): ReplayRange => { + const fromChunkSeq = alignRangeStart(chunkSeq) + return { fromChunkSeq, toChunkSeq: fromChunkSeq + RANGE_SIZE - 1 } +} + +/** 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): ReadonlyArray => { + if (chunks.length === 0) return [] + const seed = chunks.find((chunk) => chunk.is_checkpoint) ?? chunks[0]! + return rangesCovering(chunks, seed.chunk_seq, INITIAL_WINDOW_BYTES) +} + +/** + * Grid ranges starting at `fromChunkSeq`, extended until `byteBudget` is met. + * + * Always returns at least one range, so a single chunk larger than the whole + * budget still loads instead of stalling playback forever. + */ +export const rangesCovering = ( + chunks: ReadonlyArray, + fromChunkSeq: number, + byteBudget: number, +): ReadonlyArray => { + const ranges: Array = [] + let bytes = 0 + let cursor = fromChunkSeq + const lastSeq = chunks[chunks.length - 1]?.chunk_seq + if (lastSeq === undefined) return [] + while (cursor <= lastSeq) { + const range = rangeContaining(cursor) + if (ranges.some((existing) => existing.fromChunkSeq === range.fromChunkSeq)) break + ranges.push(range) + for (const chunk of chunks) { + if (chunk.chunk_seq < range.fromChunkSeq || chunk.chunk_seq > range.toChunkSeq) continue + bytes += chunk.byte_size + } + if (bytes >= byteBudget) break + cursor = range.toChunkSeq + 1 + } + 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..11e603c99 --- /dev/null +++ b/apps/web/src/components/replays/use-replay-chunk-loader.ts @@ -0,0 +1,289 @@ +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, + PREFETCH_AHEAD_MS, + RANGE_SIZE, + type ReplayChunkMeta, + type ReplayRange, + chunkAtOffset, + chunkStartMs, + checkpointAtOrBefore, + initialRanges, + 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]) + + // 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)) + }, [enabled, chunks, 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 a + // placeholder would fire a real request for a chunk window we have no reason + // to believe exists, so hold at the first grid slot: that is where playback + // starts for most sessions, making it a prefetch rather than a wasted call. + const subscribedRange = + activeRange ?? loaded[loaded.length - 1]?.range ?? { fromChunkSeq: 0, toChunkSeq: RANGE_SIZE - 1 } + 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 rows = committed.chunks as ReadonlyArray<{ chunk_seq: number; events: string }> + const events = decodeRange(rows) + const lastChunkSeq = rows.reduce((max, row) => Math.max(max, row.chunk_seq), activeRange.fromChunkSeq) + setLoaded((current) => [...current, { key, range: activeRange, events, lastChunkSeq }]) + setQueue((current) => current.slice(1)) + }, [rangeResult, activeRange, loadedKeys]) + + // 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(loadedThroughSeq + 1) + if (loadedKeys.has(rangeKey(next))) return current + return [next] + }) + }, [hasMoreChunks, loadedThroughSeq, loadedKeys]) + + 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, anchor.chunk_seq, INITIAL_WINDOW_BYTES)) + return true + }, + [chunks, 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 } } /** From 3ea4afbe6398372f12936703230890abc6e5c293 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 4 Aug 2026 18:17:53 +0200 Subject: [PATCH 2/3] fix(replays): size chunk ranges by bytes, not a fixed chunk count MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A fixed 16-chunk range assumed uniform chunk sizes. They are not: a flush is ~100 KB but a full DOM snapshot runs far larger, and real sessions carry 838 KB chunks — 16 of those is ~13 MB against an 8 MB response ceiling, so the request comes back 413 and that stretch of the recording cannot be loaded at all. Ranges are now planned from the manifest's byte sizes. Walking greedily from the first chunk keeps boundaries stable as a live session appends, which is what the atom cache keys on. --- .../routes/v2/phase1-resources.http.test.ts | 52 ++++++++ .../replays/replay-progressive-load.test.ts | 26 ++-- .../components/replays/replay-range.test.ts | 100 +++++++++----- .../src/components/replays/replay-range.ts | 123 +++++++++++++----- .../replays/use-replay-chunk-loader.ts | 28 ++-- 5 files changed, 243 insertions(+), 86 deletions(-) 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 c55453ec1..a01cb26ae 100644 --- a/apps/api/src/routes/v2/phase1-resources.http.test.ts +++ b/apps/api/src/routes/v2/phase1-resources.http.test.ts @@ -1245,6 +1245,58 @@ describe("v2 session_replays over HTTP", () => { 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 diff --git a/apps/web/src/components/replays/replay-progressive-load.test.ts b/apps/web/src/components/replays/replay-progressive-load.test.ts index 9781b7187..31ff03370 100644 --- a/apps/web/src/components/replays/replay-progressive-load.test.ts +++ b/apps/web/src/components/replays/replay-progressive-load.test.ts @@ -1,12 +1,13 @@ import { describe, expect, it } from "vitest" import { INITIAL_WINDOW_BYTES, - RANGE_SIZE, + MAX_CHUNKS_PER_RANGE, type ReplayChunkMeta, type ReplayRange, checkpointAtOrBefore, chunkAtOffset, initialRanges, + planRanges, rangeContaining, rangeKey, rangesCovering, @@ -38,6 +39,7 @@ const chunk = (seq: number, overrides: Partial = {}): ReplayChu /** 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, @@ -45,7 +47,7 @@ const manifest = Array.from({ length: 100 }, (_, seq) => chunk(seq)) * pull the next grid range. */ const simulate = (chunks: ReadonlyArray, throughMs: number) => { - const fetched: Array = [...initialRanges(chunks)] + 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 @@ -53,8 +55,8 @@ const simulate = (chunks: ReadonlyArray, throughMs: number) => while (playhead <= throughMs) { const loadedUntilMs = (loadedThrough() + 1) * CHUNK_MS if (playhead + 20_000 >= loadedUntilMs && loadedThrough() < lastSeq) { - const next = rangeContaining(loadedThrough() + 1) - if (!fetched.some((r) => rangeKey(r) === rangeKey(next))) fetched.push(next) + 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 } @@ -63,8 +65,8 @@ const simulate = (chunks: ReadonlyArray, throughMs: number) => describe("opening window", () => { it("starts on a couple of MB, not the whole session", () => { - const ranges = initialRanges(manifest) - const bytes = ranges.length * RANGE_SIZE * 100_000 + 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) @@ -73,7 +75,7 @@ describe("opening window", () => { it("is independent of session length", () => { const long = Array.from({ length: 2000 }, (_, seq) => chunk(seq)) - expect(initialRanges(long)).toEqual(initialRanges(manifest)) + expect(initialRanges(long, planRanges(long))).toEqual(initialRanges(manifest, plan)) }) }) @@ -105,7 +107,7 @@ describe("playing forward", () => { 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(chunkAtOffset(manifest, 12 * CHUNK_MS)!.chunk_seq) + const back = rangeContaining(plan, chunkAtOffset(manifest, 12 * CHUNK_MS)!.chunk_seq)! expect(forward).toContain(rangeKey(back)) }) }) @@ -113,7 +115,7 @@ describe("playing forward", () => { describe("seeking", () => { const loadedAfterOpening = () => { const covered = new Set() - for (const r of initialRanges(manifest)) { + for (const r of initialRanges(manifest, plan)) { for (let s = r.fromChunkSeq; s <= r.toChunkSeq; s++) covered.add(s) } return covered @@ -138,7 +140,7 @@ describe("seeking", () => { 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, anchor.chunk_seq, INITIAL_WINDOW_BYTES) + 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) @@ -165,9 +167,9 @@ describe("request volume", () => { // 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(Math.ceil(manifest.length / RANGE_SIZE)) + expect(fetched.length).toBe(plan.length) for (const r of fetched) { - expect(r.toChunkSeq - r.fromChunkSeq + 1).toBe(RANGE_SIZE) + 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 index 4152d1425..2c1fff24b 100644 --- a/apps/web/src/components/replays/replay-range.test.ts +++ b/apps/web/src/components/replays/replay-range.test.ts @@ -1,13 +1,14 @@ import { describe, expect, it } from "vitest" import { INITIAL_WINDOW_BYTES, - RANGE_SIZE, + MAX_BYTES_PER_RANGE, + MAX_CHUNKS_PER_RANGE, type ReplayChunkMeta, - alignRangeStart, checkpointAtOrBefore, chunkAtOffset, chunkStartMs, initialRanges, + planRanges, manifestDurationMs, rangeContaining, rangesCovering, @@ -30,37 +31,72 @@ const chunk = (seq: number, overrides: Partial = {}): ReplayChu const manifest = (count: number, overrides: (seq: number) => Partial = () => ({})) => Array.from({ length: count }, (_, seq) => chunk(seq, overrides(seq))) -describe("range grid alignment", () => { - it("snaps any chunk in a slot to the same range", () => { - // The cache-correctness property: without it the playhead mints a new atom - // key every tick and scrubbing back refetches everything. - const ranges = [0, 1, 7, 15].map((seq) => rangeContaining(seq)) - for (const range of ranges) { - expect(range).toEqual({ fromChunkSeq: 0, toChunkSeq: RANGE_SIZE - 1 }) +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) } - expect(rangeContaining(RANGE_SIZE)).toEqual({ - fromChunkSeq: RANGE_SIZE, - toChunkSeq: RANGE_SIZE * 2 - 1, - }) }) - it("produces one identical atom key for every position inside a slot", () => { - // Keys are JSON.stringify of the input, so this also pins property order. + 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( - [0, 3, 9, 15].map((seq) => - JSON.stringify(replayRangeInput("sess_1", window, rangeContaining(seq))), + [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("keeps ranges within the server's per-request chunk cap", () => { - expect(RANGE_SIZE).toBeLessThanOrEqual(40) + it("returns nothing for a chunk past the end of the plan", () => { + expect(rangeContaining(planRanges(manifest(4)), 999)).toBeUndefined() }) - it("aligns down, never up — chunk 0 must not land in a negative slot", () => { - expect(alignRangeStart(0)).toBe(0) + it("keeps ranges within the server's per-request chunk cap", () => { + expect(MAX_CHUNKS_PER_RANGE).toBeLessThanOrEqual(40) }) }) @@ -68,8 +104,8 @@ 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)) - const bytes = ranges.length * RANGE_SIZE * 100_000 + 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) }) @@ -77,7 +113,7 @@ describe("initialRanges", () => { 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 }))) + const ranges = initialRanges(manifest(64, () => ({ byte_size: 4_000_000 })), planRanges(manifest(64, () => ({ byte_size: 4_000_000 })))) expect(ranges).toHaveLength(1) }) @@ -85,27 +121,27 @@ describe("initialRanges", () => { // 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)[0]).toEqual(rangeContaining(20)) + expect(initialRanges(chunks, planRanges(chunks))[0]).toEqual(rangeContaining(planRanges(chunks), 20)) }) it("returns nothing for an empty manifest", () => { - expect(initialRanges([])).toEqual([]) + 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))).toEqual([rangeContaining(0)]) + 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), 0, 0)).toHaveLength(1) + expect(rangesCovering(manifest(4), planRanges(manifest(4)), 0, 0)).toHaveLength(1) }) it("returns nothing when there is nothing to cover", () => { - expect(rangesCovering([], 0, INITIAL_WINDOW_BYTES)).toEqual([]) + expect(rangesCovering([], planRanges([]), 0, INITIAL_WINDOW_BYTES)).toEqual([]) }) }) @@ -168,12 +204,12 @@ describe("sessions recorded before this change", () => { // pick an anchor rather than refusing. const chunks = manifest(8, () => ({ is_checkpoint: false })) expect(checkpointAtOrBefore(chunks, 5)?.chunk_seq).toBe(0) - expect(initialRanges(chunks)[0]).toEqual(rangeContaining(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)).toEqual([rangeContaining(0)]) + expect(initialRanges(chunks, planRanges(chunks))).toEqual([{ fromChunkSeq: 0, toChunkSeq: 0 }]) expect(chunkAtOffset(chunks, 0)?.chunk_seq).toBe(0) expect(manifestDurationMs(chunks)).toBe(5_000) }) @@ -182,7 +218,7 @@ describe("sessions recorded before this change", () => { // 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)[0]).toEqual(rangeContaining(137)) + expect(initialRanges(chunks, planRanges(chunks))[0]).toEqual(rangeContaining(planRanges(chunks), 137)) expect(checkpointAtOrBefore(chunks, 150)?.chunk_seq).toBe(137) }) }) diff --git a/apps/web/src/components/replays/replay-range.ts b/apps/web/src/components/replays/replay-range.ts index 29779b855..0d8b597ef 100644 --- a/apps/web/src/components/replays/replay-range.ts +++ b/apps/web/src/components/replays/replay-range.ts @@ -10,18 +10,20 @@ import type { ReplayPartitionWindow } from "./replay-format" */ /** - * Chunks per fetched range. + * Upper bound on chunks per fetched range. * - * Ranges are aligned to a fixed grid of this size, which is the single load- - * bearing decision here: an unaligned range starting at the playhead would mint - * a new atom-family key on every tick, so nothing would ever be served from - * cache and scrubbing back would refetch. Aligned, the same stretch of the - * recording always resolves to the same key. - * - * Must stay <= the server's `max_chunks_per_request` (40); the manifest echoes - * the real cap so a mismatch is detectable rather than silent. + * 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 RANGE_SIZE = 16 +export const MAX_BYTES_PER_RANGE = 8_000_000 /** * Bytes to load before starting playback. @@ -49,15 +51,61 @@ export interface ReplayRange { readonly toChunkSeq: number } -/** Snap a chunk sequence down to its grid slot. */ -export const alignRangeStart = (chunkSeq: number) => Math.floor(chunkSeq / RANGE_SIZE) * RANGE_SIZE - -/** The grid range containing `chunkSeq`. */ -export const rangeContaining = (chunkSeq: number): ReplayRange => { - const fromChunkSeq = alignRangeStart(chunkSeq) - return { fromChunkSeq, toChunkSeq: fromChunkSeq + RANGE_SIZE - 1 } +/** + * 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}` @@ -166,38 +214,47 @@ export const checkpointAtOrBefore = ( * 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): ReadonlyArray => { +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, seed.chunk_seq, INITIAL_WINDOW_BYTES) + return rangesCovering(chunks, plan, seed.chunk_seq, INITIAL_WINDOW_BYTES) } /** - * Grid ranges starting at `fromChunkSeq`, extended until `byteBudget` is met. + * Planned ranges from the one containing `fromChunkSeq`, extended until + * `byteBudget` is met. * - * Always returns at least one range, so a single chunk larger than the whole - * budget still loads instead of stalling playback forever. + * 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 - let cursor = fromChunkSeq - const lastSeq = chunks[chunks.length - 1]?.chunk_seq - if (lastSeq === undefined) return [] - while (cursor <= lastSeq) { - const range = rangeContaining(cursor) - if (ranges.some((existing) => existing.fromChunkSeq === range.fromChunkSeq)) break + for (let i = startIndex; i < plan.length; i++) { + const range = plan[i]! ranges.push(range) - for (const chunk of chunks) { - if (chunk.chunk_seq < range.fromChunkSeq || chunk.chunk_seq > range.toChunkSeq) continue - bytes += chunk.byte_size - } + bytes += bytesIn(range) if (bytes >= byteBudget) break - cursor = range.toChunkSeq + 1 } 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 index 11e603c99..c0a77f783 100644 --- a/apps/web/src/components/replays/use-replay-chunk-loader.ts +++ b/apps/web/src/components/replays/use-replay-chunk-loader.ts @@ -5,14 +5,16 @@ 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, - RANGE_SIZE, type ReplayChunkMeta, type ReplayRange, chunkAtOffset, chunkStartMs, checkpointAtOrBefore, initialRanges, + planRanges, manifestStartMs, rangeContaining, rangeKey, @@ -107,6 +109,14 @@ export function useReplayChunkLoader({ 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. // @@ -123,8 +133,8 @@ export function useReplayChunkLoader({ setSeedCount(0) setIsSeeking(false) setSeekTargetMs(null) - setQueue(initialRanges(chunks)) - }, [enabled, chunks, sessionId]) + setQueue(initialRanges(chunks, plan)) + }, [enabled, chunks, plan, sessionId]) const activeRange = queue[0] const loadedKeys = React.useMemo(() => new Set(loaded.map((entry) => entry.key)), [loaded]) @@ -138,7 +148,7 @@ export function useReplayChunkLoader({ // to believe exists, so hold at the first grid slot: that is where playback // starts for most sessions, making it a prefetch rather than a wasted call. const subscribedRange = - activeRange ?? loaded[loaded.length - 1]?.range ?? { fromChunkSeq: 0, toChunkSeq: RANGE_SIZE - 1 } + activeRange ?? loaded[loaded.length - 1]?.range ?? plan[0] ?? { fromChunkSeq: 0, toChunkSeq: 0 } const rangeResult = useAtomValue( getReplayEventsResultAtom({ data: replayRangeInput(sessionId, window, subscribedRange) }), ) @@ -215,11 +225,11 @@ export function useReplayChunkLoader({ if (!hasMoreChunks) return setQueue((current) => { if (current.length > 0) return current - const next = rangeContaining(loadedThroughSeq + 1) - if (loadedKeys.has(rangeKey(next))) return current + const next = rangeContaining(plan, loadedThroughSeq + 1) + if (next === undefined || loadedKeys.has(rangeKey(next))) return current return [next] }) - }, [hasMoreChunks, loadedThroughSeq, loadedKeys]) + }, [hasMoreChunks, loadedThroughSeq, loadedKeys, plan]) const reportProgress = React.useCallback( (offsetMs: number) => { @@ -262,10 +272,10 @@ export function useReplayChunkLoader({ setSeedCount(0) setIsSeeking(true) setSeekTargetMs(offsetMs) - setQueue(rangesCovering(chunks, anchor.chunk_seq, INITIAL_WINDOW_BYTES)) + setQueue(rangesCovering(chunks, plan, anchor.chunk_seq, INITIAL_WINDOW_BYTES)) return true }, - [chunks, loaded, loadedThroughSeq], + [chunks, plan, loaded, loadedThroughSeq], ) const bufferState: ReplayBufferState = isSeeking From c1f2173b7829f6917bc3e38c7ea02b0f92036f39 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 4 Aug 2026 18:32:40 +0200 Subject: [PATCH 3/3] fix(replays): drop overlapping chunks and clamp the page size MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects found reviewing the combined change. A live session's plan has a partial trailing range, so a range loaded as 40-45 becomes 40-55 once more chunks land. That is a different cache key, so it is fetched again — and chunks 40-45 were handed to addEvent a second time, replaying those mutations into the live engine. Commits now drop chunks already loaded, and anything outside the range being committed. Separately, limit=100 is inside the public 1-100 range but above the per-request chunk cap. The lookahead asked for 101 rows, got the SQL cap of 41, and read 41 <= 100 as "no more pages" — a short page reported as complete, silently dropping the rest of the recording. The page size is clamped to the same cap the SQL enforces. --- .../routes/v2/phase1-resources.http.test.ts | 32 +++++++++++++++++++ .../api/src/routes/v2/session-replays.http.ts | 29 +++++++++++------ .../replays/replay-progressive-load.test.ts | 27 ++++++++++++++++ .../replays/use-replay-chunk-loader.ts | 32 +++++++++++++++---- 4 files changed, 104 insertions(+), 16 deletions(-) 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 a01cb26ae..d1b7aeb91 100644 --- a/apps/api/src/routes/v2/phase1-resources.http.test.ts +++ b/apps/api/src/routes/v2/phase1-resources.http.test.ts @@ -1245,6 +1245,38 @@ describe("v2 session_replays over HTTP", () => { 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 diff --git a/apps/api/src/routes/v2/session-replays.http.ts b/apps/api/src/routes/v2/session-replays.http.ts index 35f5f0cc1..58c445616 100644 --- a/apps/api/src/routes/v2/session-replays.http.ts +++ b/apps/api/src/routes/v2/session-replays.http.ts @@ -4,6 +4,7 @@ import { MAX_REPLAY_CHUNKS_PER_REQUEST, MAX_REPLAY_EVENTS_RESPONSE_BYTES, MAX_REPLAY_MANIFEST_CHUNKS, + LIST_LIMIT_DEFAULT, MapleApiV2, dependencyUnavailable, invalidRequest, @@ -37,14 +38,6 @@ const decodeTraceId = Schema.decodeSync(TraceId) /** Warehouse errors → the proper v2 envelope (400/429/502/503 per tag). */ const mapWarehouseError = warehouseToV2("session_replay_query") -/** - * As above, 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. - */ /** * Refuse a chunk range whose payload would blow the response budget, before a * byte of it is fetched. @@ -68,6 +61,14 @@ const assertRangeFitsBudget = (rows: ReadonlyArray<{ readonly byteSize: number } ) } +/** + * 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( @@ -333,7 +334,17 @@ const HttpV2SessionReplaysGroup = HttpApiBuilder.group(MapleApiV2, "sessionRepla // 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. - const page = yield* paginateOffsetQuery(query, ({ limit, offset }) => { + // + // 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, diff --git a/apps/web/src/components/replays/replay-progressive-load.test.ts b/apps/web/src/components/replays/replay-progressive-load.test.ts index 31ff03370..0799c30ae 100644 --- a/apps/web/src/components/replays/replay-progressive-load.test.ts +++ b/apps/web/src/components/replays/replay-progressive-load.test.ts @@ -162,6 +162,33 @@ describe("seeking", () => { }) }) +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. diff --git a/apps/web/src/components/replays/use-replay-chunk-loader.ts b/apps/web/src/components/replays/use-replay-chunk-loader.ts index c0a77f783..abbc64d44 100644 --- a/apps/web/src/components/replays/use-replay-chunk-loader.ts +++ b/apps/web/src/components/replays/use-replay-chunk-loader.ts @@ -143,10 +143,10 @@ export function useReplayChunkLoader({ // 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 a - // placeholder would fire a real request for a chunk window we have no reason - // to believe exists, so hold at the first grid slot: that is where playback - // starts for most sessions, making it a prefetch rather than a wasted call. + // 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( @@ -175,12 +175,30 @@ export function useReplayChunkLoader({ .onSuccess((result) => result) .orElse(() => null) if (committed === null) return - const rows = committed.chunks as ReadonlyArray<{ chunk_seq: number; events: string }> + 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), activeRange.fromChunkSeq) + 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]) + }, [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.