From e7384fd87510b9ac64cd4059bc2650b88651110d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sun, 26 Apr 2026 06:13:52 +0000
Subject: [PATCH] feat: add streaming timeout protection

Add streamingTimeoutMs field to PluginSettings (default 120000ms). In
streamingFetch(), use Promise.race between the stream reader and a timeout
promise that aborts the AbortController and throws a clear timeout error.
External cancellation signals are forwarded to the same controller so only
one abort path is needed.

https://claude.ai/code/session_016QvEfqw6YZ3RjwBHrJ4w8S
---
 src/settings.ts  |  1 +
 src/streaming.ts | 64 ++++++++++++++++++++++++++++++++++++++----------
 src/types.ts     |  1 +
 3 files changed, 53 insertions(+), 13 deletions(-)

diff --git a/src/settings.ts b/src/settings.ts
index 88c8b15..70c69ff 100644
--- a/src/settings.ts
+++ b/src/settings.ts
@@ -41,6 +41,7 @@ export const DEFAULT_SETTINGS: PluginSettings = {
   exportFolder: 'Reading/Articles',
   cliTimeoutMs: 120000,
   streaming: true,
+  streamingTimeoutMs: 120000,
 };
 
 export const API_FORMATS: Record<string, ApiFormat> = {
diff --git a/src/streaming.ts b/src/streaming.ts
index 4d898d0..dccfaef 100644
--- a/src/streaming.ts
+++ b/src/streaming.ts
@@ -70,24 +70,15 @@ export interface StreamProgress {
   done: boolean;
 }
 
-/**
- * Perform a streaming fetch with SSE parsing.
- * Uses the native Fetch API (available in Electron/Obsidian).
- * Returns the full accumulated text when done.
- */
-export async function streamingFetch(
+async function doStreamingFetch(
   url: string,
   headers: Record<string, string>,
   body: unknown,
   extractDelta: DeltaExtractor,
-  onProgress?: (progress: StreamProgress) => void,
-  signal?: AbortSignal,
-  settings?: PluginSettings | null,
+  onProgress: ((progress: StreamProgress) => void) | undefined,
+  signal: AbortSignal,
+  settings: PluginSettings | null | undefined,
 ): Promise<string> {
-  if (typeof globalThis.fetch !== 'function') {
-    throw new Error('Streaming requires fetch API');
-  }
-
   const response = await globalThis.fetch(url, {
     method: 'POST',
     headers,
@@ -138,3 +129,50 @@ export async function streamingFetch(
   onProgress?.({ accumulated, done: true });
   return accumulated;
 }
+
+/**
+ * Perform a streaming fetch with SSE parsing and configurable timeout.
+ * Uses the native Fetch API (available in Electron/Obsidian).
+ * Returns the full accumulated text when done.
+ */
+export async function streamingFetch(
+  url: string,
+  headers: Record<string, string>,
+  body: unknown,
+  extractDelta: DeltaExtractor,
+  onProgress?: (progress: StreamProgress) => void,
+  signal?: AbortSignal,
+  settings?: PluginSettings | null,
+): Promise<string> {
+  if (typeof globalThis.fetch !== 'function') {
+    throw new Error('Streaming requires fetch API');
+  }
+
+  const timeoutMs = settings?.streamingTimeoutMs ?? 120000;
+  const timeoutController = new AbortController();
+
+  if (signal) {
+    if (signal.aborted) {
+      timeoutController.abort();
+    } else {
+      signal.addEventListener('abort', () => timeoutController.abort(), { once: true });
+    }
+  }
+
+  let timeoutId: ReturnType<typeof setTimeout> | null = null;
+  const timeoutPromise = new Promise<never>((_, reject) => {
+    timeoutId = setTimeout(() => {
+      timeoutController.abort();
+      reject(new Error(`Streaming timed out after ${timeoutMs}ms`));
+    }, timeoutMs);
+  });
+
+  try {
+    return await Promise.race([
+      doStreamingFetch(url, headers, body, extractDelta, onProgress, timeoutController.signal, settings),
+      timeoutPromise,
+    ]);
+  } finally {
+    if (timeoutId !== null) clearTimeout(timeoutId);
+  }
+}
diff --git a/src/types.ts b/src/types.ts
index 6b5770f..a519a0c 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -66,6 +66,7 @@ export interface PluginSettings {
   exportFolder: string;
   cliTimeoutMs: number;
   streaming: boolean;
+  streamingTimeoutMs: number;
 }
 
 /* ---------- Provider types ---------- */