feat(session-ingest): add feature-flagged direct ingest

services/session-ingest/.dev.vars.example

@@ -0,0 +1,3 @@
+DIRECT_INGEST_PERCENT="0"
+DIRECT_INGEST_USER_IDS=""
+DIRECT_INGEST_MAX_BYTES="4194304"

services/session-ingest/package.json

@@ -4,7 +4,7 @@
   "scripts": {
     "dev": "wrangler dev",
     "deploy": "wrangler deploy --minify",
-    "cf-typegen": "wrangler types --env-interface CloudflareBindings",
+    "cf-typegen": "wrangler types --include-runtime=false --env-interface CloudflareBindings",
     "test": "vitest run",
     "test:watch": "vitest",
     "test:integration": "vitest run --config vitest.workers.config.ts",

services/session-ingest/src/env.ts

@@ -1,7 +1,17 @@
 import type { NotificationsBinding } from './notifications-binding.js';
 import type { O11YBinding } from './o11y-binding.js';
 
-export type Env = Omit<Cloudflare.Env, 'O11Y' | 'NOTIFICATIONS'> & {
+export type Env = Omit<
+  Cloudflare.Env,
+  | 'O11Y'
+  | 'NOTIFICATIONS'
+  | 'DIRECT_INGEST_PERCENT'
+  | 'DIRECT_INGEST_USER_IDS'
+  | 'DIRECT_INGEST_MAX_BYTES'
+> & {
   O11Y: O11YBinding;
   NOTIFICATIONS: NotificationsBinding;
+  DIRECT_INGEST_PERCENT: string;
+  DIRECT_INGEST_USER_IDS: string;
+  DIRECT_INGEST_MAX_BYTES: string;
 };

services/session-ingest/src/ingest/bounded-stream-reader.test.ts

@@ -0,0 +1,99 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { readBoundedStream } from './bounded-stream-reader';
+
+function streamFromChunks(chunks: number[][], cancel = vi.fn()) {
+  let nextChunk = 0;
+  return {
+    stream: new ReadableStream<Uint8Array>(
+      {
+        pull(controller) {
+          const chunk = chunks[nextChunk];
+          if (chunk === undefined) {
+            controller.close();
+            return;
+          }
+          nextChunk += 1;
+          controller.enqueue(Uint8Array.from(chunk));
+        },
+        cancel,
+      },
+      { highWaterMark: 0 }
+    ),
+    cancel,
+  };
+}
+
+describe('readBoundedStream', () => {
+  it('accepts bytes exactly at both limits', async () => {
+    const { stream, cancel } = streamFromChunks([
+      [1, 2],
+      [3, 4],
+    ]);
+
+    await expect(readBoundedStream(stream, 4, 4)).resolves.toEqual({
+      ok: true,
+      bytes: Uint8Array.from([1, 2, 3, 4]),
+    });
+    expect(cancel).not.toHaveBeenCalled();
+  });
+
+  it('rejects actual bytes over the declared size', async () => {
+    const { stream } = streamFromChunks([[1, 2, 3]]);
+
+    await expect(readBoundedStream(stream, 2, 10)).resolves.toEqual({
+      ok: false,
+      reason: 'too_large',
+      limit: 'declared_bytes',
+    });
+  });
+
+  it('rejects a declaration over the configured cap without consuming bytes', async () => {
+    let pulled = false;
+    const cancel = vi.fn();
+    const stream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        pulled = true;
+        controller.enqueue(Uint8Array.from([1]));
+      },
+      cancel,
+    });
+
+    await expect(readBoundedStream(stream, 11, 10)).resolves.toEqual({
+      ok: false,
+      reason: 'too_large',
+      limit: 'configured_cap',
+    });
+    expect(pulled).toBe(false);
+    expect(cancel).toHaveBeenCalledOnce();
+  });
+
+  it('detects overflow accumulated across multiple chunks', async () => {
+    const { stream, cancel } = streamFromChunks([[1, 2], [3, 4], [5]]);
+
+    await expect(readBoundedStream(stream, 4, 10)).resolves.toEqual({
+      ok: false,
+      reason: 'too_large',
+      limit: 'declared_bytes',
+    });
+    expect(cancel).toHaveBeenCalledOnce();
+  });
+
+  it('cancels on actual-byte overflow and releases the reader lock', async () => {
+    const { stream, cancel } = streamFromChunks([[1, 2, 3]]);
+
+    await readBoundedStream(stream, 2, 10);
+
+    expect(cancel).toHaveBeenCalledOnce();
+    expect(() => stream.getReader()).not.toThrow();
+  });
+
+  it('preserves every byte and chunk ordering', async () => {
+    const { stream } = streamFromChunks([[0, 255], [], [17, 42, 128]]);
+
+    await expect(readBoundedStream(stream, 5, 10)).resolves.toEqual({
+      ok: true,
+      bytes: Uint8Array.from([0, 255, 17, 42, 128]),
+    });
+  });
+});

services/session-ingest/src/ingest/bounded-stream-reader.ts

