feat: read session resumption

Author: infiniteregrets

.speakeasy/gen.lock

@@ -5,8 +5,8 @@ management:
   docVersion: 1.0.0
   speakeasyVersion: 1.487.0
   generationVersion: 2.506.0
-  releaseVersion: 0.3.7
-  configChecksum: cf61e0b559919d1742f2d4b5043ee292
+  releaseVersion: 0.3.8
+  configChecksum: f1f08ec2a1fe81cf49886a5f8d670405
   repoURL: https://github.com/s2-streamstore/s2-sdk-typescript.git
   installationURL: https://github.com/s2-streamstore/s2-sdk-typescript
   published: true

.speakeasy/gen.yaml

@@ -16,7 +16,7 @@ generation:
     oAuth2ClientCredentialsEnabled: true
     oAuth2PasswordEnabled: true
 typescript:
-  version: 0.3.7
+  version: 0.3.8
   additionalDependencies:
     dependencies:
       uuid: ^9.0.1

.speakeasy/workflow.lock

@@ -2,7 +2,7 @@ speakeasyVersion: 1.487.0
 sources:
     S2 API:
         sourceNamespace: s-2-api
-        sourceRevisionDigest: sha256:2f76d85136855c3f3c58d4d97488242b1e51bfc3412801901f1d0cc127b8df6c
+        sourceRevisionDigest: sha256:66a149fc312cb9f820ffd854f9309c69612ee8620c5dbde3c29c73dab3f04d36
         sourceBlobDigest: sha256:eda67f586b78ceb1300a3662669e008175f0667b68ff804d4de23fcab20998d1
         tags:
             - latest
@@ -11,10 +11,10 @@ targets:
     streamstore:
         source: S2 API
         sourceNamespace: s-2-api
-        sourceRevisionDigest: sha256:2f76d85136855c3f3c58d4d97488242b1e51bfc3412801901f1d0cc127b8df6c
+        sourceRevisionDigest: sha256:66a149fc312cb9f820ffd854f9309c69612ee8620c5dbde3c29c73dab3f04d36
         sourceBlobDigest: sha256:eda67f586b78ceb1300a3662669e008175f0667b68ff804d4de23fcab20998d1
         codeSamplesNamespace: s-2-api-typescript-code-samples
-        codeSamplesRevisionDigest: sha256:e1de59eedebdbc0f1ff721f1eb03212012f8b9e44ef4df4600cec80183b42221
+        codeSamplesRevisionDigest: sha256:363ebcc961d9681148a7bae02a1f0173fc1642cbfd957e6ad6006636443da852
 workflow:
     workflowVersion: 1.0.0
     speakeasyVersion: latest

jsr.json

@@ -2,7 +2,7 @@
 
 {
   "name": "@s2-dev/streamstore",
-  "version": "0.3.7",
+  "version": "0.3.8",
   "exports": {
     ".": "./src/index.ts",    
     "./models/errors": "./src/models/errors/index.ts",    

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@s2-dev/streamstore",
-  "version": "0.3.7",
+  "version": "0.3.8",
   "author": "Speakeasy",
   "main": "./index.js",
   "sideEffects": false,

package.json

@@ -9,10 +9,12 @@ import {
     Output,
     ReadResponse,
     ReadResponseOutput,
+    SequencedRecord,
+    SequencedRecordBatch,
     StreamConfig,
     StreamInfo,
 } from "./models/components";
-import { NotFoundError } from "./models/errors";
+import { ErrorResponse, NotFoundError } from "./models/errors";
 import {
     GetBasinConfigRequest,
     ListBasinsRequest,
@@ -263,16 +265,20 @@ class Stream {
 
     async *readStream(request: ReadRequest): AsyncGenerator<ReadResponse, void, undefined> {
         let currentRequest: ReadRequest = { ...request };
+        let backoffMs = 100;
+        const maxBackoffMs = 5000;
+        const maxRetries = 5;
+        let retryCount = 0;
 
         while (true) {
             let stream: EventStream<ReadResponse> | undefined;
             try {
                 const response = await this._stream.read(
                     { ...currentRequest, stream: this.streamName },
-                    { 
-                      serverURL: this.basinURL,
-                      timeoutMs: -1, // disable only for streaming
-                      acceptHeaderOverride: ReadAcceptEnum.textEventStream 
+                    {
+                        serverURL: this.basinURL,
+                        timeoutMs: -1, // disable only for streaming
+                        acceptHeaderOverride: ReadAcceptEnum.textEventStream
                     }
                 );
                 stream = response.readResponse;
@@ -281,7 +287,7 @@ class Stream {
                 for await (const event of stream) {
                     yield event;
 
-                    if (event.event === 'output') {
+                    if (event.event === 'message') {
                         const output = event as ReadResponseOutput;
                         if ('batch' in output.data) {
                             const batch = output.data.batch;
@@ -291,18 +297,43 @@ class Stream {
                                     currentRequest = { ...currentRequest, startSeqNum: lastRecord.seqNum + 1 };
                                 }
                             }
+                            if (currentRequest.limit) {
+                                if (currentRequest.limit.count != null) {
+                                    currentRequest.limit.count = Math.max(0, currentRequest.limit.count - batch.records.length);
+                                }
+                                if (currentRequest.limit.bytes != null) {
+                                    currentRequest.limit.bytes = Math.max(0, currentRequest.limit.bytes - meteredBatch(batch));
+                                }
+                            }
                         }
                     }
                 }
                 return;
             } catch (error) {
-                console.error("Error while reading stream", error);
-                return;
+                if (error instanceof ErrorResponse || error instanceof NotFoundError) return;
+                if (retryCount >= maxRetries) {
+                    throw error;
+                }
+                retryCount++;
+                await new Promise(resolve => setTimeout(resolve, backoffMs));
+                backoffMs = Math.min(backoffMs * 2, maxBackoffMs);
             }
         }
     }
 }
 
+function meteredRecord(batch: SequencedRecord): number {
+    const fixed = 8 + (2 * batch.headers.length);
+    const headerSize = batch.headers.reduce((acc, header) => acc + header.name.length + header.value.length, 0);
+    const bodySize = batch.body.length;
+
+    return fixed + headerSize + bodySize;
+}
+
+function meteredBatch(batch: SequencedRecordBatch): number {
+    return batch.records.reduce((acc, record) => acc + meteredRecord(record), 0);
+}
+
 export function genS2RequestToken(): string {
     return v4().replace(/-/g, "");
 }

src/index.extras.ts

@@ -54,7 +54,7 @@ export function serverURLFromOptions(options: SDKOptions): URL | null {
 export const SDK_METADATA = {
   language: "typescript",
   openapiDocVersion: "1.0.0",
-  sdkVersion: "0.3.7",
+  sdkVersion: "0.3.8",
   genVersion: "2.506.0",
-  userAgent: "speakeasy-sdk/typescript 0.3.7 2.506.0 1.0.0 @s2-dev/streamstore",
+  userAgent: "speakeasy-sdk/typescript 0.3.8 2.506.0 1.0.0 @s2-dev/streamstore",
 } as const;