fix: implement loading timeout, SSE reconnect, and message retry (#857)
* fix: implement loading timeout, SSE reconnect, and message retry - Loading timeout: 30s timeout with retry/cancel buttons to prevent infinite loading - SSE reconnect: Auto-reconnect up to 3 times with exponential backoff (1s, 2s, 4s) - Message retry: Ensure critical messages reach webview with 5s timeout and 3 retries This fix prevents the chat from getting stuck in a permanent loading state and improves reliability of SSE connections and webview communication. Fixes #851 * fix: harden vscode bridge retry flow --------- Co-authored-by: Bohdan Triapitsyn <artmore@protonmail.com>
This commit is contained in:
committed by
GitHub
co-authored by
Bohdan Triapitsyn
parent
1fed81439c
commit
4cb918f6cb
@@ -23,6 +23,24 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
private _sseStreams = new Map<string, AbortController>();
|
private _sseStreams = new Map<string, AbortController>();
|
||||||
private readonly _webviewDevServerUrl: string | null;
|
private readonly _webviewDevServerUrl: string | null;
|
||||||
|
|
||||||
|
// Message delivery confirmation and retry
|
||||||
|
private readonly _pendingMessages = new Set<string>();
|
||||||
|
private readonly _messageTimeouts = new Map<string, NodeJS.Timeout>();
|
||||||
|
private readonly _MESSAGE_TIMEOUT = 5000; // 5 seconds
|
||||||
|
private readonly _MAX_RETRIES = 3;
|
||||||
|
|
||||||
|
private _createMessageId(): string {
|
||||||
|
return `msg_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
|
||||||
|
}
|
||||||
|
|
||||||
|
private _clearPendingMessages(): void {
|
||||||
|
for (const timeout of this._messageTimeouts.values()) {
|
||||||
|
clearTimeout(timeout);
|
||||||
|
}
|
||||||
|
this._messageTimeouts.clear();
|
||||||
|
this._pendingMessages.clear();
|
||||||
|
}
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
private readonly _context: vscode.ExtensionContext,
|
private readonly _context: vscode.ExtensionContext,
|
||||||
private readonly _extensionUri: vscode.Uri,
|
private readonly _extensionUri: vscode.Uri,
|
||||||
@@ -34,6 +52,7 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
public resolveWebviewView(
|
public resolveWebviewView(
|
||||||
webviewView: vscode.WebviewView
|
webviewView: vscode.WebviewView
|
||||||
) {
|
) {
|
||||||
|
this._clearPendingMessages();
|
||||||
this._view = webviewView;
|
this._view = webviewView;
|
||||||
|
|
||||||
const distUri = vscode.Uri.joinPath(this._extensionUri, 'dist');
|
const distUri = vscode.Uri.joinPath(this._extensionUri, 'dist');
|
||||||
@@ -46,11 +65,24 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
webviewView.webview.html = this._getHtmlForWebview(webviewView.webview);
|
webviewView.webview.html = this._getHtmlForWebview(webviewView.webview);
|
||||||
// Send theme payload (including optional Shiki theme JSON) after the webview is set up.
|
// Send theme payload (including optional Shiki theme JSON) after the webview is set up.
|
||||||
void this.updateTheme(vscode.window.activeColorTheme.kind);
|
void this.updateTheme(vscode.window.activeColorTheme.kind);
|
||||||
|
|
||||||
// Send cached connection status and API URL (may have been set before webview was resolved)
|
// Send cached connection status and API URL (may have been set before webview was resolved)
|
||||||
this._sendCachedState();
|
this._sendCachedState();
|
||||||
|
|
||||||
webviewView.webview.onDidReceiveMessage(async (message: BridgeRequest) => {
|
webviewView.onDidDispose(() => {
|
||||||
|
this._clearPendingMessages();
|
||||||
|
});
|
||||||
|
|
||||||
|
webviewView.webview.onDidReceiveMessage(async (message: (BridgeRequest & { _msgId?: string }) | { type: 'bridge:ack'; _msgId: string }) => {
|
||||||
|
if (message.type === 'bridge:ack' && typeof message._msgId === 'string') {
|
||||||
|
this._confirmMessage(message._msgId);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!('id' in message) || typeof message.id !== 'string') {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
if (message.type === 'restartApi') {
|
if (message.type === 'restartApi') {
|
||||||
await this._openCodeManager?.restart();
|
await this._openCodeManager?.restart();
|
||||||
return;
|
return;
|
||||||
@@ -58,13 +90,13 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
|
|
||||||
if (message.type === 'api:sse:start') {
|
if (message.type === 'api:sse:start') {
|
||||||
const response = await this._startSseProxy(message);
|
const response = await this._startSseProxy(message);
|
||||||
webviewView.webview.postMessage(response);
|
void this._sendMessageWithRetry(response);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (message.type === 'api:sse:stop') {
|
if (message.type === 'api:sse:stop') {
|
||||||
const response = await this._stopSseProxy(message);
|
const response = await this._stopSseProxy(message);
|
||||||
webviewView.webview.postMessage(response);
|
void this._sendMessageWithRetry(response);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -72,7 +104,7 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
manager: this._openCodeManager,
|
manager: this._openCodeManager,
|
||||||
context: this._context,
|
context: this._context,
|
||||||
});
|
});
|
||||||
webviewView.webview.postMessage(response);
|
void this._sendMessageWithRetry(response);
|
||||||
|
|
||||||
if (message.type === 'api:config/settings:save' && response.success) {
|
if (message.type === 'api:config/settings:save' && response.success) {
|
||||||
void vscode.commands.executeCommand('openchamber.internal.settingsSynced', response.data);
|
void vscode.commands.executeCommand('openchamber.internal.settingsSynced', response.data);
|
||||||
@@ -190,11 +222,80 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Message delivery confirmation
|
||||||
|
private _confirmMessage(messageId: string) {
|
||||||
|
this._pendingMessages.delete(messageId);
|
||||||
|
|
||||||
|
const timeout = this._messageTimeouts.get(messageId);
|
||||||
|
if (timeout) {
|
||||||
|
clearTimeout(timeout);
|
||||||
|
this._messageTimeouts.delete(messageId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send message with retry mechanism
|
||||||
|
private async _sendMessageWithRetry(response: BridgeResponse, retryCount: number = 0, messageId?: string): Promise<boolean> {
|
||||||
|
if (!this._view) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
const pendingMessageId = messageId ?? this._createMessageId();
|
||||||
|
const existingTimeout = this._messageTimeouts.get(pendingMessageId);
|
||||||
|
if (existingTimeout) {
|
||||||
|
clearTimeout(existingTimeout);
|
||||||
|
this._messageTimeouts.delete(pendingMessageId);
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
const delivered = await this._view.webview.postMessage({
|
||||||
|
...response,
|
||||||
|
_msgId: pendingMessageId,
|
||||||
|
});
|
||||||
|
if (!delivered) {
|
||||||
|
throw new Error('Webview rejected message delivery');
|
||||||
|
}
|
||||||
|
|
||||||
|
this._pendingMessages.add(pendingMessageId);
|
||||||
|
|
||||||
|
const timeout = setTimeout(() => {
|
||||||
|
if (!this._pendingMessages.has(pendingMessageId)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (retryCount < this._MAX_RETRIES) {
|
||||||
|
console.warn(`[Message Retry] Message ${pendingMessageId} not confirmed, retrying (${retryCount + 1}/${this._MAX_RETRIES})...`);
|
||||||
|
void this._sendMessageWithRetry(response, retryCount + 1, pendingMessageId);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
console.error(`[Message Retry] Message ${pendingMessageId} failed after ${this._MAX_RETRIES} retries`);
|
||||||
|
this._pendingMessages.delete(pendingMessageId);
|
||||||
|
this._messageTimeouts.delete(pendingMessageId);
|
||||||
|
}, this._MESSAGE_TIMEOUT);
|
||||||
|
|
||||||
|
this._messageTimeouts.set(pendingMessageId, timeout);
|
||||||
|
return true;
|
||||||
|
|
||||||
|
} catch (error) {
|
||||||
|
console.error(`[Message Retry] Failed to send message:`, error);
|
||||||
|
|
||||||
|
if (retryCount < this._MAX_RETRIES) {
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 100 * (retryCount + 1)));
|
||||||
|
return this._sendMessageWithRetry(response, retryCount + 1, pendingMessageId);
|
||||||
|
}
|
||||||
|
|
||||||
|
this._pendingMessages.delete(pendingMessageId);
|
||||||
|
this._messageTimeouts.delete(pendingMessageId);
|
||||||
|
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private _sendCachedState() {
|
private _sendCachedState() {
|
||||||
if (!this._view) {
|
if (!this._view) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
this._view.webview.postMessage({
|
this._view.webview.postMessage({
|
||||||
type: 'connectionStatus',
|
type: 'connectionStatus',
|
||||||
status: this._cachedStatus,
|
status: this._cachedStatus,
|
||||||
|
|||||||
@@ -27,6 +27,10 @@ const SSE_RESPONSE_HEADERS = {
|
|||||||
'cache-control': 'no-cache',
|
'cache-control': 'no-cache',
|
||||||
} as const;
|
} as const;
|
||||||
|
|
||||||
|
// SSE reconnect configuration
|
||||||
|
const MAX_RECONNECTS = 3;
|
||||||
|
const BASE_RECONNECT_DELAY = 1000; // 1 second
|
||||||
|
|
||||||
const serializeSseEventBlock = (event: StreamEvent<unknown>): string => {
|
const serializeSseEventBlock = (event: StreamEvent<unknown>): string => {
|
||||||
const lines: string[] = [];
|
const lines: string[] = [];
|
||||||
if (typeof event.id === 'string' && event.id.length > 0) {
|
if (typeof event.id === 'string' && event.id.length > 0) {
|
||||||
@@ -103,28 +107,65 @@ export const openSseProxy = async ({
|
|||||||
const { pathname, directory } = normalizeSsePath(path);
|
const { pathname, directory } = normalizeSsePath(path);
|
||||||
const resolvedDirectory = directory || resolveDefaultDirectory(manager);
|
const resolvedDirectory = directory || resolveDefaultDirectory(manager);
|
||||||
|
|
||||||
const connect = async () => {
|
// Reconnect logic with exponential backoff
|
||||||
if (pathname === '/global/event') {
|
let reconnectAttempts = 0;
|
||||||
try {
|
|
||||||
return await client.global.event(getSseOptions(signal, onChunk));
|
|
||||||
} catch (error) {
|
|
||||||
if ((error as Error)?.name === 'AbortError' || signal.aborted) {
|
|
||||||
throw error;
|
|
||||||
}
|
|
||||||
return client.event.subscribe(
|
|
||||||
{ directory: resolvedDirectory },
|
|
||||||
getSseOptions(signal, onChunk, resolvedDirectory),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return client.event.subscribe(
|
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
|
||||||
{ directory: resolvedDirectory },
|
|
||||||
getSseOptions(signal, onChunk),
|
const connect = async (): Promise<{ stream: AsyncIterable<unknown> }> => {
|
||||||
);
|
try {
|
||||||
|
console.log(`[SSE] Connecting to ${pathname} (attempt ${reconnectAttempts + 1}/${MAX_RECONNECTS + 1})`);
|
||||||
|
|
||||||
|
if (pathname === '/global/event') {
|
||||||
|
try {
|
||||||
|
const result = await client.global.event(getSseOptions(signal, onChunk));
|
||||||
|
// Reset reconnect counter on successful connection
|
||||||
|
reconnectAttempts = 0;
|
||||||
|
return result;
|
||||||
|
} catch (error) {
|
||||||
|
if ((error as Error)?.name === 'AbortError' || signal.aborted) {
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
// Fallback to directory event on error
|
||||||
|
console.warn('[SSE] Global event failed, falling back to directory event', error);
|
||||||
|
const result = client.event.subscribe(
|
||||||
|
{ directory: resolvedDirectory },
|
||||||
|
getSseOptions(signal, onChunk, resolvedDirectory),
|
||||||
|
);
|
||||||
|
reconnectAttempts = 0;
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const result = client.event.subscribe(
|
||||||
|
{ directory: resolvedDirectory },
|
||||||
|
getSseOptions(signal, onChunk),
|
||||||
|
);
|
||||||
|
reconnectAttempts = 0;
|
||||||
|
return result;
|
||||||
|
} catch (error) {
|
||||||
|
// Implement reconnect logic
|
||||||
|
if (!signal.aborted && reconnectAttempts < MAX_RECONNECTS) {
|
||||||
|
reconnectAttempts++;
|
||||||
|
const delay = BASE_RECONNECT_DELAY * Math.pow(2, reconnectAttempts - 1); // Exponential backoff
|
||||||
|
|
||||||
|
console.warn(
|
||||||
|
`[SSE] Connection failed (attempt ${reconnectAttempts}/${MAX_RECONNECTS}), ` +
|
||||||
|
`retrying in ${delay}ms...`,
|
||||||
|
error
|
||||||
|
);
|
||||||
|
|
||||||
|
await sleep(delay);
|
||||||
|
return connect(); // Recursive retry
|
||||||
|
}
|
||||||
|
|
||||||
|
console.error(`[SSE] Connection failed after ${reconnectAttempts} attempts`, error);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
const result = await connect();
|
const result = await connect();
|
||||||
|
|
||||||
const run = (async () => {
|
const run = (async () => {
|
||||||
try {
|
try {
|
||||||
for await (const _ of result.stream) {
|
for await (const _ of result.stream) {
|
||||||
@@ -135,7 +176,32 @@ export const openSseProxy = async ({
|
|||||||
}
|
}
|
||||||
} catch (error: unknown) {
|
} catch (error: unknown) {
|
||||||
const cause = (error as { cause?: { code?: string } } | null)?.cause;
|
const cause = (error as { cause?: { code?: string } } | null)?.cause;
|
||||||
if (!signal.aborted && cause?.code !== 'UND_ERR_SOCKET') {
|
|
||||||
|
// Attempt reconnect on socket errors
|
||||||
|
if (!signal.aborted) {
|
||||||
|
if (cause?.code === 'UND_ERR_SOCKET' || cause?.code === 'ECONNRESET') {
|
||||||
|
console.warn('[SSE] Socket error detected, attempting reconnect...');
|
||||||
|
|
||||||
|
if (reconnectAttempts < MAX_RECONNECTS) {
|
||||||
|
reconnectAttempts++;
|
||||||
|
const delay = BASE_RECONNECT_DELAY * Math.pow(2, reconnectAttempts - 1);
|
||||||
|
await sleep(delay);
|
||||||
|
|
||||||
|
// Attempt to reconnect
|
||||||
|
try {
|
||||||
|
const newResult = await connect();
|
||||||
|
for await (const _ of newResult.stream) {
|
||||||
|
void _;
|
||||||
|
if (signal.aborted) break;
|
||||||
|
}
|
||||||
|
return; // Successfully reconnected
|
||||||
|
} catch (reconnectError) {
|
||||||
|
console.error('[SSE] Reconnect failed', reconnectError);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Re-throw if we couldn't recover
|
||||||
throw error;
|
throw error;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -47,6 +47,11 @@ window.addEventListener('message', (event: MessageEvent<BridgeResponse>) => {
|
|||||||
const response = event.data;
|
const response = event.data;
|
||||||
if (!response || typeof response.id !== 'string') return;
|
if (!response || typeof response.id !== 'string') return;
|
||||||
|
|
||||||
|
const messageId = (response as BridgeResponse & { _msgId?: unknown })._msgId;
|
||||||
|
if (typeof messageId === 'string' && messageId.length > 0) {
|
||||||
|
getVSCodeAPI().postMessage({ type: 'bridge:ack', _msgId: messageId });
|
||||||
|
}
|
||||||
|
|
||||||
const pending = pendingRequests.get(response.id);
|
const pending = pendingRequests.get(response.id);
|
||||||
if (pending) {
|
if (pending) {
|
||||||
pendingRequests.delete(response.id);
|
pendingRequests.delete(response.id);
|
||||||
|
|||||||
Reference in New Issue
Block a user