@@ -0,0 +1,58 @@
+export type BoundedStreamReadResult =
+  | { ok: true; bytes: Uint8Array }
+  | { ok: false; reason: 'too_large'; limit: 'declared_bytes' | 'configured_cap' };
+
+export async function readBoundedStream(
+  stream: ReadableStream<Uint8Array>,
+  declaredBytes: number,
+  configuredCap: number
+): Promise<BoundedStreamReadResult> {
+  const reader = stream.getReader();
+
+  try {
+    if (declaredBytes > configuredCap) {
+      await cancelReader(reader);
+      return { ok: false, reason: 'too_large', limit: 'configured_cap' };
+    }
+
+    const chunks: Uint8Array[] = [];
+    let totalBytes = 0;
+
+    while (true) {
+      const result = await reader.read();
+      if (result.done) {
+        const bytes = new Uint8Array(totalBytes);
+        let offset = 0;
+        for (const chunk of chunks) {
+          bytes.set(chunk, offset);
+          offset += chunk.byteLength;
+        }
+        return { ok: true, bytes };
+      }
+
+      totalBytes += result.value.byteLength;
+      if (totalBytes > declaredBytes || totalBytes > configuredCap) {
+        await cancelReader(reader);
+        return {
+          ok: false,
+          reason: 'too_large',
+          // Route callers reject declaredBytes > configuredCap before reading, so the
+          // streaming overflow normally reports declared_bytes. The cap branch keeps
+          // this helper defensive for standalone callers.
+          limit: totalBytes > configuredCap ? 'configured_cap' : 'declared_bytes',
+        };
+      }
+      chunks.push(result.value);
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}
+
+async function cancelReader(reader: ReadableStreamDefaultReader<Uint8Array>): Promise<void> {
+  try {
+    await reader.cancel('stream exceeded ingest byte limit');
+  } catch {
+    // The typed overflow result remains authoritative if cancellation itself fails.
+  }
+}

services/session-ingest/src/ingest/direct-ingest-rollout.test.ts

@@ -0,0 +1,108 @@
+import { describe, expect, it } from 'vitest';
+
+import {
+  getDirectIngestUserBucket,
+  parseDirectIngestConfig,
+  selectDirectIngestUser,
+  type DirectIngestConfig,
+} from './direct-ingest-rollout';
+
+function validInput() {
+  return {
+    DIRECT_INGEST_PERCENT: '25',
+    DIRECT_INGEST_MAX_BYTES: '4194304',
+    DIRECT_INGEST_USER_IDS: '',
+  };
+}
+
+function config(overrides: Partial<DirectIngestConfig> = {}): DirectIngestConfig {
+  return { percent: 25, maxBytes: 4_194_304, userIds: new Set(), ...overrides };
+}
+
+describe('parseDirectIngestConfig', () => {
+  it.each([undefined, '', ' ', '-1', '1.5', '01', '101', 'Infinity', '25 '])(
+    'fails closed for invalid percent %j',
+    percent => {
+      expect(parseDirectIngestConfig({ ...validInput(), DIRECT_INGEST_PERCENT: percent })).toEqual({
+        ok: false,
+        reason: 'invalid_percent',
+      });
+    }
+  );
+
+  it.each([undefined, '', ' ', '0', '-1', '1.5', '01', '4194305', 'Infinity', '9007199254740992'])(
+    'fails closed for invalid max bytes %j',
+    maxBytes => {
+      expect(
+        parseDirectIngestConfig({ ...validInput(), DIRECT_INGEST_MAX_BYTES: maxBytes })
+      ).toEqual({ ok: false, reason: 'invalid_max_bytes' });
+    }
+  );
+
+  it('accepts percent boundaries and the supported direct body cap', () => {
+    expect(
+      parseDirectIngestConfig({
+        ...validInput(),
+        DIRECT_INGEST_PERCENT: '0',
+        DIRECT_INGEST_MAX_BYTES: '4194304',
+      })
+    ).toMatchObject({ ok: true, config: { percent: 0, maxBytes: 4194304 } });
+
+    expect(
+      parseDirectIngestConfig({ ...validInput(), DIRECT_INGEST_PERCENT: '100' })
+    ).toMatchObject({ ok: true, config: { percent: 100 } });
+  });
+
+  it('trims comma-separated user IDs and removes empty and duplicate values', () => {
+    const result = parseDirectIngestConfig({
+      ...validInput(),
+      DIRECT_INGEST_USER_IDS: ' user-a, user-b ,,user-a, ',
+    });
+
+    expect(result.ok && [...result.config.userIds]).toEqual(['user-a', 'user-b']);
+  });
+
+  it('fails closed when the user ID setting is missing', () => {
+    expect(parseDirectIngestConfig({ ...validInput(), DIRECT_INGEST_USER_IDS: undefined })).toEqual(
+      { ok: false, reason: 'invalid_user_ids' }
+    );
+  });
+});
+
+describe('selectDirectIngestUser', () => {
+  it('selects no percentage users at zero percent', async () => {
+    await expect(selectDirectIngestUser(config({ percent: 0 }), 'user-a')).resolves.toEqual({
+      selected: false,
+      reason: 'not_selected',
+      bucket: null,
+    });
+  });
+
+  it('selects every user at 100 percent', async () => {
+    await expect(selectDirectIngestUser(config({ percent: 100 }), 'user-a')).resolves.toEqual({
+      selected: true,
+      reason: 'percentage',
+      bucket: null,
+    });
+  });
+
+  it('allows an allowlisted user at zero percent', async () => {
+    await expect(
+      selectDirectIngestUser(config({ percent: 0, userIds: new Set(['user-a']) }), 'user-a')
+    ).resolves.toEqual({ selected: true, reason: 'allowlist', bucket: null });
+  });
+
+  it('deterministically selects from the SHA-256 user bucket', async () => {
+    const bucket = 63;
+
+    await expect(getDirectIngestUserBucket('stable-user')).resolves.toBe(63);
+    await expect(getDirectIngestUserBucket('usr_test')).resolves.toBe(84);
+    await expect(getDirectIngestUserBucket('rollout-user-a')).resolves.toBe(13);
+    await expect(
+      selectDirectIngestUser(config({ percent: bucket + 1 }), 'stable-user')
+    ).resolves.toEqual({ selected: true, reason: 'percentage', bucket });
+    await expect(
+      selectDirectIngestUser(config({ percent: bucket }), 'stable-user')
+    ).resolves.toEqual({ selected: false, reason: 'not_selected', bucket });
+  });
+});

services/session-ingest/src/ingest/direct-ingest-rollout.ts

@@ -0,0 +1,91 @@
+import { INGEST_CHUNK_MAX_BYTES } from '../util/ingest-limits';
+
+export type DirectIngestConfigInput = {
+  DIRECT_INGEST_PERCENT: string | undefined;
+  DIRECT_INGEST_MAX_BYTES: string | undefined;
+  DIRECT_INGEST_USER_IDS: string | undefined;
+};
+
+export type DirectIngestConfig = {
+  percent: number;
+  maxBytes: number;
+  userIds: ReadonlySet<string>;
+};
+
+export type DirectIngestConfigErrorReason =
+  | 'invalid_percent'
+  | 'invalid_max_bytes'
+  | 'invalid_user_ids';
+
+export type DirectIngestConfigResult =
+  | { ok: true; config: DirectIngestConfig }
+  | { ok: false; reason: DirectIngestConfigErrorReason };
+
+export type DirectIngestSelectionResult =
+  | { selected: true; reason: 'allowlist' | 'percentage'; bucket: number | null }
+  | { selected: false; reason: 'not_selected'; bucket: number | null };
+
+const unsignedIntegerPattern = /^(0|[1-9]\d*)$/;
+
+export function parseDirectIngestConfig(input: DirectIngestConfigInput): DirectIngestConfigResult {
+  const percent = parseUnsignedInteger(input.DIRECT_INGEST_PERCENT);
+  if (percent === null || percent > 100) {
+    return { ok: false, reason: 'invalid_percent' };
+  }
+
+  const maxBytes = parseUnsignedInteger(input.DIRECT_INGEST_MAX_BYTES);
+  if (maxBytes === null || maxBytes === 0 || maxBytes > INGEST_CHUNK_MAX_BYTES) {
+    return { ok: false, reason: 'invalid_max_bytes' };
+  }
+
+  if (input.DIRECT_INGEST_USER_IDS === undefined) {
+    return { ok: false, reason: 'invalid_user_ids' };
+  }
+
+  const userIds = new Set(
+    input.DIRECT_INGEST_USER_IDS.split(',')
+      .map(userId => userId.trim())
+      .filter(userId => userId.length > 0)
+  );
+
+  return { ok: true, config: { percent, maxBytes, userIds } };
+}
+
+export async function selectDirectIngestUser(
+  config: DirectIngestConfig,
+  kiloUserId: string
+): Promise<DirectIngestSelectionResult> {
+  if (config.userIds.has(kiloUserId)) {
+    return { selected: true, reason: 'allowlist', bucket: null };
+  }
+
+  if (config.percent === 0) {
+    return { selected: false, reason: 'not_selected', bucket: null };
+  }
+
+  if (config.percent === 100) {
+    return { selected: true, reason: 'percentage', bucket: null };
+  }
+
+  const bucket = await getDirectIngestUserBucket(kiloUserId);
+  return bucket < config.percent
+    ? { selected: true, reason: 'percentage', bucket }
+    : { selected: false, reason: 'not_selected', bucket };
+}
+
+export async function getDirectIngestUserBucket(kiloUserId: string): Promise<number> {
+  const digest = new Uint8Array(
+    await crypto.subtle.digest('SHA-256', new TextEncoder().encode(kiloUserId))
+  );
+  const firstFourBytes = new DataView(digest.buffer, digest.byteOffset, 4).getUint32(0);
+  return firstFourBytes % 100;
+}
+
+function parseUnsignedInteger(value: string | undefined): number | null {
+  if (value === undefined || !unsignedIntegerPattern.test(value)) {
+    return null;
+  }
+
+  const parsed = Number(value);
+  return Number.isSafeInteger(parsed) ? parsed : null;
+}