adopt s2-lite, upgrade s2 package with support

Author: ericallam

apps/webapp/app/env.server.ts

@@ -1344,6 +1344,8 @@ const EnvironmentSchema = z
 
     REALTIME_STREAMS_S2_BASIN: z.string().optional(),
     REALTIME_STREAMS_S2_ACCESS_TOKEN: z.string().optional(),
+    REALTIME_STREAMS_S2_ENDPOINT: z.string().optional(),
+    REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS: z.enum(["true", "false"]).default("false"),
     REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS: z.coerce
       .number()
       .int()

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -10,6 +10,12 @@ export type S2RealtimeStreamsOptions = {
   accessToken: string; // "Bearer" token issued in S2 console
   streamPrefix?: string; // defaults to ""
 
+  // Custom endpoint for s2-lite (self-hosted)
+  endpoint?: string; // e.g., "http://localhost:4566/v1"
+
+  // Skip access token issuance (s2-lite doesn't support /access-tokens)
+  skipAccessTokens?: boolean;
+
   // Read behavior
   s2WaitSeconds?: number;
 
@@ -37,8 +43,11 @@ type S2AppendAck = {
 export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
   private readonly basin: string;
   private readonly baseUrl: string;
+  private readonly accountUrl: string;
+  private readonly endpoint?: string;
   private readonly token: string;
   private readonly streamPrefix: string;
+  private readonly skipAccessTokens: boolean;
 
   private readonly s2WaitSeconds: number;
 
@@ -56,9 +65,12 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
 
   constructor(opts: S2RealtimeStreamsOptions) {
     this.basin = opts.basin;
-    this.baseUrl = `https://${this.basin}.b.aws.s2.dev/v1`;
+    this.baseUrl = opts.endpoint ?? `https://${this.basin}.b.aws.s2.dev/v1`;
+    this.accountUrl = opts.endpoint ?? `https://aws.s2.dev/v1`;
+    this.endpoint = opts.endpoint;
     this.token = opts.accessToken;
     this.streamPrefix = opts.streamPrefix ?? "";
+    this.skipAccessTokens = opts.skipAccessTokens ?? false;
 
     this.s2WaitSeconds = opts.s2WaitSeconds ?? 60;
 
@@ -80,17 +92,20 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     runId: string,
     streamId: string
   ): Promise<{ responseHeaders?: Record<string, string> }> {
-    const id = randomUUID();
-
-    const accessToken = await this.getS2AccessToken(id);
+    const accessToken = this.skipAccessTokens
+      ? this.token
+      : await this.getS2AccessToken(randomUUID());
 
     return {
       responseHeaders: {
         "X-S2-Access-Token": accessToken,
-        "X-S2-Stream-Name": `/runs/${runId}/${streamId}`,
+        "X-S2-Stream-Name": this.skipAccessTokens
+          ? this.toStreamName(runId, streamId)
+          : `/runs/${runId}/${streamId}`,
         "X-S2-Basin": this.basin,
         "X-S2-Flush-Interval-Ms": this.flushIntervalMs.toString(),
         "X-S2-Max-Retries": this.maxRetries.toString(),
+        ...(this.endpoint ? { "X-S2-Endpoint": this.endpoint } : {}),
       },
     };
   }
@@ -142,6 +157,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
           Authorization: `Bearer ${this.token}`,
           Accept: "text/event-stream",
           "S2-Format": "raw",
+          "S2-Basin": this.basin,
         },
       }
     );
@@ -236,7 +252,8 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
       headers: {
         Authorization: `Bearer ${this.token}`,
         "Content-Type": "application/json",
-        "S2-Format": "raw", // UTF-8 JSON encoding (no base64 overhead) when your data is text. :contentReference[oaicite:8]{index=8}
+        "S2-Format": "raw",
+        "S2-Basin": this.basin,
       },
       body: JSON.stringify(body),
     });
