fix: handle unexpected end of stream (#46)

- we can't know if a specific async type is done or not as it's the
responsibility of the type itself
- we can know if a stream is interrupted prematurely, so I added a check
for that
- we treat the "stream interrupted" as a message that we enqueue - if
the interruption happens when the iterator has already consumed what it
needs, we're all good

src/async/asyncErrors.ts

@@ -0,0 +1,44 @@
+import { TsonError } from "../errors.js";
+import { isPlainObject } from "../internals/isPlainObject.js";
+
+function getErrorMessageFromUnknown(unknown: unknown): null | string {
+	if (typeof unknown === "string") {
+		return unknown;
+	}
+
+	if (unknown instanceof Error) {
+		return unknown.message;
+	}
+
+	if (isPlainObject(unknown) && typeof unknown["message"] === "string") {
+		return unknown["message"];
+	}
+
+	return null;
+}
+
+export class TsonPromiseRejectionError extends TsonError {
+	constructor(cause: unknown) {
+		// get error message from cause if possible
+		const message = getErrorMessageFromUnknown(cause) ?? "Promise rejected";
+		super(message, { cause });
+		this.name = "TsonPromiseRejectionError";
+	}
+
+	static from(cause: unknown) {
+		return cause instanceof Error
+			? cause
+			: new TsonPromiseRejectionError(cause);
+	}
+}
+
+export class TsonStreamInterruptedError extends TsonError {
+	constructor(cause: unknown) {
+		super(
+			"Stream interrupted: " +
+				(getErrorMessageFromUnknown(cause) ?? "unknown reason"),
+			{ cause },
+		);
+		this.name = "TsonStreamInterruptedError";
+	}
+}

src/async/asyncTypes.ts

@@ -1,7 +1,11 @@
 import { TsonError } from "../errors.js";
-import { TsonType } from "../sync/syncTypes.js";
-import { TsonBranded, TsonTypeTesterCustom } from "../sync/syncTypes.js";
-import { serialized } from "../sync/syncTypes.js";
+import {
+	TsonBranded,
+	TsonType,
+	TsonTypeTesterCustom,
+	serialized,
+} from "../sync/syncTypes.js";
+import { TsonStreamInterruptedError } from "./asyncErrors.js";
 
 export type TsonAsyncStringifierIterable<TValue> = AsyncIterable<string> & {
 	[serialized]: TValue;
@@ -33,7 +37,9 @@ export interface TsonTransformerSerializeDeserializeAsync<
 		/**
 		 * Reader for the ReadableStream of values
 		 */
-		reader: ReadableStreamDefaultReader<TSerializedValue>;
+		reader: ReadableStreamDefaultReader<
+			TSerializedValue | TsonStreamInterruptedError
+		>;
 	}) => TValue;
 
 	/**