fix(socket-handler): Fix socket handler race condition by differentiating between async and sync handlers. Now, async socket handlers complete their setup before initial data is emitted, ensuring that no data is lost. Documentation and tests have been updated to reflect this change.

This commit is contained in:
2025-05-29 00:24:57 +00:00
parent d42fa8b1e9
commit e4aade4a9a
11 changed files with 1338 additions and 6 deletions

View File

@ -399,6 +399,15 @@ export class RouteConnectionHandler {
this.handleStaticAction(socket, record, route, initialChunk);
return;
case 'socket-handler':
logger.log('info', `Handling socket-handler action for route ${route.name}`, {
connectionId,
routeName: route.name,
component: 'route-handler'
});
this.handleSocketHandlerAction(socket, record, route, initialChunk);
return;
default:
logger.log('error', `Unknown action type '${(route.action as any).type}' for connection ${connectionId}`, {
connectionId,
@ -776,6 +785,75 @@ export class RouteConnectionHandler {
}, record, initialChunk);
}
/**
* Handle a socket-handler action for a route
*/
private async handleSocketHandlerAction(
socket: plugins.net.Socket,
record: IConnectionRecord,
route: IRouteConfig,
initialChunk?: Buffer
): Promise<void> {
const connectionId = record.id;
if (!route.action.socketHandler) {
logger.log('error', 'socket-handler action missing socketHandler function', {
connectionId,
routeName: route.name,
component: 'route-handler'
});
socket.destroy();
this.connectionManager.cleanupConnection(record, 'missing_handler');
return;
}
try {
// Call the handler
const result = route.action.socketHandler(socket);
// Handle async handlers properly
if (result instanceof Promise) {
result
.then(() => {
// Emit initial chunk after async handler completes
if (initialChunk && initialChunk.length > 0) {
socket.emit('data', initialChunk);
}
})
.catch(error => {
logger.log('error', 'Socket handler error', {
connectionId,
routeName: route.name,
error: error.message,
component: 'route-handler'
});
if (!socket.destroyed) {
socket.destroy();
}
this.connectionManager.cleanupConnection(record, 'handler_error');
});
} else {
// For sync handlers, emit on next tick
if (initialChunk && initialChunk.length > 0) {
process.nextTick(() => {
socket.emit('data', initialChunk);
});
}
}
} catch (error) {
logger.log('error', 'Socket handler error', {
connectionId,
routeName: route.name,
error: error.message,
component: 'route-handler'
});
if (!socket.destroyed) {
socket.destroy();
}
this.connectionManager.cleanupConnection(record, 'handler_error');
}
}
/**
* Setup improved error handling for the outgoing connection
*/