@@ -265,7 +282,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
 
   private async s2IssueAccessToken(id: string): Promise<string> {
     // POST /v1/access-tokens
-    const res = await fetch(`https://aws.s2.dev/v1/access-tokens`, {
+    const res = await fetch(`${this.accountUrl}/access-tokens`, {
       method: "POST",
       headers: {
         Authorization: `Bearer ${this.token}`,
@@ -316,6 +333,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
         Authorization: `Bearer ${this.token}`,
         Accept: "text/event-stream",
         "S2-Format": "raw",
+        "S2-Basin": this.basin,
       },
       signal: opts.signal,
     });

apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts

@@ -40,6 +40,8 @@ export function getRealtimeStreamInstance(
       return new S2RealtimeStreams({
         basin: env.REALTIME_STREAMS_S2_BASIN,
         accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN,
+        endpoint: env.REALTIME_STREAMS_S2_ENDPOINT,
+        skipAccessTokens: env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true",
         streamPrefix: [
           "org",
           environment.organization.id,

apps/webapp/app/v3/services/deployment.server.ts

@@ -368,7 +368,7 @@ export class DeploymentService extends BaseService {
     );
 
     return fromPromise(
-      stream.append(events.map((event) => AppendRecord.make(JSON.stringify(event)))),
+      stream.append(events.map((event) => AppendRecord.string({ body: JSON.stringify(event) }))),
       (error) => ({
         type: "failed_to_append_to_event_log" as const,
         cause: error,

apps/webapp/package.json

@@ -107,7 +107,7 @@
     "@remix-run/serve": "2.1.0",
     "@remix-run/server-runtime": "2.1.0",
     "@remix-run/v1-meta": "^0.1.3",
-    "@s2-dev/streamstore": "^0.17.2",
+    "@s2-dev/streamstore": "^0.22.5",
     "@sentry/remix": "9.46.0",
     "@slack/web-api": "7.9.1",
     "@socket.io/redis-adapter": "^8.3.0",

docker/docker-compose.yml

@@ -143,6 +143,30 @@ services:
     networks:
       - app_network
 
+  s2:
+    image: ghcr.io/s2-streamstore/s2
+    command: lite
+    ports:
+      - "4566:80"
+    networks:
+      - app_network
+    healthcheck:
+      test: ["CMD-SHELL", "wget -qO- http://localhost:80/v1/basins?limit=1 || exit 1"]
+      interval: 2s
+      timeout: 3s
+      retries: 5
+      start_period: 3s
+
+  s2-init:
+    image: curlimages/curl:latest
+    depends_on:
+      s2:
+        condition: service_healthy
+    networks:
+      - app_network
+    restart: "no"
+    entrypoint: ["sh", "-c", "curl -sf -X PUT http://s2:80/v1/basins/trigger-local -H 'Content-Type: application/json' -d '{\"config\":{\"create_stream_on_append\":true,\"create_stream_on_read\":true}}' && echo ' Basin trigger-local ready' || echo ' Basin trigger-local already exists'"]
+
   toxiproxy:
     container_name: toxiproxy
     image: ghcr.io/shopify/toxiproxy:latest

packages/core/package.json

@@ -177,16 +177,16 @@
     "@opentelemetry/core": "2.0.1",
     "@opentelemetry/exporter-logs-otlp-http": "0.203.0",
     "@opentelemetry/exporter-metrics-otlp-http": "0.203.0",
-    "@opentelemetry/host-metrics": "^0.37.0",
     "@opentelemetry/exporter-trace-otlp-http": "0.203.0",
+    "@opentelemetry/host-metrics": "^0.37.0",
     "@opentelemetry/instrumentation": "0.203.0",
     "@opentelemetry/resources": "2.0.1",
     "@opentelemetry/sdk-logs": "0.203.0",
     "@opentelemetry/sdk-metrics": "2.0.1",
     "@opentelemetry/sdk-trace-base": "2.0.1",
     "@opentelemetry/sdk-trace-node": "2.0.1",
     "@opentelemetry/semantic-conventions": "1.36.0",
-    "@s2-dev/streamstore": "0.17.3",
+    "@s2-dev/streamstore": "0.22.5",
     "dequal": "^2.0.3",
     "eventsource": "^3.0.5",
     "eventsource-parser": "^3.0.0",

packages/core/src/v3/realtimeStreams/streamInstance.ts

@@ -52,6 +52,7 @@ export class StreamInstance<T> implements StreamsWriter {
             basin: parsedResponse.basin,
             stream: parsedResponse.streamName ?? this.options.key,
             accessToken: parsedResponse.accessToken,
+            endpoint: parsedResponse.endpoint,
             source: this.options.source,
             signal: this.options.signal,
             debug: this.options.debug,
@@ -103,6 +104,7 @@ type ParsedStreamResponse =
       version: "v2";
       accessToken: string;
       basin: string;
+      endpoint?: string;
       flushIntervalMs?: number;
       maxRetries?: number;
       streamName?: string;
@@ -123,6 +125,7 @@ function parseCreateStreamResponse(
     return { version: "v1" };
   }
 
+  const endpoint = headers?.["x-s2-endpoint"];
   const flushIntervalMs = headers?.["x-s2-flush-interval-ms"];
   const maxRetries = headers?.["x-s2-max-retries"];
   const streamName = headers?.["x-s2-stream-name"];
@@ -131,6 +134,7 @@ function parseCreateStreamResponse(
     version: "v2",
     accessToken,
     basin,
+    endpoint,
     flushIntervalMs: flushIntervalMs ? parseInt(flushIntervalMs) : undefined,
     maxRetries: maxRetries ? parseInt(maxRetries) : undefined,
     streamName,

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -6,12 +6,13 @@ export type StreamsWriterV2Options<T = any> = {
   basin: string;
   stream: string;
   accessToken: string;
+  endpoint?: string; // Custom S2 endpoint (for s2-lite)
   source: ReadableStream<T>;
   signal?: AbortSignal;
   flushIntervalMs?: number; // Used as lingerDuration for BatchTransform (default 200ms)
   maxRetries?: number; // Not used with appendSession, kept for compatibility
   debug?: boolean; // Enable debug logging (default false)
-  maxQueuedBytes?: number; // Max queued bytes for appendSession (default 10MB)
+  maxInflightBytes?: number; // Max queued bytes for appendSession (default 10MB)
 };
 
 /**
@@ -50,18 +51,28 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
   private streamPromise: Promise<void>;
   private readonly flushIntervalMs: number;
   private readonly debug: boolean;
-  private readonly maxQueuedBytes: number;
+  private readonly maxInflightBytes: number;
   private aborted = false;
   private sessionWritable: WritableStream<any> | null = null;
 
   constructor(private options: StreamsWriterV2Options<T>) {
     this.debug = options.debug ?? false;
-    this.s2Client = new S2({ accessToken: options.accessToken });
+    this.s2Client = new S2({
+      accessToken: options.accessToken,
+      ...(options.endpoint
+        ? {
+            endpoints: {
+              account: options.endpoint,
+              basin: options.endpoint,
+            },
+          }
+        : {}),
+    });
     this.flushIntervalMs = options.flushIntervalMs ?? 200;
-    this.maxQueuedBytes = options.maxQueuedBytes ?? 1024 * 1024 * 10; // 10MB default
+    this.maxInflightBytes = options.maxInflightBytes ?? 1024 * 1024 * 10; // 10MB default
 
     this.log(
-      `[S2MetadataStream] Initializing: basin=${options.basin}, stream=${options.stream}, flushIntervalMs=${this.flushIntervalMs}, maxQueuedBytes=${this.maxQueuedBytes}`
+      `[S2MetadataStream] Initializing: basin=${options.basin}, stream=${options.stream}, flushIntervalMs=${this.flushIntervalMs}, maxInflightBytes=${this.maxInflightBytes}`
     );
 
     // Check if already aborted
@@ -124,7 +135,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       const stream = basin.stream(this.options.stream);
 
       const session = await stream.appendSession({
-        maxQueuedBytes: this.maxQueuedBytes,
+        maxInflightBytes: this.maxInflightBytes,
       });
 
       this.sessionWritable = session.writable;
@@ -141,7 +152,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
                 return;
               }
               // Convert each chunk to JSON string and wrap in AppendRecord
-              controller.enqueue(AppendRecord.make(JSON.stringify({ data: chunk, id: nanoid(7) })));
+              controller.enqueue(AppendRecord.string({ body: JSON.stringify({ data: chunk, id: nanoid(7) }) }));
             },
           })
         )
@@ -158,9 +169,9 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       const lastAcked = session.lastAckedPosition();
 
       if (lastAcked?.end) {
-        const recordsWritten = lastAcked.end.seq_num;
+        const recordsWritten = lastAcked.end.seqNum;
         this.log(
-          `[S2MetadataStream] Written ${recordsWritten} records, ending at seq_num=${lastAcked.end.seq_num}`
+          `[S2MetadataStream] Written ${recordsWritten} records, ending at seqNum=${lastAcked.end.seqNum}`
         );
       }
     } catch (error) {