fix: handle 416 explicitly in s2s read session (#99)

Author: infiniteregrets

packages/streamstore/src/error.ts

@@ -356,26 +356,35 @@ export class FencingTokenMismatchError extends S2Error {
  * This occurs when you specify a `startSeqNum` that is greater than the current tail
  * of the stream (HTTP 416 Range Not Satisfiable).
  *
+ * The `tail` property contains the current tail position of the stream.
+ *
  * To handle this gracefully, you can set `clamp: true` in your read options to
  * automatically start from the tail instead of throwing an error.
  */
 export class RangeNotSatisfiableError extends S2Error {
+	/** The current tail position of the stream. */
+	public readonly tail?: { seq_num: number; timestamp: number };
+
 	constructor({
-		message = "Range not satisfiable: requested position is beyond the stream tail. Use 'clamp: true' to start from the tail instead.",
 		code,
 		status = 416,
+		tail,
 	}: {
-		message?: string;
 		code?: string;
 		status?: number;
+		tail?: { seq_num: number; timestamp: number };
 	} = {}) {
+		const message = tail
+			? `Range not satisfiable: requested position is beyond the stream tail (seq_num=${tail.seq_num}). Use 'clamp: true' to start from the tail instead.`
+			: "Range not satisfiable: requested position is beyond the stream tail. Use 'clamp: true' to start from the tail instead.";
 		super({
 			message,
 			code,
 			status,
 			origin: "server",
 		});
 		this.name = "RangeNotSatisfiableError";
+		this.tail = tail;
 	}
 }
 

packages/streamstore/src/lib/stream/transport/fetch/shared.ts

@@ -53,7 +53,15 @@ export async function streamRead<Format extends "string" | "bytes" = "string">(
 	if (response.error) {
 		const status = response.response.status;
 		if (status === 416) {
-			throw new RangeNotSatisfiableError({ status });
+			const err = response.error as {
+				tail?: { seq_num: number; timestamp: number };
+				code?: string;
+			};
+			throw new RangeNotSatisfiableError({
+				status,
+				tail: err.tail,
+				code: err.code,
+			});
 		}
 		throw makeServerError(
 			{ status, statusText: response.response.statusText },

packages/streamstore/src/lib/stream/transport/s2s/index.ts

@@ -404,23 +404,40 @@ class S2SReadSession<Format extends "string" | "bytes" = "string">
 
 					stream.on("data", (chunk: Buffer) => {
 						try {
-							if ((responseCode ?? 500) >= 400) {
+							const status = responseCode ?? 500;
+							if (status >= 400) {
 								const errorText = textDecoder.decode(chunk);
+								debug("error response: status=%d body=%s", status, errorText);
+								if (status === 416) {
+									try {
+										const errorJson = JSON.parse(errorText);
+										safeError(
+											new RangeNotSatisfiableError({
+												status,
+												code: errorJson.code,
+												tail: errorJson.tail,
+											}),
+										);
+									} catch {
+										safeError(new RangeNotSatisfiableError({ status }));
+									}
+									return;
+								}
 								try {
 									const errorJson = JSON.parse(errorText);
 									safeError(
 										new S2Error({
 											message: errorJson.message ?? "Unknown error",
 											code: errorJson.code,
-											status: responseCode,
+											status,
 											origin: "server",
 										}),
 									);
 								} catch {
 									safeError(
 										new S2Error({
 											message: errorText || "Unknown error",
-											status: responseCode,
+											status,
 											origin: "server",
 										}),
 									);
@@ -441,7 +458,13 @@ class S2SReadSession<Format extends "string" | "bytes" = "string">
 
 											// Map known read errors
 											if (status === 416) {
-												safeError(new RangeNotSatisfiableError({ status }));
+												safeError(
+													new RangeNotSatisfiableError({
+														status,
+														code: errorJson.code,
+														tail: errorJson.tail,
+													}),
+												);
 											} else {
 												safeError(
 													makeServerError(