From 280705e9dc262961b08302855161bf7c06670952 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 14:19:10 +0000 Subject: [PATCH 1/2] feat(replays): store rrweb chunk payloads in R2, not ClickHouse MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `session_replay_events` held the raw rrweb payload as an opaque `Events` String. Nothing ever queried inside it — the only reader selects the whole column for one (OrgId, SessionId) and orders by ChunkSeq — so it was a key/value blob fetch paying warehouse storage prices, and it is the bulk of a session's bytes. The ingest gateway now PUTs the browser's gzip verbatim to an S3-compatible bucket and writes a thin index row with an empty `Events`; the API refills the payload on read, so the HTTP response shape is unchanged. `session_events` — the distilled structured stream — deliberately stays in ClickHouse. It is queried across sessions (the event-match semi-join and the active/idle window function), which is what a column store is for. No schema change. An empty `Events` is already an unambiguous "blob-backed" marker, because the SDK returns early on an empty buffer and never uploads a chunk with no events, and the object key derives from (OrgId, SessionId, ChunkSeq). Adding a StorageKey column would have cost a ClickHouse migration, a SCHEMA_VERSION bump (which would route every not-yet-migrated BYO-ClickHouse org's logs/traces/metrics to Tinybird), and the first-ever versioned local-schema migration edge — for a table local mode never reads. Ordering is the invariant: the PUT happens before the row is enqueued, so a row can never point at a missing object. A failed PUT returns non-2xx and the SDK drops the chunk, which it already does without retrying. The bucket lifecycle expires objects at 32 days against the table's 30-day TTL, so the row always disappears first — a session that lists as recorded but plays back empty is the one failure with no good client-side handling. Ships dark and degrades by construction: an unset INGEST_REPLAY_R2_ENDPOINT keeps today's inline path, which is also what self-hosted and BYO-ClickHouse deployments run, and a missing R2 binding makes API hydration a no-op. Rollback is unsetting one env var. SigV4 is hand-rolled (~100 lines) against the hmac/sha2/chrono/reqwest crates the binary already links, rather than pulling the smithy stack in for one verb; it is pinned to AWS's published PUT Object test vector. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019vF2QDhdj77aJrjm89fgiq --- apps/api/alchemy.run.ts | 25 + apps/api/src/platform/ReplayBlobStore.test.ts | 137 +++++ apps/api/src/platform/ReplayBlobStore.ts | 136 +++++ apps/api/src/routes/v1/session-replay.http.ts | 22 +- .../api/src/routes/v2/session-replays.http.ts | 28 +- apps/cli/src/server/schema/local-inserts.json | 2 +- apps/cli/src/server/schema/local-schema.sql | 2 +- apps/ingest/src/clickhouse_insert_mappings.rs | 2 +- apps/ingest/src/lib.rs | 1 + apps/ingest/src/main.rs | 486 +++++++++++++++++- apps/ingest/src/metrics.rs | 15 + apps/ingest/src/r2.rs | 444 ++++++++++++++++ .../services/atoms/warehouse-query-atoms.ts | 4 +- lib/effect-cloudflare/package.json | 3 +- lib/effect-cloudflare/src/r2-bucket.ts | 26 + .../domain/src/generated/clickhouse-schema.ts | 2 +- .../generated/tinybird-project-manifest.ts | 4 +- packages/domain/src/http/session-replay.ts | 7 +- packages/domain/src/tinybird/datasources.ts | 22 +- packages/query-engine/src/ch/tables.ts | 5 +- 20 files changed, 1328 insertions(+), 45 deletions(-) create mode 100644 apps/api/src/platform/ReplayBlobStore.test.ts create mode 100644 apps/api/src/platform/ReplayBlobStore.ts create mode 100644 apps/ingest/src/r2.rs diff --git a/apps/api/alchemy.run.ts b/apps/api/alchemy.run.ts index 8102cddca..ae1c2023b 100644 --- a/apps/api/alchemy.run.ts +++ b/apps/api/alchemy.run.ts @@ -136,6 +136,28 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) => name: planetScaleWebhookQueueName, }) + // Session-replay rrweb payloads. The ingest gateway (a Railway container, + // not a Worker) writes these over the S3 API with SigV4; this binding is + // the read side, hydrating `session_replay_events` rows whose `Events` is + // empty. Stage-isolated, so a pr/stg deploy can never serve or overwrite + // prd recordings. + // + // The 32-day expiry is deliberately LONGER than the table's 30-day TTL: + // the row must disappear before the object does. The other way round + // leaves a session that lists as recorded but plays back empty, which is + // the one failure mode with no good client-side handling. + const replayBlobs = yield* Cloudflare.R2.Bucket("replay-blobs", { + name: resolveWorkerName("replay-blobs", stage), + lifecycleRules: [ + { + id: "expire-replay-chunks", + enabled: true, + prefix: "v1/", + deleteObjectsTransition: { condition: { type: "Age", maxAge: 32 * 24 * 60 * 60 } }, + }, + ], + }) + const worker = (yield* Cloudflare.Worker("api", { name: resolveWorkerName("api", stage), main: path.join(import.meta.dirname, "src", "worker.ts"), @@ -167,6 +189,9 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) => AI: Cloudflare.AI.Gateway("maple-api-ai"), CHAT_SESSION: chatSession, MCP_SESSIONS: mcpSessions, + // Read side of the replay payload store; absent bindings degrade to + // inline-only hydration (see platform/ReplayBlobStore.ts). + REPLAY_BLOBS: replayBlobs, VCS_SYNC_QUEUE: vcsSyncQueue, VCS_SYNC_QUEUE_NAME: vcsSyncQueueName, PLANETSCALE_WEBHOOK_QUEUE: planetScaleWebhookQueue, diff --git a/apps/api/src/platform/ReplayBlobStore.test.ts b/apps/api/src/platform/ReplayBlobStore.test.ts new file mode 100644 index 000000000..c9ccaf7e3 --- /dev/null +++ b/apps/api/src/platform/ReplayBlobStore.test.ts @@ -0,0 +1,137 @@ +import { describe, expect, it } from "vitest" +import { Effect, Layer } from "effect" +import { layerFromEnvRecord } from "@maple/effect-cloudflare/worker-environment" +import { ReplayBlobStore, replayObjectKey, REPLAY_BLOBS_BINDING } from "./ReplayBlobStore" + +const gzip = async (text: string): Promise => { + const stream = new Blob([text]).stream().pipeThrough(new CompressionStream("gzip")) + return new Uint8Array(await new Response(stream).arrayBuffer()) +} + +/** Minimal stand-in for the R2 binding: only `get` is exercised. */ +const fakeBucket = (objects: Record, onGet?: (key: string) => void) => ({ + get: async (key: string) => { + onGet?.(key) + const bytes = objects[key] + if (!bytes) return null + return { bytes: async () => bytes } + }, +}) + +const runWithBucket = ( + bucket: unknown, + program: (store: typeof ReplayBlobStore.Service) => Effect.Effect, +) => + Effect.runPromise( + Effect.gen(function* () { + const store = yield* ReplayBlobStore + return yield* program(store) + }).pipe( + Effect.provide( + ReplayBlobStore.layer.pipe( + Layer.provide(layerFromEnvRecord({ [REPLAY_BLOBS_BINDING]: bucket })), + ), + ), + ), + ) + +describe("replayObjectKey", () => { + // These expectations are duplicated verbatim in `replay_object_key`'s tests + // in apps/ingest/src/r2.rs. Nothing at runtime reconciles the two — the + // writer and the reader agree only by construction — and a divergence reads + // to a user as "every recording is empty", not as an error. So both suites + // pin the same strings. + it("matches the ingest gateway's key scheme", () => { + expect(replayObjectKey("org_123", "sess_abc", 7)).toBe("v1/org_123/sess_abc/00000007.json.gz") + }) + + it("zero-pads so lexicographic order is playback order", () => { + const keys = [ + replayObjectKey("o", "s", 10), + replayObjectKey("o", "s", 2), + replayObjectKey("o", "s", 1), + ].sort() + expect(keys).toEqual([ + "v1/o/s/00000001.json.gz", + "v1/o/s/00000002.json.gz", + "v1/o/s/00000010.json.gz", + ]) + }) + + it("does not truncate a chunk seq wider than the pad", () => { + expect(replayObjectKey("o", "s", 123_456_789)).toBe("v1/o/s/123456789.json.gz") + }) +}) + +describe("ReplayBlobStore.hydrate", () => { + const events = '[{"type":2,"timestamp":1}]' + + it("fills in payloads for blob-backed chunks", async () => { + const bucket = fakeBucket({ + "v1/org_1/sess_1/00000000.json.gz": await gzip(events), + }) + const result = await runWithBucket(bucket, (store) => + store.hydrate("org_1", "sess_1", [{ chunkSeq: 0, events: "", byteSize: 26 }]), + ) + expect(result).toEqual([{ chunkSeq: 0, events, byteSize: 26 }]) + }) + + it("leaves pre-cutover chunks untouched and never fetches them", async () => { + // The dual-read. A row written before the R2 cutover carries its payload + // inline; going to the bucket for it would 404 and silently drop a chunk + // that was there all along. + const fetched: string[] = [] + const bucket = fakeBucket({}, (key) => fetched.push(key)) + const result = await runWithBucket(bucket, (store) => + store.hydrate("org_1", "sess_1", [{ chunkSeq: 0, events }]), + ) + expect(result).toEqual([{ chunkSeq: 0, events }]) + expect(fetched).toEqual([]) + }) + + it("drops a chunk whose object is missing rather than failing the request", async () => { + const bucket = fakeBucket({ + "v1/org_1/sess_1/00000001.json.gz": await gzip(events), + }) + const result = await runWithBucket(bucket, (store) => + store.hydrate("org_1", "sess_1", [ + { chunkSeq: 0, events: "" }, + { chunkSeq: 1, events: "" }, + ]), + ) + expect(result).toEqual([{ chunkSeq: 1, events }]) + }) + + it("preserves chunk order despite concurrent fetches", async () => { + const objects: Record = {} + for (let seq = 0; seq < 20; seq++) { + objects[replayObjectKey("org_1", "sess_1", seq)] = await gzip(`[${seq}]`) + } + const result = await runWithBucket(objects && fakeBucket(objects), (store) => + store.hydrate( + "org_1", + "sess_1", + Array.from({ length: 20 }, (_, seq) => ({ chunkSeq: seq, events: "" })), + ), + ) + expect(result.map((chunk) => chunk.events)).toEqual( + Array.from({ length: 20 }, (_, seq) => `[${seq}]`), + ) + }) + + it("is a no-op when the binding is absent", async () => { + // Self-hosted and the Docker image of this API have no R2 at all. Every + // row there carries its payload inline, so hydration must pass through + // rather than erroring on a missing binding. + const chunks = [{ chunkSeq: 0, events }] + const result = await Effect.runPromise( + Effect.gen(function* () { + const store = yield* ReplayBlobStore + return yield* store.hydrate("org_1", "sess_1", chunks) + }).pipe( + Effect.provide(ReplayBlobStore.layer.pipe(Layer.provide(layerFromEnvRecord({})))), + ), + ) + expect(result).toEqual(chunks) + }) +}) diff --git a/apps/api/src/platform/ReplayBlobStore.ts b/apps/api/src/platform/ReplayBlobStore.ts new file mode 100644 index 000000000..c79c9141c --- /dev/null +++ b/apps/api/src/platform/ReplayBlobStore.ts @@ -0,0 +1,136 @@ +import { Context, Effect, Layer, Option } from "effect" +import { R2Bucket, type R2BucketClient } from "@maple/effect-cloudflare/r2-bucket" +import { WorkerEnvironment } from "@/platform/WorkerEnvironment" + +// --------------------------------------------------------------------------- +// ReplayBlobStore — reads rrweb chunk payloads out of R2. +// +// The ingest gateway stores each replay chunk's gzipped rrweb array as an +// object and writes a `session_replay_events` row that carries only the chunk's +// metadata, with an empty `Events`. This service turns those rows back into the +// inline payloads the player already expects, so the HTTP response shape is +// unchanged. +// +// Two states are normal, not errors: +// +// - **No binding.** Self-hosted installs and the Docker image of this API have +// no R2. `WorkerEnvironment` resolves to `{}` outside a Worker isolate, so +// `bindOptional` yields `None` and hydration becomes a no-op. +// - **Non-empty `events` on a row.** That row predates the R2 cutover (or came +// from a BYO-ClickHouse org, which never stops writing inline). It passes +// through untouched. This is the dual-read, and it is the whole migration +// strategy: no backfill, the old rows age out on the table's 30-day TTL. +// --------------------------------------------------------------------------- + +/** Binding name; must match the key in `apps/api/alchemy.run.ts`. */ +export const REPLAY_BLOBS_BINDING = "REPLAY_BLOBS" + +/** + * Object key for one replay chunk. + * + * **Must stay byte-identical to `replay_object_key` in `apps/ingest/src/r2.rs`.** + * Nothing at runtime cross-checks the two — a divergence reads as "every + * recording is empty", so the shared shape is asserted in both test suites. + */ +export const replayObjectKey = (orgId: string, sessionId: string, chunkSeq: number): string => + `v1/${orgId}/${sessionId}/${String(chunkSeq).padStart(8, "0")}.json.gz` + +/** The subset of a `session_replay_events` row hydration needs. */ +export interface HydratableChunk { + readonly chunkSeq: number + readonly events: string +} + +export interface ReplayBlobStoreShape { + /** + * Fill in `events` for every blob-backed chunk, preserving order. + * + * Chunks whose object is missing are **dropped**, not failed: a single + * absent chunk should cost you that slice of the recording, not the whole + * session. This matches how the player already tolerates a chunk it can't + * parse. + */ + readonly hydrate: ( + orgId: string, + sessionId: string, + chunks: readonly T[], + ) => Effect.Effect +} + +// Chunk fetches are independent; the player needs all of them before it can +// build the Replayer, so latency is the max, not the sum — but only if they +// overlap. Sequential gets across a long session would be a visible regression +// against the single ClickHouse round trip this replaces. +const FETCH_CONCURRENCY = 8 + +const decodeGzip = (bytes: Uint8Array): Promise => + new Response( + new Blob([bytes as unknown as BlobPart]).stream().pipeThrough(new DecompressionStream("gzip")), + ).text() + +const makeHydrate = + (bucket: R2BucketClient) => + (orgId: string, sessionId: string, chunks: readonly T[]) => + Effect.forEach( + chunks, + (chunk) => { + // Pre-cutover row: the payload is already in hand. + if (chunk.events !== "") return Effect.succeed(Option.some(chunk)) + const key = replayObjectKey(orgId, sessionId, chunk.chunkSeq) + return bucket.get(key).pipe( + Effect.flatMap((object) => + object === null + ? Effect.succeed(Option.none()) + : object + .bytes() + .pipe( + Effect.flatMap((bytes) => + Effect.promise(() => decodeGzip(bytes)), + ), + Effect.map((events) => Option.some({ ...chunk, events })), + ), + ), + // A failed fetch degrades the recording rather than the request. + // Logged at warning because a nonzero rate here means either the + // bucket lifecycle is outrunning the table TTL or the ingest-side + // key scheme has drifted from this one. + Effect.catch((cause) => + Effect.logWarning("replay chunk payload unavailable").pipe( + Effect.annotateLogs({ orgId, sessionId, key, cause: String(cause) }), + Effect.as(Option.none()), + ), + ), + ) + }, + { concurrency: FETCH_CONCURRENCY }, + ).pipe(Effect.map((results) => results.filter(Option.isSome).map((result) => result.value))) + +export class ReplayBlobStore extends Context.Service()( + "@maple/api/platform/ReplayBlobStore", + { + make: Effect.gen(function* () { + // The R2 client's methods each demand `WorkerEnvironment` again at call + // time, so capture it here and close over it. Otherwise the requirement + // leaks into every caller's type, and the routes would have to know + // they're on Cloudflare. + const env = yield* WorkerEnvironment + const bucket = yield* R2Bucket.bindOptional(R2Bucket(REPLAY_BLOBS_BINDING)) + if (Option.isNone(bucket)) { + // Expected on self-hosted / Docker / tests. Every row will carry its + // payload inline there, so hydration has nothing to do. + return { hydrate: (_orgId, _sessionId, chunks) => Effect.succeed([...chunks]) } + } + const hydrate = makeHydrate(bucket.value) + return { + hydrate: (orgId, sessionId, chunks) => + hydrate(orgId, sessionId, chunks).pipe( + Effect.provideService(WorkerEnvironment, env), + ), + } + }), + }, +) { + static readonly layer = Layer.effect(this, this.make) +} + +export const ReplayBlobStoreLive = ReplayBlobStore.layer.pipe(Layer.provide(WorkerEnvironment.layer)) diff --git a/apps/api/src/routes/v1/session-replay.http.ts b/apps/api/src/routes/v1/session-replay.http.ts index feb58f2e7..179f4d48b 100644 --- a/apps/api/src/routes/v1/session-replay.http.ts +++ b/apps/api/src/routes/v1/session-replay.http.ts @@ -13,17 +13,19 @@ import { TraceId, UserId, } from "@maple/domain/http" -import { Effect, Option, Schema } from "effect" +import { Effect, Layer, 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) -export const HttpSessionReplaysLive = HttpApiBuilder.group(MapleApi, "sessionReplays", (handlers) => +const HttpSessionReplaysGroup = HttpApiBuilder.group(MapleApi, "sessionReplays", (handlers) => Effect.gen(function* () { const warehouse = yield* WarehouseQueryService + const blobs = yield* ReplayBlobStore return handlers .handle("listReplays", ({ payload }) => @@ -195,12 +197,15 @@ export const HttpSessionReplaysLive = HttpApiBuilder.group(MapleApi, "sessionRep sessionId: payload.sessionId, }, ) - const chunks = yield* warehouse.compiledQuery(tenant, compiled, { + const rows = yield* warehouse.compiledQuery(tenant, compiled, { profile: "list", context: "getReplayEvents", }) - // rrweb payloads come straight from ClickHouse (no R2 / presigning); - // each chunk's `events` is the rrweb array JSON the player parses. + // 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 }) }), ) @@ -294,3 +299,10 @@ export const HttpSessionReplaysLive = HttpApiBuilder.group(MapleApi, "sessionRep ) }), ) + +// 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/v2/session-replays.http.ts b/apps/api/src/routes/v2/session-replays.http.ts index 609a1a0ae..e1baba5c0 100644 --- a/apps/api/src/routes/v2/session-replays.http.ts +++ b/apps/api/src/routes/v2/session-replays.http.ts @@ -18,8 +18,9 @@ import type { V2SessionTranscriptEvent, } from "@maple/domain/http/v2" import { CH, formatWarehouseDateTime } from "@maple/query-engine" -import { Effect, Option, Schema } from "effect" +import { Effect, Layer, Option, Schema } from "effect" import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService" +import { ReplayBlobStore, ReplayBlobStoreLive } from "@/platform/ReplayBlobStore" import { warehouseToV2 } from "./warehouse-error-map" const decodeSessionId = Schema.decodeSync(SessionId) @@ -51,9 +52,10 @@ const chToIsoOrNull = (value: string | null): Timestamp | null => (value === nul const nullableUserId = (value: string | null): string | null => (value ? value : null) -export const HttpV2SessionReplaysLive = HttpApiBuilder.group(MapleApiV2, "sessionReplays", (handlers) => +const HttpV2SessionReplaysGroup = HttpApiBuilder.group(MapleApiV2, "sessionReplays", (handlers) => Effect.gen(function* () { const warehouse = yield* WarehouseQueryService + const blobs = yield* ReplayBlobStore const requireSession = Effect.fn("HttpV2SessionReplays.requireSession")(function* ( tenant: CurrentTenant.TenantSchema, @@ -218,12 +220,20 @@ export const HttpV2SessionReplaysLive = HttpApiBuilder.group(MapleApiV2, "sessio CH.sessionReplayEventsQuery({ startTime: windowStart, endTime: windowEnd }), { orgId: tenant.orgId, sessionId: params.id }, ) - const rows = yield* warehouse + const stored = yield* warehouse .compiledQuery(tenant, compiled, { profile: "list", context: "v2GetReplayEvents" }) .pipe(Effect.mapError(mapWarehouseError)) - if (rows.length === 0) { + if (stored.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( (row) => ({ @@ -237,8 +247,7 @@ export const HttpV2SessionReplaysLive = HttpApiBuilder.group(MapleApiV2, "sessio events: row.events, }) satisfies V2SessionReplayChunk, ) - const page = yield* paginateArray(chunks, query) - return { object: "list" as const, ...page } + return { object: "list" as const, ...page, data: chunks } }), ) .handle("transcript", ({ params, query }) => @@ -340,3 +349,10 @@ export const HttpV2SessionReplaysLive = HttpApiBuilder.group(MapleApiV2, "sessio ) }), ) + +// 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 HttpV2SessionReplaysLive = HttpV2SessionReplaysGroup.pipe(Layer.provide(ReplayBlobStoreLive)) diff --git a/apps/cli/src/server/schema/local-inserts.json b/apps/cli/src/server/schema/local-inserts.json index ee160d0e4..54ae65f31 100644 --- a/apps/cli/src/server/schema/local-inserts.json +++ b/apps/cli/src/server/schema/local-inserts.json @@ -1,5 +1,5 @@ { - "projectRevision": "53294ef75d1afb2e4c9ce6ba2e80e9900e4996e6731838a1ce1902803588c58d", + "projectRevision": "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06", "orgPlaceholder": "__ORG__", "datasources": { "traces": { diff --git a/apps/cli/src/server/schema/local-schema.sql b/apps/cli/src/server/schema/local-schema.sql index aeb41736f..e788933c7 100644 --- a/apps/cli/src/server/schema/local-schema.sql +++ b/apps/cli/src/server/schema/local-schema.sql @@ -1,6 +1,6 @@ -- This file is generated by scripts/generate-clickhouse-schema-sql.ts -- Do not edit manually. Run `bun run clickhouse:schema` to regenerate. --- projectRevision: 53294ef75d1afb2e4c9ce6ba2e80e9900e4996e6731838a1ce1902803588c58d +-- projectRevision: 6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06 -- localSchemaVersion: 1 CREATE TABLE IF NOT EXISTS alert_checks ( diff --git a/apps/ingest/src/clickhouse_insert_mappings.rs b/apps/ingest/src/clickhouse_insert_mappings.rs index c4cefc85d..00e4963b1 100644 --- a/apps/ingest/src/clickhouse_insert_mappings.rs +++ b/apps/ingest/src/clickhouse_insert_mappings.rs @@ -1,7 +1,7 @@ // This file is generated by scripts/generate-clickhouse-insert-mappings.ts // Do not edit manually. -pub const PROJECT_REVISION: &str = "53294ef75d1afb2e4c9ce6ba2e80e9900e4996e6731838a1ce1902803588c58d"; +pub const PROJECT_REVISION: &str = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06"; // Gate for BYO-ClickHouse ingest readiness — the migration version, NOT the // Tinybird-coupled PROJECT_REVISION. Compared against // org_clickhouse_settings.schema_version. See @maple/domain/clickhouse diff --git a/apps/ingest/src/lib.rs b/apps/ingest/src/lib.rs index 53cd2ff9e..885b407f1 100644 --- a/apps/ingest/src/lib.rs +++ b/apps/ingest/src/lib.rs @@ -2,6 +2,7 @@ pub mod clickhouse_insert_mappings; pub mod metrics; pub mod otel; pub mod otlp_json; +pub mod r2; pub mod session_analytics; pub mod telemetry; pub mod usage_metrics; diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index d654b28ba..ae38a4718 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -38,6 +38,7 @@ use maple_ingest::otel::{ record_stage_error, rejection_loses_data, resolve_config_internal_span, ResourceConfig, }; use maple_ingest::otlp_json; +use maple_ingest::r2::{replay_object_key, ReplayBlobStore}; use maple_ingest::session_analytics::{ derive_referrer_host, sanitize_session_event, sanitize_session_meta, }; @@ -99,6 +100,17 @@ fn is_sentinel_token(token: &str) -> bool { type HmacSha256 = Hmac; +/// Credentials for the S3-compatible endpoint that holds replay chunk payloads. +#[derive(Clone)] +struct ReplayBlobStoreConfig { + endpoint: String, + bucket: String, + access_key_id: String, + secret_access_key: String, + region: String, + timeout: Duration, +} + #[derive(Clone)] struct AppConfig { port: u16, @@ -123,6 +135,11 @@ struct AppConfig { /// Ceiling on the total decompressed rrweb payload a single replay session /// may accumulate. 0 disables the cap. See `ReplaySessionBudget`. replay_max_session_bytes: u64, + /// Where replay chunk payloads are stored. `None` — the default, and the + /// only option for self-hosted and BYO-ClickHouse deployments — keeps the + /// rrweb JSON inline in the `session_replay_events` row. `Some` diverts the + /// payload to R2 and writes a thin index row with an empty `events`. + replay_blob_store: Option, /// Whether `Cf-IPCountry` on an inbound request can be believed. /// /// Off by default, and that default is the safe one: this gateway is a @@ -417,6 +434,51 @@ impl AppConfig { 1024 * 1024 * 1024, )?; + // Replay payload storage. An unset endpoint is the signal for "keep the + // rrweb JSON inline in ClickHouse" — that is what self-hosted and + // BYO-ClickHouse deployments run, and it is also how this ships dark on + // the managed path until the credentials are set. Anything half-set is a + // misconfiguration we refuse to boot on rather than silently falling + // back to inline, which would look identical in metrics until someone + // noticed the warehouse bill hadn't moved. + let replay_blob_store = { + let endpoint = std::env::var("INGEST_REPLAY_R2_ENDPOINT") + .ok() + .map(|v| v.trim().to_string()) + .filter(|v| !v.is_empty()); + match endpoint { + None => None, + Some(endpoint) => { + let required = |name: &str| -> Result { + std::env::var(name) + .ok() + .map(|v| v.trim().to_string()) + .filter(|v| !v.is_empty()) + .ok_or_else(|| { + format!("{name} is required when INGEST_REPLAY_R2_ENDPOINT is set") + }) + }; + let timeout_ms = parse_u64( + "INGEST_REPLAY_R2_TIMEOUT_MS", + std::env::var("INGEST_REPLAY_R2_TIMEOUT_MS").ok(), + 5_000, + )?; + Some(ReplayBlobStoreConfig { + endpoint, + bucket: required("INGEST_REPLAY_R2_BUCKET")?, + access_key_id: required("INGEST_REPLAY_R2_ACCESS_KEY_ID")?, + secret_access_key: required("INGEST_REPLAY_R2_SECRET_ACCESS_KEY")?, + region: std::env::var("INGEST_REPLAY_R2_REGION") + .ok() + .map(|v| v.trim().to_string()) + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "auto".to_string()), + timeout: Duration::from_millis(timeout_ms), + }) + } + } + }; + // Default off — see the field doc. Set it on services that are only // reachable through Cloudflare. let trust_proxy_geo = parse_bool( @@ -446,6 +508,7 @@ impl AppConfig { ingest_key_cache_ttl_secs, org_routing_cache_ttl_secs, replay_max_session_bytes, + replay_blob_store, trust_proxy_geo, }) } @@ -831,6 +894,9 @@ struct AppState { /// (local dev, or a loopback endpoint). usage_metrics: Option>, replay_session_budget: ReplaySessionBudget, + /// `Some` when replay payloads go to R2; `None` keeps them inline in the + /// `session_replay_events` row. See `AppConfig::replay_blob_store`. + replay_blob_store: Option, } #[derive(Clone)] @@ -1600,6 +1666,10 @@ async fn main() { last_known: DashMap::new(), }); + // Same pooled client as every other outbound call; `http_client` itself is + // moved into the state below. + let http_client_for_blobs = http_client.clone(); + let state = Arc::new(AppState { resolver: IngestKeyResolver { store: Arc::clone(&store), @@ -1629,6 +1699,17 @@ async fn main() { autumn_entitlements, usage_metrics: usage_metrics.clone(), replay_session_budget: ReplaySessionBudget::new(config.replay_max_session_bytes), + replay_blob_store: config.replay_blob_store.as_ref().map(|blob| { + ReplayBlobStore::new( + http_client_for_blobs, + blob.endpoint.clone(), + blob.bucket.clone(), + blob.access_key_id.clone(), + blob.secret_access_key.clone(), + blob.region.clone(), + blob.timeout, + ) + }), }); let cors = CorsLayer::new() @@ -2538,6 +2619,30 @@ async fn handle_session_events_inner( Ok(count) } +/// Decompressed length of a gzip payload, without materializing it. +/// +/// Same number `read_to_string(...).len()` would produce, and the same +/// rejection of malformed gzip — it just doesn't keep the bytes. Used on the +/// blob-store path, where the decompressed text is never needed but `ByteSize` +/// and the per-session budget are still denominated in decompressed bytes. +fn decompressed_len(body: &[u8]) -> Result { + use std::io::Read as _; + let mut decoder = flate2::read::GzDecoder::new(body); + let mut buffer = [0u8; 64 * 1024]; + let mut total: u64 = 0; + loop { + match decoder.read(&mut buffer) { + Ok(0) => return Ok(total), + Ok(n) => total += n as u64, + Err(e) => { + return Err(ApiError::bad_request(format!( + "failed to gunzip replay chunk: {e}" + ))) + } + } + } +} + async fn handle_replay_blob( State(state): State>, headers: HeaderMap, @@ -2562,6 +2667,9 @@ async fn handle_replay_blob( "maple.ingest.clickhouse_ready" = tracing::field::Empty, "maple.ingest.destination" = tracing::field::Empty, "maple.replay.truncated" = tracing::field::Empty, + "maple.replay.storage" = tracing::field::Empty, + "maple.replay.object_key" = tracing::field::Empty, + "maple.replay.blob_put_ms" = tracing::field::Empty, ); let span_handle = span.clone(); match handle_replay_blob_inner(&state, &headers, body) @@ -2631,6 +2739,15 @@ async fn handle_replay_blob_inner( if !is_safe_replay_id(&session_id) { return Err(ApiError::bad_request("invalid x-maple-session-id")); } + // The org id comes from the resolved key rather than the request, so this is + // a guard against a malformed key row, not against the caller. It matters + // because the id is now a path segment in a signed URL, not just a quoted + // SQL param. + if state.replay_blob_store.is_some() && !is_safe_replay_id(&org_id) { + return Err(ApiError::service_unavailable( + "organization id is not storage-key safe", + )); + } let chunk_seq: u32 = replay_header(headers, "x-maple-chunk-seq") .and_then(|v| v.parse().ok()) .ok_or_else(|| ApiError::bad_request("missing or invalid x-maple-chunk-seq header"))?; @@ -2658,16 +2775,27 @@ async fn handle_replay_blob_inner( )); } - // The SDK gzips the rrweb event array (native CompressionStream). Decompress - // here so the events land in ClickHouse as queryable JSON text (the column is - // ZSTD-compressed by the warehouse) — no R2 blob store on the replay path. - use std::io::Read as _; - let mut decoder = flate2::read::GzDecoder::new(&body[..]); - let mut events_json = String::new(); - decoder - .read_to_string(&mut events_json) - .map_err(|e| ApiError::bad_request(format!("failed to gunzip replay chunk: {e}")))?; - let byte_size = events_json.len() as u64; + // The SDK gzips the rrweb event array (native CompressionStream). + // + // With a blob store configured we never need the decompressed text — the + // gzip is stored verbatim — but we still decode it, for two reasons: this is + // what rejects malformed gzip from a hostile client, and `byte_size` is a + // published API field and the input to `ReplaySessionBudget`, whose ceiling + // is denominated in *decompressed* bytes. So decode and discard, counting. + // What that avoids is the part that actually cost: materializing a + // multi-megabyte String, JSON-escaping it, and pushing it through the WAL. + let (events_json, byte_size) = if state.replay_blob_store.is_some() { + (None, decompressed_len(&body)?) + } else { + use std::io::Read as _; + let mut decoder = flate2::read::GzDecoder::new(&body[..]); + let mut events_json = String::new(); + decoder + .read_to_string(&mut events_json) + .map_err(|e| ApiError::bad_request(format!("failed to gunzip replay chunk: {e}")))?; + let byte_size = events_json.len() as u64; + (Some(events_json), byte_size) + }; // Accept the chunk that crosses the ceiling so the recording truncates on a // chunk boundary; `is_exhausted` rejects everything after it. @@ -2688,6 +2816,57 @@ async fn handle_replay_blob_inner( metrics::replay_session_truncated(&org_id); } + // Store the payload before the row that indexes it. The ordering is the + // invariant: a row must never point at an object that isn't there, and a + // failed PUT returns non-2xx so the SDK drops the chunk (it already does not + // retry) rather than leaving an unplayable gap in a listed session. The + // reverse — an object with no row — is harmless and gets swept by the + // bucket's lifecycle rule. + let events_json = match (&state.replay_blob_store, events_json) { + (Some(store), _) => { + let key = replay_object_key(&org_id, &session_id, chunk_seq); + Span::current().record("maple.replay.storage", "r2"); + Span::current().record("maple.replay.object_key", key.as_str()); + let started = Instant::now(); + // Verbatim gzip: ~10x smaller at rest than the JSON text, no + // recompression, and the Content-Encoding lets a reader hand the + // bytes to a browser to inflate. + store + .put_object(&key, body.to_vec(), "application/json", Some("gzip")) + .await + .map_err(|e| { + warn!( + org_id = %org_id, + session_id = %session_id, + chunk_seq, + error = %e, + "replay chunk blob upload failed" + ); + metrics::replay_blob_put_failed(&org_id); + ApiError::service_unavailable("failed to store replay chunk") + })?; + Span::current().record( + "maple.replay.blob_put_ms", + started.elapsed().as_millis() as i64, + ); + // The index row carries the chunk's metadata; the payload lives in + // R2 under a key derived from (OrgId, SessionId, ChunkSeq). An empty + // `events` is what marks the row as blob-backed on read — the SDK + // never posts an empty chunk, so it cannot occur otherwise. + String::new() + } + (None, Some(events_json)) => { + Span::current().record("maple.replay.storage", "inline"); + events_json + } + // Unreachable: `events_json` is only `None` when a store is configured. + (None, None) => { + return Err(ApiError::service_unavailable( + "replay chunk was neither stored nor decoded", + )) + } + }; + // Row → session_replay_events. Tinybird parses the space-separated datetime // into DateTime64(9); `events` is stored verbatim as a String column. let timestamp = chrono::Utc::now() @@ -6099,6 +6278,7 @@ mod tests { ingest_key_cache_ttl_secs: 60, org_routing_cache_ttl_secs: 5, replay_max_session_bytes: 1024 * 1024 * 1024, + replay_blob_store: None, trust_proxy_geo: false, }, http_client, @@ -6140,9 +6320,34 @@ mod tests { autumn_entitlements: None, usage_metrics: None, replay_session_budget: ReplaySessionBudget::new(1024 * 1024 * 1024), + replay_blob_store: None, } } + /// Point a state's replay payloads at a fake S3 endpoint. Mirrors what + /// `INGEST_REPLAY_R2_*` does in `Config::from_env`. + fn with_replay_blob_store(mut state: AppState, endpoint: String) -> AppState { + let config = ReplayBlobStoreConfig { + endpoint, + bucket: "replays".to_string(), + access_key_id: "test-access-key".to_string(), + secret_access_key: "test-secret-key".to_string(), + region: "auto".to_string(), + timeout: Duration::from_secs(5), + }; + state.replay_blob_store = Some(ReplayBlobStore::new( + state.http_client.clone(), + config.endpoint.clone(), + config.bucket.clone(), + config.access_key_id.clone(), + config.secret_access_key.clone(), + config.region.clone(), + config.timeout, + )); + state.config.replay_blob_store = Some(config); + state + } + #[tokio::test] async fn replay_budget_truncates_session_at_its_ceiling() { let budget = ReplaySessionBudget::new(1_000); @@ -6293,6 +6498,267 @@ mod tests { let _ = std::fs::remove_dir_all(&queue_dir); } + /// What a fake R2 recorded for one PUT. + #[derive(Debug)] + struct CapturedPut { + path: String, + authorization: String, + content_type: String, + content_encoding: String, + body: Vec, + } + + async fn fake_r2_put( + axum::extract::State(tx): axum::extract::State< + tokio::sync::mpsc::UnboundedSender, + >, + Path(path): Path, + headers: HeaderMap, + body: Bytes, + ) -> StatusCode { + let header = |name: &str| { + headers + .get(name) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .to_string() + }; + let _ = tx.send(CapturedPut { + path, + authorization: header("authorization"), + content_type: header("content-type"), + content_encoding: header("content-encoding"), + body: body.to_vec(), + }); + StatusCode::OK + } + + async fn always_500() -> StatusCode { + StatusCode::INTERNAL_SERVER_ERROR + } + + fn gzip_bytes(plain: &str) -> Bytes { + let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); + encoder.write_all(plain.as_bytes()).unwrap(); + Bytes::from(encoder.finish().unwrap()) + } + + fn replay_blob_headers(raw_key: &str, session_id: &str, chunk_seq: u32) -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert( + "authorization", + format!("Bearer {raw_key}").parse().unwrap(), + ); + headers.insert("x-maple-session-id", session_id.parse().unwrap()); + headers.insert("x-maple-chunk-seq", chunk_seq.to_string().parse().unwrap()); + headers.insert("x-maple-event-count", "3".parse().unwrap()); + headers.insert("x-maple-duration-ms", "1200".parse().unwrap()); + headers + } + + /// Total bytes the pipeline has committed to disk. The WAL is appended + /// before a frame reaches the export channel, so this growing is the + /// observable "a row was enqueued". + fn queue_dir_bytes(dir: &PathBuf) -> u64 { + fn walk(dir: &std::path::Path) -> u64 { + let Ok(entries) = std::fs::read_dir(dir) else { + return 0; + }; + entries + .flatten() + .map(|entry| match entry.metadata() { + Ok(meta) if meta.is_dir() => walk(&entry.path()), + Ok(meta) => meta.len(), + Err(_) => 0, + }) + .sum() + } + walk(dir) + } + + async fn replay_blob_test_state( + raw_key: &str, + org_id: &str, + queue_dir: PathBuf, + ) -> AppState { + let store = Arc::new(FakeKeyStore::default()); + store.insert_private( + raw_key, + KeyRow { + org_id: org_id.to_string(), + // Routes to ClickHouse. The fixture's `WriteMode::Forward` has no + // Tinybird pipeline, so a Tinybird-destined chunk would 503 in + // `native_rows_pipeline_for` before reaching the blob path. + self_managed: true, + clickhouse_ready: true, + spend_paused: false, + paused_features: Vec::new(), + }, + ); + test_app_state( + store, + queue_dir, + "http://127.0.0.1:1".to_string(), + Duration::from_secs(30), + ) + .await + } + + #[tokio::test] + async fn replay_chunk_payload_goes_to_the_blob_store_verbatim() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new() + .route("/{*path}", axum::routing::put(fake_r2_put)) + .with_state(tx); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let queue_dir = unique_main_test_dir("replay-blob-put"); + let state = replay_blob_test_state( + "maple_sk_test_replay_blob", + "org_replay_blob", + queue_dir.clone(), + ) + .await; + let state = with_replay_blob_store(state, format!("http://{addr}")); + + let events = r#"[{"type":2,"timestamp":1}]"#; + let gzipped = gzip_bytes(events); + handle_replay_blob_inner( + &state, + &replay_blob_headers("maple_sk_test_replay_blob", "sess_42", 7), + gzipped.clone(), + ) + .await + .expect("blob upload should succeed"); + + let captured = rx.recv().await.expect("the blob store should see a PUT"); + + // Key scheme: bucket first, then the v1/{org}/{session}/{seq}.json.gz + // that the API side reconstructs from the ClickHouse row. + assert_eq!( + captured.path, + "replays/v1/org_replay_blob/sess_42/00000007.json.gz" + ); + // Stored verbatim — not re-gzipped, not the decompressed text. A + // recompression here would silently double ingest CPU and break the + // Content-Encoding contract the reader depends on. + assert_eq!(captured.body, gzipped.to_vec()); + assert_eq!(captured.content_type, "application/json"); + assert_eq!(captured.content_encoding, "gzip"); + assert!( + captured.authorization.starts_with( + "AWS4-HMAC-SHA256 Credential=test-access-key/" + ), + "expected a SigV4 authorization header, got {:?}", + captured.authorization + ); + assert!(captured.authorization.contains("/auto/s3/aws4_request")); + + let _ = std::fs::remove_dir_all(&queue_dir); + } + + #[tokio::test] + async fn a_failed_blob_upload_rejects_the_chunk_and_enqueues_no_row() { + // The orphan-prevention invariant. A row whose payload never landed is + // an unplayable gap in a session that still lists as recorded; the SDK + // does not retry, so the only safe failure is to drop the chunk whole. + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new().route("/{*path}", axum::routing::put(always_500)); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let queue_dir = unique_main_test_dir("replay-blob-fail"); + let state = replay_blob_test_state( + "maple_sk_test_replay_fail", + "org_replay_fail", + queue_dir.clone(), + ) + .await; + let state = with_replay_blob_store(state, format!("http://{addr}")); + + let before = queue_dir_bytes(&queue_dir); + let error = handle_replay_blob_inner( + &state, + &replay_blob_headers("maple_sk_test_replay_fail", "sess_fail", 1), + gzip_bytes(r#"[{"type":2,"timestamp":1}]"#), + ) + .await + .expect_err("a blob store 500 must reject the chunk"); + + assert_eq!(error.status, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + queue_dir_bytes(&queue_dir), + before, + "no index row may be committed when the payload was not stored" + ); + + let _ = std::fs::remove_dir_all(&queue_dir); + } + + #[tokio::test] + async fn replay_chunks_stay_inline_when_no_blob_store_is_configured() { + // The self-hosted / BYO-ClickHouse path, and the pre-cutover managed + // path: unset credentials must behave exactly as before. + let queue_dir = unique_main_test_dir("replay-blob-inline"); + let state = replay_blob_test_state( + "maple_sk_test_replay_inline", + "org_replay_inline", + queue_dir.clone(), + ) + .await; + assert!(state.replay_blob_store.is_none()); + + let before = queue_dir_bytes(&queue_dir); + handle_replay_blob_inner( + &state, + &replay_blob_headers("maple_sk_test_replay_inline", "sess_inline", 0), + gzip_bytes(r#"[{"type":2,"timestamp":1}]"#), + ) + .await + .expect("the inline path should accept the chunk"); + + assert!( + queue_dir_bytes(&queue_dir) > before, + "the inline path must still enqueue a row carrying the payload" + ); + + let _ = std::fs::remove_dir_all(&queue_dir); + } + + #[test] + fn decompressed_len_matches_read_to_string_and_rejects_garbage() { + // `byte_size` is a published API field and the input to the per-session + // budget, both denominated in decompressed bytes — the streaming counter + // must not quietly redefine it as compressed bytes. + for payload in [ + "[]", + r#"[{"type":2,"timestamp":1}]"#, + &"x".repeat(256 * 1024), + ] { + let gzipped = gzip_bytes(payload); + assert_eq!( + decompressed_len(&gzipped).expect("valid gzip should decode"), + payload.len() as u64, + "byte count drifted for a {}-byte payload", + payload.len() + ); + } + + let error = decompressed_len(b"not gzip at all") + .expect_err("malformed gzip must still be rejected"); + assert_eq!(error.status, StatusCode::BAD_REQUEST); + } + #[tokio::test] async fn resolve_ingest_key_returns_self_managed_false_when_no_settings_row() { let store = Arc::new(FakeKeyStore::default()); diff --git a/apps/ingest/src/metrics.rs b/apps/ingest/src/metrics.rs index 5cbcc5d8a..602c6a634 100644 --- a/apps/ingest/src/metrics.rs +++ b/apps/ingest/src/metrics.rs @@ -71,6 +71,13 @@ static REPLAY_SESSION_CHUNK_DROPPED_TOTAL: LazyLock> = LazyLock::ne .build() }); +static REPLAY_BLOB_PUT_FAILED_TOTAL: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("ingest_replay_blob_put_failed_total") + .with_description("Replay chunks rejected because their payload could not be stored") + .build() +}); + static CLOUDFLARE_BATCHES_TOTAL: LazyLock> = LazyLock::new(|| { METER .u64_counter("ingest_cloudflare_batches_total") @@ -431,6 +438,14 @@ pub fn replay_session_chunk_dropped(org_id: &str) { REPLAY_SESSION_CHUNK_DROPPED_TOTAL.add(1, &[KeyValue::new("org_id", org_id.to_string())]); } +/// A replay chunk's payload could not be written to the blob store, so the +/// chunk was rejected and no index row was enqueued. The SDK does not retry, so +/// every increment here is a permanent gap in a recording — this should sit at +/// zero, and it is the signal to watch during the R2 cutover. +pub fn replay_blob_put_failed(org_id: &str) { + REPLAY_BLOB_PUT_FAILED_TOTAL.add(1, &[KeyValue::new("org_id", org_id.to_string())]); +} + /// Current in-flight request count for an org. pub fn org_requests_in_flight(org_id: &str, value: u64) { ORG_REQUESTS_IN_FLIGHT.record(value, &[KeyValue::new("org_id", org_id.to_string())]); diff --git a/apps/ingest/src/r2.rs b/apps/ingest/src/r2.rs new file mode 100644 index 000000000..137a7b1ed --- /dev/null +++ b/apps/ingest/src/r2.rs @@ -0,0 +1,444 @@ +//! Minimal S3-compatible object writer for Cloudflare R2. +//! +//! The gateway runs as a container (Railway), not on Workers, so it can't reach +//! R2 through a native binding — it signs S3 requests with SigV4 like any other +//! client. That is the whole reason this module exists. +//! +//! Deliberately not `aws-sdk-s3`: we issue exactly one verb (`PUT`) against one +//! bucket with a known-length in-memory payload. Signing that is ~100 lines on +//! top of the `hmac`/`sha2`/`chrono`/`reqwest` crates the binary already links +//! for ingest-key hashing, versus pulling the whole smithy stack in for it. +//! +//! Scope: replay chunk payloads only. If this ever grows a second caller or a +//! second verb, that is the moment to reconsider the dependency. + +use std::time::Duration; + +use chrono::{DateTime, Utc}; +use hmac::{Hmac, Mac}; +use sha2::{Digest, Sha256}; + +type HmacSha256 = Hmac; + +const ALGORITHM: &str = "AWS4-HMAC-SHA256"; +const SERVICE: &str = "s3"; + +/// Object-key scheme version. Bumping this is how a future key layout change +/// stays decodable: rows written under `v1/` keep resolving while new rows land +/// under `v2/`, for the 30 days it takes the old ones to age out. +const KEY_SCHEME: &str = "v1"; + +#[derive(Debug)] +pub enum R2Error { + /// The request never got a response (DNS, TLS, connect, timeout). + Transport(String), + /// R2 answered with a non-2xx status. + Status { status: u16, body: String }, +} + +impl std::fmt::Display for R2Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + R2Error::Transport(message) => write!(f, "r2 transport error: {message}"), + R2Error::Status { status, body } => { + write!(f, "r2 responded {status}: {body}") + } + } + } +} + +impl std::error::Error for R2Error {} + +/// Storage key for one replay chunk. +/// +/// `orgId` leads so a per-tenant lifecycle rule or a "delete this org's data" +/// sweep is a prefix operation. `chunkSeq` is zero-padded so lexicographic list +/// order equals playback order. +/// +/// Deliberately carries no date component: the read side reconstructs this key +/// from the ClickHouse row's `(OrgId, SessionId, ChunkSeq)` alone, and a date +/// would couple it to the gateway's `Utc::now()` in a way that disagrees with +/// the row's own `Timestamp` across a UTC midnight boundary. +pub fn replay_object_key(org_id: &str, session_id: &str, chunk_seq: u32) -> String { + format!("{KEY_SCHEME}/{org_id}/{session_id}/{chunk_seq:08}.json.gz") +} + +/// Writes replay chunk payloads to an S3-compatible bucket. +#[derive(Clone)] +pub struct ReplayBlobStore { + client: reqwest::Client, + /// Origin only, no trailing slash: `https://.r2.cloudflarestorage.com`. + endpoint: String, + host: String, + bucket: String, + access_key_id: String, + secret_access_key: String, + region: String, + timeout: Duration, +} + +impl ReplayBlobStore { + pub fn new( + client: reqwest::Client, + endpoint: String, + bucket: String, + access_key_id: String, + secret_access_key: String, + region: String, + timeout: Duration, + ) -> Self { + let endpoint = endpoint.trim_end_matches('/').to_string(); + let host = host_from_endpoint(&endpoint); + Self { + client, + endpoint, + host, + bucket, + access_key_id, + secret_access_key, + region, + timeout, + } + } + + /// PUT one object. `body` is stored verbatim — the caller owns compression. + pub async fn put_object( + &self, + key: &str, + body: Vec, + content_type: &str, + content_encoding: Option<&str>, + ) -> Result<(), R2Error> { + self.put_object_at(key, body, content_type, content_encoding, Utc::now()) + .await + } + + async fn put_object_at( + &self, + key: &str, + body: Vec, + content_type: &str, + content_encoding: Option<&str>, + now: DateTime, + ) -> Result<(), R2Error> { + let canonical_uri = format!("/{}/{}", self.bucket, uri_encode_path(key)); + let payload_sha256 = hex(&Sha256::digest(&body)); + + let mut headers: Vec<(String, String)> = vec![ + ("content-type".to_string(), content_type.to_string()), + ("host".to_string(), self.host.clone()), + ("x-amz-content-sha256".to_string(), payload_sha256.clone()), + ("x-amz-date".to_string(), amz_date(now)), + ]; + if let Some(encoding) = content_encoding { + headers.push(("content-encoding".to_string(), encoding.to_string())); + } + + let authorization = authorization_header(&SigningParams { + method: "PUT", + canonical_uri: &canonical_uri, + canonical_query: "", + headers: &headers, + payload_sha256: &payload_sha256, + now, + access_key_id: &self.access_key_id, + secret_access_key: &self.secret_access_key, + region: &self.region, + service: SERVICE, + }); + + let mut request = self + .client + .put(format!("{}{}", self.endpoint, canonical_uri)) + .timeout(self.timeout) + .header("authorization", authorization); + for (name, value) in &headers { + // `host` is set by reqwest from the URL; re-setting it risks a + // mismatch with what we just signed. + if name != "host" { + request = request.header(name.as_str(), value.as_str()); + } + } + + let response = request + .body(body) + .send() + .await + .map_err(|e| R2Error::Transport(e.to_string()))?; + + let status = response.status(); + if status.is_success() { + return Ok(()); + } + // R2 returns an XML error document; keep a bounded prefix for the log. + let body = response.text().await.unwrap_or_default(); + Err(R2Error::Status { + status: status.as_u16(), + body: body.chars().take(512).collect(), + }) + } +} + +struct SigningParams<'a> { + method: &'a str, + canonical_uri: &'a str, + canonical_query: &'a str, + /// Lowercase header names; sorted internally, so call order is free. + headers: &'a [(String, String)], + payload_sha256: &'a str, + now: DateTime, + access_key_id: &'a str, + secret_access_key: &'a str, + region: &'a str, + service: &'a str, +} + +fn authorization_header(params: &SigningParams) -> String { + let mut headers: Vec<(String, String)> = params + .headers + .iter() + .map(|(name, value)| (name.to_ascii_lowercase(), value.trim().to_string())) + .collect(); + headers.sort_by(|a, b| a.0.cmp(&b.0)); + + let signed_headers = headers + .iter() + .map(|(name, _)| name.as_str()) + .collect::>() + .join(";"); + let canonical_headers = headers + .iter() + .map(|(name, value)| format!("{name}:{value}\n")) + .collect::(); + + let canonical_request = format!( + "{}\n{}\n{}\n{}\n{}\n{}", + params.method, + params.canonical_uri, + params.canonical_query, + canonical_headers, + signed_headers, + params.payload_sha256, + ); + + let datestamp = params.now.format("%Y%m%d").to_string(); + let scope = format!( + "{}/{}/{}/aws4_request", + datestamp, params.region, params.service + ); + let string_to_sign = format!( + "{}\n{}\n{}\n{}", + ALGORITHM, + amz_date(params.now), + scope, + hex(&Sha256::digest(canonical_request.as_bytes())), + ); + + let signing_key = signing_key( + params.secret_access_key, + &datestamp, + params.region, + params.service, + ); + let signature = hex(&hmac(&signing_key, string_to_sign.as_bytes())); + + format!( + "{ALGORITHM} Credential={}/{scope}, SignedHeaders={signed_headers}, Signature={signature}", + params.access_key_id, + ) +} + +fn signing_key(secret_access_key: &str, datestamp: &str, region: &str, service: &str) -> Vec { + let mut key = hmac( + format!("AWS4{secret_access_key}").as_bytes(), + datestamp.as_bytes(), + ); + key = hmac(&key, region.as_bytes()); + key = hmac(&key, service.as_bytes()); + hmac(&key, b"aws4_request") +} + +fn hmac(key: &[u8], data: &[u8]) -> Vec { + let mut mac = HmacSha256::new_from_slice(key).expect("hmac accepts any key length"); + mac.update(data); + mac.finalize().into_bytes().to_vec() +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +fn amz_date(now: DateTime) -> String { + now.format("%Y%m%dT%H%M%SZ").to_string() +} + +/// RFC 3986 encoding for a URI path: `/` stays a separator, everything outside +/// the unreserved set is percent-encoded with uppercase hex. Replay keys only +/// ever contain unreserved characters, but a signature that disagrees with the +/// request line by one byte is a 403 with no useful message, so encode properly +/// rather than trusting the caller. +fn uri_encode_path(path: &str) -> String { + let mut out = String::with_capacity(path.len()); + for byte in path.as_bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' | b'/' => { + out.push(*byte as char) + } + _ => out.push_str(&format!("%{byte:02X}")), + } + } + out +} + +fn host_from_endpoint(endpoint: &str) -> String { + endpoint + .split_once("://") + .map(|(_, rest)| rest) + .unwrap_or(endpoint) + .split('/') + .next() + .unwrap_or("") + .to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone as _; + + #[test] + fn builds_the_documented_object_key() { + assert_eq!( + replay_object_key("org_123", "sess_abc", 7), + "v1/org_123/sess_abc/00000007.json.gz" + ); + } + + #[test] + fn zero_pads_chunk_seq_so_lexicographic_order_is_playback_order() { + let mut keys = vec![ + replay_object_key("o", "s", 10), + replay_object_key("o", "s", 2), + replay_object_key("o", "s", 1), + ]; + keys.sort(); + assert_eq!( + keys, + vec![ + "v1/o/s/00000001.json.gz", + "v1/o/s/00000002.json.gz", + "v1/o/s/00000010.json.gz", + ] + ); + } + + #[test] + fn chunk_seq_above_the_padding_width_still_sorts_after() { + // u32::MAX is 10 digits, so it overflows the :08 pad. Ordering only has + // to hold within a session, and a session cannot reach 100M chunks + // (the SDK flushes at most every 5s), but assert the format doesn't + // truncate. + assert_eq!( + replay_object_key("o", "s", 123_456_789), + "v1/o/s/123456789.json.gz" + ); + } + + #[test] + fn encodes_reserved_characters_in_the_path() { + assert_eq!(uri_encode_path("/test$file.text"), "/test%24file.text"); + assert_eq!(uri_encode_path("v1/a-b_c.d~e/f"), "v1/a-b_c.d~e/f"); + assert_eq!(uri_encode_path("a b"), "a%20b"); + } + + #[test] + fn extracts_host_from_endpoint() { + assert_eq!( + host_from_endpoint("https://acct.r2.cloudflarestorage.com"), + "acct.r2.cloudflarestorage.com" + ); + assert_eq!(host_from_endpoint("http://127.0.0.1:8080"), "127.0.0.1:8080"); + } + + // AWS SigV4 published test vector — "Signature Calculation: Transfer Payload + // in a Single Chunk", PUT Object example from the S3 REST API docs. Uses the + // canonical AKIAIOSFODNN7EXAMPLE credentials. A signing bug is otherwise a + // 403 with an opaque body that you debug in staging, so pin the whole + // Authorization value against a known-good vector. + const AWS_EXAMPLE_KEY_ID: &str = "AKIAIOSFODNN7EXAMPLE"; + const AWS_EXAMPLE_SECRET: &str = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"; + + // `signing_key` is covered transitively and non-circularly by the vector + // below: that signature cannot match unless all four derivation stages are + // right. A standalone assertion on the intermediate key would have to pin a + // hex string we produced ourselves, which proves nothing. + #[test] + fn matches_the_aws_put_object_vector() { + let now = Utc.with_ymd_and_hms(2013, 5, 24, 0, 0, 0).unwrap(); + let payload_sha256 = + "44ce7dd67c959e0d3524ffac1771dfbba87d2b6b4b4e99e42034a8b803f8b072".to_string(); + let headers = vec![ + ( + "date".to_string(), + "Fri, 24 May 2013 00:00:00 GMT".to_string(), + ), + ("host".to_string(), "examplebucket.s3.amazonaws.com".to_string()), + ("x-amz-content-sha256".to_string(), payload_sha256.clone()), + ("x-amz-date".to_string(), "20130524T000000Z".to_string()), + ( + "x-amz-storage-class".to_string(), + "REDUCED_REDUNDANCY".to_string(), + ), + ]; + + let authorization = authorization_header(&SigningParams { + method: "PUT", + canonical_uri: "/test%24file.text", + canonical_query: "", + headers: &headers, + payload_sha256: &payload_sha256, + now, + access_key_id: AWS_EXAMPLE_KEY_ID, + secret_access_key: AWS_EXAMPLE_SECRET, + region: "us-east-1", + service: "s3", + }); + + assert_eq!( + authorization, + "AWS4-HMAC-SHA256 \ + Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request, \ + SignedHeaders=date;host;x-amz-content-sha256;x-amz-date;x-amz-storage-class, \ + Signature=98ad721746da40c64f1a55b78f14c238d841ea1380cd77a1b5971af0ece108bd" + ); + } + + #[test] + fn sorts_headers_regardless_of_call_order() { + let now = Utc.with_ymd_and_hms(2013, 5, 24, 0, 0, 0).unwrap(); + let sha = "44ce7dd67c959e0d3524ffac1771dfbba87d2b6b4b4e99e42034a8b803f8b072".to_string(); + let build = |headers: Vec<(String, String)>| { + authorization_header(&SigningParams { + method: "PUT", + canonical_uri: "/bucket/key", + canonical_query: "", + headers: &headers, + payload_sha256: &sha, + now, + access_key_id: AWS_EXAMPLE_KEY_ID, + secret_access_key: AWS_EXAMPLE_SECRET, + region: "auto", + service: "s3", + }) + }; + let forward = build(vec![ + ("host".to_string(), "h".to_string()), + ("x-amz-date".to_string(), "20130524T000000Z".to_string()), + ]); + let reversed = build(vec![ + ("x-amz-date".to_string(), "20130524T000000Z".to_string()), + ("HOST".to_string(), "h".to_string()), + ]); + assert_eq!(forward, reversed); + } +} 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 98079e4ec..b20d5b5d7 100644 --- a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts +++ b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts @@ -250,8 +250,8 @@ export const getSessionTraceSummariesResultAtom = makeQueryAtomFamily(getSession // 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. Events come straight from ClickHouse — no R2, no -// signed URLs, no client-side fetch/gunzip. +// 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, { staleTime: 240_000, }) diff --git a/lib/effect-cloudflare/package.json b/lib/effect-cloudflare/package.json index 16be82c39..b0adcd384 100644 --- a/lib/effect-cloudflare/package.json +++ b/lib/effect-cloudflare/package.json @@ -7,7 +7,8 @@ "./config-helpers": "./src/config-helpers.ts", "./worker-environment": "./src/worker-environment.ts", "./hyperdrive-connection": "./src/hyperdrive-connection.ts", - "./workers-cache": "./src/workers-cache.ts" + "./workers-cache": "./src/workers-cache.ts", + "./r2-bucket": "./src/r2-bucket.ts" }, "scripts": { "typecheck": "tsc --noEmit" diff --git a/lib/effect-cloudflare/src/r2-bucket.ts b/lib/effect-cloudflare/src/r2-bucket.ts index bb70ac74a..bf0ab2ea4 100644 --- a/lib/effect-cloudflare/src/r2-bucket.ts +++ b/lib/effect-cloudflare/src/r2-bucket.ts @@ -7,6 +7,7 @@ import type * as runtime from "@cloudflare/workers-types" import * as Data from "effect/Data" import * as Effect from "effect/Effect" +import * as Option from "effect/Option" import * as Stream from "effect/Stream" import { WorkerEnvironment } from "./worker-environment.ts" @@ -193,4 +194,29 @@ const makeClient = (token: R2BucketToken): R2BucketClient => { export const R2Bucket = Object.assign((logicalId: string): R2BucketToken => makeToken(logicalId), { bind: (token: R2BucketToken): Effect.Effect => Effect.succeed(makeClient(token)), + /** + * `Some` only when the binding actually exists on the worker env. + * + * `bind` resolves lazily, so it hands back a client whose every call fails + * at use time when the binding is missing. That is the wrong shape for a + * deployment where absence is *expected* rather than a misconfiguration — + * the API also ships as a Docker image, and self-hosted installs have no + * R2 at all. Those callers need to branch on presence once, at layer + * construction, not rescue an error per request. + */ + bindOptional: ( + token: R2BucketToken, + ): Effect.Effect, never, WorkerEnvironment> => + WorkerEnvironment.pipe( + Effect.map((env) => { + // The env itself can be absent, not just the binding: outside a + // Worker isolate `cloudflare-workers.ts` falls back to a stub, and + // under vitest that stub's `env` may be undefined rather than `{}`. + // Both mean the same thing here. + const bindings = env as Record | undefined | null + return bindings?.[token.LogicalId] === undefined + ? Option.none() + : Option.some(makeClient(token)) + }), + ), }) diff --git a/packages/domain/src/generated/clickhouse-schema.ts b/packages/domain/src/generated/clickhouse-schema.ts index 1ce56b3e0..75ecf5b70 100644 --- a/packages/domain/src/generated/clickhouse-schema.ts +++ b/packages/domain/src/generated/clickhouse-schema.ts @@ -1,7 +1,7 @@ // This file is generated by scripts/generate-clickhouse-schema.ts // Do not edit manually. -export const projectRevision = "53294ef75d1afb2e4c9ce6ba2e80e9900e4996e6731838a1ce1902803588c58d" as const +export const projectRevision = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06" as const export const latestSnapshotStatements: ReadonlyArray = [ "CREATE TABLE IF NOT EXISTS alert_checks (\n OrgId LowCardinality(String),\n RuleId String,\n GroupKey String,\n Timestamp DateTime64(3),\n Status LowCardinality(String),\n SignalType LowCardinality(String),\n Comparator LowCardinality(String),\n Threshold Float64,\n ObservedValue Nullable(Float64),\n SampleCount UInt32,\n WindowMinutes UInt16,\n WindowStart DateTime64(3),\n WindowEnd DateTime64(3),\n ConsecutiveBreaches UInt16,\n ConsecutiveHealthy UInt16,\n IncidentId Nullable(String),\n IncidentTransition LowCardinality(String),\n EvaluationDurationMs UInt32,\n ErrorMessage Nullable(String),\n ErrorCategory LowCardinality(String)\n)\nENGINE = MergeTree\nPARTITION BY toDate(Timestamp)\nORDER BY (OrgId, RuleId, GroupKey, Timestamp)\nTTL toDate(Timestamp) + INTERVAL 365 DAY", diff --git a/packages/domain/src/generated/tinybird-project-manifest.ts b/packages/domain/src/generated/tinybird-project-manifest.ts index 46580ecc6..35c953b8a 100644 --- a/packages/domain/src/generated/tinybird-project-manifest.ts +++ b/packages/domain/src/generated/tinybird-project-manifest.ts @@ -1,7 +1,7 @@ // This file is generated by scripts/generate-tinybird-project-manifest.ts // Do not edit manually. -export const projectRevision = "53294ef75d1afb2e4c9ce6ba2e80e9900e4996e6731838a1ce1902803588c58d" as const +export const projectRevision = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06" as const export const datasources = [ { @@ -142,7 +142,7 @@ export const datasources = [ { name: "session_replay_events", content: - 'DESCRIPTION >\n Session replay rrweb events (one row per chunk, payload included). The ingest gateway gunzips the chunk and stores the event-array JSON in `Events`. Playback reads directly from ClickHouse — no R2.\n\nSCHEMA >\n OrgId LowCardinality(String) `json:$.org_id`,\n SessionId String `json:$.session_id`,\n ChunkSeq UInt32 `json:$.chunk_seq`,\n Timestamp DateTime64(9) `json:$.timestamp`,\n DurationMs UInt32 `json:$.duration_ms` DEFAULT 0,\n EventCount UInt32 `json:$.event_count` DEFAULT 0,\n ByteSize UInt32 `json:$.byte_size` DEFAULT 0,\n Events String `json:$.events`,\n IsCheckpoint UInt8 `json:$.is_checkpoint` DEFAULT 0\n\nENGINE "MergeTree"\nENGINE_PARTITION_KEY "toDate(Timestamp)"\nENGINE_SORTING_KEY "OrgId, SessionId, ChunkSeq"\nENGINE_TTL "toDate(Timestamp) + INTERVAL 30 DAY"', + 'DESCRIPTION >\n Session replay rrweb events, one row per chunk. `Events` carries the event-array JSON inline for pre-cutover rows and for deployments without a blob store; otherwise it is empty and the payload lives in R2 under v1/{OrgId}/{SessionId}/{ChunkSeq}.json.gz.\n\nSCHEMA >\n OrgId LowCardinality(String) `json:$.org_id`,\n SessionId String `json:$.session_id`,\n ChunkSeq UInt32 `json:$.chunk_seq`,\n Timestamp DateTime64(9) `json:$.timestamp`,\n DurationMs UInt32 `json:$.duration_ms` DEFAULT 0,\n EventCount UInt32 `json:$.event_count` DEFAULT 0,\n ByteSize UInt32 `json:$.byte_size` DEFAULT 0,\n Events String `json:$.events`,\n IsCheckpoint UInt8 `json:$.is_checkpoint` DEFAULT 0\n\nENGINE "MergeTree"\nENGINE_PARTITION_KEY "toDate(Timestamp)"\nENGINE_SORTING_KEY "OrgId, SessionId, ChunkSeq"\nENGINE_TTL "toDate(Timestamp) + INTERVAL 30 DAY"', }, { name: "session_replays", diff --git a/packages/domain/src/http/session-replay.ts b/packages/domain/src/http/session-replay.ts index 7bea5c3b5..ee70337e3 100644 --- a/packages/domain/src/http/session-replay.ts +++ b/packages/domain/src/http/session-replay.ts @@ -9,9 +9,10 @@ import { warehouseHttpErrors } from "./warehouse" // --------------------------------------------------------------------------- // Session replay endpoint schemas // -// Backed by the session_replays (metadata) + session_replay_events (rrweb event -// payloads) datasources, both in ClickHouse. `getReplayEvents` returns the rrweb -// event arrays inline (read straight from the warehouse — no R2, no signed URLs). +// Backed by the session_replays (metadata) + session_replay_events (chunk index) +// datasources in ClickHouse. `getReplayEvents` returns the rrweb event arrays +// inline; the API hydrates them from R2 first when the row is blob-backed, so +// the wire shape is the same either way — no signed URLs, no client-side fetch. // --------------------------------------------------------------------------- // --- List --- diff --git a/packages/domain/src/tinybird/datasources.ts b/packages/domain/src/tinybird/datasources.ts index 7e11f45a3..2f42ecfb0 100644 --- a/packages/domain/src/tinybird/datasources.ts +++ b/packages/domain/src/tinybird/datasources.ts @@ -1644,10 +1644,9 @@ export type LogsAggregatesHourlyRow = InferRow * row (not just the ended one) — otherwise a tab killed without an unload beacon * leaves only the v1 row's zeroes, and bounce rate reads 100%. * - * The rrweb event payloads live in `sessionReplayEvents` (one row per chunk, - * payload inline in ClickHouse — there is no R2 blob store); this table only - * holds small, queryable metadata so the sessions list/filter views never - * touch the multi-MB rrweb blobs. + * The rrweb event payloads live in `sessionReplayEvents` (one row per chunk); + * this table only holds small, queryable metadata so the sessions list/filter + * views never touch the multi-MB rrweb blobs. * * `TraceIds` carries the OTel trace ids observed during the session — the * correlation key that lets the trace detail view link to a replay and back. @@ -1806,12 +1805,15 @@ export const sessionReplays = defineDatasource("session_replays", { export type SessionReplaysRow = InferRow /** - * Session replay events — one row per uploaded rrweb chunk, payload included. + * Session replay events — one row per uploaded rrweb chunk. * - * The ingest gateway gunzips the chunk body and writes the rrweb event array - * JSON into `Events` (a String column ClickHouse ZSTD-compresses). Playback - * reads chunks back directly from here — there is no R2 blob store on the - * replay path. + * `Events` holds the rrweb event array as JSON text, but only for chunks + * written before the R2 cutover and for deployments with no blob store + * configured (self-hosted, BYO-ClickHouse). On the managed path the gateway + * stores the chunk's gzip in R2 under a key derived from + * `(OrgId, SessionId, ChunkSeq)` and writes `Events = ''`; the API refills it + * on read. So an empty `Events` means "blob-backed", never "empty chunk" — the + * SDK never uploads a chunk with no events. * * `IsCheckpoint=1` marks chunks that contain a full rrweb DOM snapshot, so the * player can seek to a timestamp by loading the nearest preceding checkpoint @@ -1823,7 +1825,7 @@ export type SessionReplaysRow = InferRow */ export const sessionReplayEvents = defineDatasource("session_replay_events", { description: - "Session replay rrweb events (one row per chunk, payload included). The ingest gateway gunzips the chunk and stores the event-array JSON in `Events`. Playback reads directly from ClickHouse — no R2.", + "Session replay rrweb events, one row per chunk. `Events` carries the event-array JSON inline for pre-cutover rows and for deployments without a blob store; otherwise it is empty and the payload lives in R2 under v1/{OrgId}/{SessionId}/{ChunkSeq}.json.gz.", schema: { OrgId: column(t.string().lowCardinality(), { jsonPath: "$.org_id" }), SessionId: column(t.string(), { jsonPath: "$.session_id" }), diff --git a/packages/query-engine/src/ch/tables.ts b/packages/query-engine/src/ch/tables.ts index 828f5076d..a3fe4e225 100644 --- a/packages/query-engine/src/ch/tables.ts +++ b/packages/query-engine/src/ch/tables.ts @@ -574,8 +574,9 @@ export const SessionReplayEvents = table("session_replay_events", { DurationMs: T.uint32, EventCount: T.uint32, ByteSize: T.uint32, - // The rrweb event array for this chunk, stored as a JSON string. ClickHouse - // ZSTD-compresses this column; playback reads it back directly (no R2). + // The rrweb event array for this chunk as a JSON string — inline for + // pre-cutover rows and blob-store-less deployments, empty when the payload + // lives in R2 (the API refills it on read). Events: T.string, IsCheckpoint: T.uint8, }) From bf9ebbaf5dedd60aebc76deff31b6c8bb8af80d5 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 4 Aug 2026 18:25:23 +0200 Subject: [PATCH 2/2] fix(replays): expire every replay object, not just the v1 prefix The lifecycle rule was pinned to prefix "v1/", but replay_object_key versions its scheme on purpose so a future format change can write under a new prefix while the old one ages out. The first such change would have silently stopped expiring anything, and the bucket would grow forever with no failing test to catch it. The bucket is single-purpose, so the rule covers all of it. --- apps/api/alchemy.run.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/api/alchemy.run.ts b/apps/api/alchemy.run.ts index 4b5d051b7..29ed7ffd4 100644 --- a/apps/api/alchemy.run.ts +++ b/apps/api/alchemy.run.ts @@ -148,11 +148,16 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) => // the one failure mode with no good client-side handling. const replayBlobs = yield* Cloudflare.R2.Bucket("replay-blobs", { name: resolveWorkerName("replay-blobs", stage), + // Deliberately unprefixed, so the rule covers whatever key scheme is + // current. `replay_object_key` is versioned (`v1/…`) precisely so a + // format change can write under a new prefix while the old one ages + // out — a rule pinned to `v1/` would silently stop expiring anything + // the moment that happens, and the bucket would grow forever with no + // failing test to catch it. Nothing else writes here. lifecycleRules: [ { id: "expire-replay-chunks", enabled: true, - prefix: "v1/", deleteObjectsTransition: { condition: { type: "Age", maxAge: 32 * 24 * 60 * 60 } }, }, ],