andras/vault-link
andras/vault-link
1
0
Fork 0
Code Issues Pull requests 19 Projects Releases 55 Packages Wiki Activity Actions 26

Various improvements #169

Merged
schmelczer merged 78 commits from asch/saturday into main 2025-11-30 15:24:52 +00:00
Conversation 2 Commits 78 Files changed 112 +100 -66
1 changed files with 100 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit d2356f1e4d - Show all commits

Stop leaking promises in ws manager

Prev Next
Andras Schmelczer 2025-11-27 21:21:43 +00:00
commit d2356f1e4d

168
frontend/sync-client/src/services/websocket-manager.ts
View file

@ -7,6 +7,7 @@ import type { ClientCursors } from "./types/ClientCursors";
import { createPromise } from "../utils/create-promise"; import { createPromise } from "../utils/create-promise";
import type { WebSocketVaultUpdate } from "./types/WebSocketVaultUpdate"; import type { WebSocketVaultUpdate } from "./types/WebSocketVaultUpdate";
import { awaitAll } from "../utils/await-all"; import { awaitAll } from "../utils/await-all";
import { WEBSOCKET_DISCONNECT_TIMEOUT_IN_S } from "../consts";
export class WebSocketManager { export class WebSocketManager {
private readonly webSocketStatusChangeListeners: (( private readonly webSocketStatusChangeListeners: ((
@ -87,7 +88,6 @@ export class WebSocketManager {
this.isStopped = true; this.isStopped = true;
// Clear pending reconnect timeout
if (this.reconnectTimeoutId !== undefined) { if (this.reconnectTimeoutId !== undefined) {
clearTimeout(this.reconnectTimeoutId); clearTimeout(this.reconnectTimeoutId);
this.reconnectTimeoutId = undefined; this.reconnectTimeoutId = undefined;
@ -95,10 +95,40 @@ export class WebSocketManager {
this.webSocket?.close(1000, "WebSocketManager has been stopped"); this.webSocket?.close(1000, "WebSocketManager has been stopped");
while (this.isWebSocketConnected) { // eslint-disable-next-line @typescript-eslint/init-declarations
await promise; let timeoutId: ReturnType<typeof setTimeout> | undefined;
const timeoutPromise = new Promise<void>((_, reject) => {
timeoutId = setTimeout(() => {
reject(
new Error(
`Timeout waiting for WebSocket to close after ${WEBSOCKET_DISCONNECT_TIMEOUT_IN_S} seconds`
)
);
}, WEBSOCKET_DISCONNECT_TIMEOUT_IN_S * 1000);
});
try {
while (this.isWebSocketConnected) {
await Promise.race([promise, timeoutPromise]);
}
} catch (error) {
this.logger.error(
`Error while waiting for WebSocket to close: ${String(error)}`
);
// Force cleanup even if close didn't work
this.resolveDisconnectingPromise();
this.resolveDisconnectingPromise = null;
} finally {
// Clear timeout to prevent unhandled rejection
if (timeoutId !== undefined) {
clearTimeout(timeoutId);
}
} }
await this.waitUntilFinished();
}
public async waitUntilFinished(): Promise<void> {
await awaitAll(this.outstandingPromises); await awaitAll(this.outstandingPromises);
} }
@ -112,41 +142,57 @@ export class WebSocketManager {
); );
} }
webSocket.send(JSON.stringify(message)); try {
webSocket.send(JSON.stringify(message));
} catch (error) {
this.logger.error(
`Failed to send handshake message: ${String(error)}`
);
throw error;
}
} }
public updateLocalCursors(cursorPositions: CursorPositionFromClient): void { public updateLocalCursors(cursorPositions: CursorPositionFromClient): void {
if (!this.isWebSocketConnected) { if (!this.isWebSocketConnected || !this.webSocket) {
// A missing cursor update is fine, we can just skip it if needed // A missing cursor update is fine, we can just skip it if needed
this.logger.warn( this.logger.warn(
"WebSocket is not connected, cannot send cursor positions" "WebSocket is not connected, cannot send cursor positions"
); );
return; return;
} }
const message: WebSocketClientMessage = { const message: WebSocketClientMessage = {
type: "cursorPositions", type: "cursorPositions",
...cursorPositions ...cursorPositions
}; };
const { webSocket } = this;
if (!webSocket) { try {
this.logger.warn( this.webSocket.send(JSON.stringify(message));
"WebSocket is not connected, cannot send cursor positions" this.logger.debug(
`Sent cursor positions: ${JSON.stringify(cursorPositions)}`
);
} catch (error) {
this.logger.warn(
`Failed to send cursor positions: ${String(error)}`
); );
return;
} }
webSocket.send(JSON.stringify(message));
this.logger.debug(
`Sent cursor positions: ${JSON.stringify(cursorPositions)}`
);
} }
private initializeWebSocket(): void { private initializeWebSocket(): void {
try { // Clean up old WebSocket handlers to prevent race conditions
this.webSocket?.close(); if (this.webSocket) {
} catch (e) { try {
this.logger.error( // Remove handlers to prevent them from firing after new connection
`Failed to close previous WebSocket connection: ${e}` this.webSocket.onopen = null;
); this.webSocket.onclose = null;
this.webSocket.onmessage = null;
this.webSocket.onerror = null;
this.webSocket.close();
} catch (e) {
this.logger.error(
`Failed to close previous WebSocket connection: ${e}`
);
}
} }
const wsUri = new URL(this.settings.getSettings().remoteUri); const wsUri = new URL(this.settings.getSettings().remoteUri);
@ -171,13 +217,25 @@ export class WebSocketManager {
event.data event.data
) as WebSocketServerMessage; ) as WebSocketServerMessage;
void this.handleWebSocketMessage(message).catch( // Track the message handling promise
(error: unknown) => { const messageHandlingPromise = this.handleWebSocketMessage(
message
)
.catch((error: unknown) => {
this.logger.error( this.logger.error(
`Error handling WebSocket message: ${String(error)}` `Error handling WebSocket message: ${String(error)}`
); );
} })
); .finally(() => {
const index = this.outstandingPromises.indexOf(
messageHandlingPromise
);
if (index !== -1) {
void this.outstandingPromises.splice(index, 1); // ignore the returned promise
}
});
void this.outstandingPromises.push(messageHandlingPromise); // ignore the returned promise
} catch (error) { } catch (error) {
this.logger.error( this.logger.error(
`Error parsing WebSocket message: ${String(error)}` `Error parsing WebSocket message: ${String(error)}`
@ -186,7 +244,7 @@ export class WebSocketManager {
}; };
this.webSocket.onclose = (event): void => { this.webSocket.onclose = (event): void => {
this.logger.error( this.logger.warn(
`WebSocket closed with code ${event.code} (${event.reason == "" ? "unknown reason" : event.reason})` `WebSocket closed with code ${event.code} (${event.reason == "" ? "unknown reason" : event.reason})`
); );
this.webSocketStatusChangeListeners.forEach((listener) => this.webSocketStatusChangeListeners.forEach((listener) =>
@ -209,28 +267,16 @@ export class WebSocketManager {
message: WebSocketServerMessage message: WebSocketServerMessage
): Promise<void> { ): Promise<void> {
if (message.type === "vaultUpdate") { if (message.type === "vaultUpdate") {
const promises = this.remoteVaultUpdateListeners.map( await awaitAll(
async (listener) => { this.remoteVaultUpdateListeners.map(async (listener) => {
const trackedPromise = listener(message) await listener(message).catch((error: unknown) => {
.catch((error: unknown) => { this.logger.error(
this.logger.error( `Error in vault update listener: ${String(error)}`
`Error in vault update listener: ${String(error)}` );
); });
}) })
.finally(() => {
const index =
this.outstandingPromises.indexOf(
trackedPromise
);
if (index !== -1) {
// eslint-disable-next-line @typescript-eslint/no-floating-promises
this.outstandingPromises.splice(index, 1);
}
});
await trackedPromise;
}
); );
this.outstandingPromises.push(...promises);
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
} else if (message.type === "cursorPositions") { } else if (message.type === "cursorPositions") {
this.logger.debug( this.logger.debug(
@ -239,28 +285,16 @@ export class WebSocketManager {
const filteredClients = message.clients.filter( const filteredClients = message.clients.filter(
(client) => client.deviceId !== this.deviceId (client) => client.deviceId !== this.deviceId
); );
const promises = this.remoteCursorsUpdateListeners.map(
async (listener) => { await awaitAll(
const trackedPromise = listener(filteredClients) this.remoteCursorsUpdateListeners.map(async (listener) => {
.catch((error: unknown) => { await listener(filteredClients).catch((error: unknown) => {
this.logger.error( this.logger.error(
`Error in cursor positions listener: ${String(error)}` `Error in cursor positions listener: ${String(error)}`
); );
}) });
.finally(() => { })
const index =
this.outstandingPromises.indexOf(
trackedPromise
);
if (index !== -1) {
// eslint-disable-next-line @typescript-eslint/no-floating-promises
this.outstandingPromises.splice(index, 1);
}
});
await trackedPromise;
}
); );
this.outstandingPromises.push(...promises);
} else { } else {
this.logger.warn( this.logger.warn(
`Received unknown message type: ${JSON.stringify(message)}` `Received unknown message type: ${JSON.stringify(message)}`