🐛 Fail fast on Redis task dispatch when no instance is connected #10958

In multi-user mode, plugin task requests are published to a Redis channel
keyed by user token. When no MCP server instance held a plugin connection
for that token (e.g. after the user navigated away from the workspace),
the publish reached zero subscribers and the request was silently dropped,
so every tool call stalled until the 30-second task timeout instead of
failing with a meaningful error.

RedisBridge.sendTaskRequest now returns the PUBLISH receiver count and
releases its response-channel subscription when the request reached no
receiver (or publishing failed), since no response can arrive. PluginBridge
uses the count to reject the pending task immediately with the multi-user
connection error message; publish failures likewise reject the task instead
of surfacing as an unhandled rejection followed by a timeout. The pending-
task settlement logic shared with the timeout handler is extracted into a
rejectPendingTask helper.

AI-assisted-by: claude-fable-5

mcp/packages/server/src/PluginBridge.ts

@@ -20,6 +20,8 @@ interface ClientConnection {
  * over these connections.
  */
 export class PluginBridge {
+    public static readonly MULTIUSER_CONNECTION_ERROR_MESSAGE = `No Penpot instance connected for user token. Please ensure that Penpot is connected and that the MCP client connection is using the correct token.`;
+
     private readonly logger = createLogger("PluginBridge");
     private readonly wsServer: WebSocketServer;
     private readonly connectedClients: Map<WebSocket, ClientConnection> = new Map();
@@ -189,6 +191,35 @@ export class PluginBridge {
         this.logger.info(`Task ${response.id} completed: success=${response.success}`);
     }
 
+    /**
+     * Rejects a still-pending task with the given error, releasing its correlation state.
+     *
+     * Clears the task's timeout (if armed) and removes the task from the pending-task
+     * index before rejecting its promise. Safe to call for a task that has already been
+     * settled (e.g. by a response or a timeout), in which case nothing happens.
+     *
+     * @param taskId - The ID of the task to reject
+     * @param error - The error with which to reject the task
+     * @returns Whether the task was still pending and has been rejected
+     */
+    private rejectPendingTask(taskId: string, error: Error): boolean {
+        const pendingTask = this.pendingTasks.get(taskId);
+        if (!pendingTask) {
+            return false;
+        }
+
+        const timeoutHandle = this.taskTimeouts.get(taskId);
+        if (timeoutHandle) {
+            clearTimeout(timeoutHandle);
+            this.taskTimeouts.delete(taskId);
+        }
+        this.pendingTasks.delete(taskId);
+
+        pendingTask.rejectWithError(error);
+        this.logger.info(`Task ${taskId} rejected: ${error.message}`);
+        return true;
+    }
+
     /**
      * Determines the client connection to use for executing a task.
      *
@@ -207,9 +238,7 @@ export class PluginBridge {
 
             const connection = this.clientsByToken.get(sessionContext.userToken);
             if (!connection) {
-                throw new Error(
-                    `No plugin instance connected for user token. Please ensure the plugin is running and connected with the correct token.`
-                );
+                throw new Error(PluginBridge.MULTIUSER_CONNECTION_ERROR_MESSAGE);
             }
 
             return connection;
@@ -257,6 +286,10 @@ export class PluginBridge {
      * `resolveWithResult`/`rejectWithError` methods. The same correlation and timeout
      * handling therefore applies regardless of the transport.
      *
+     * When routing via Redis, the task is rejected immediately (rather than timing out)
+     * if the published request reached no instance, i.e. if no instance holds a plugin
+     * connection for the session's user token, or if publishing fails outright.
+     *
      * @param task - The task to dispatch
      * @param useRedis - Whether to route the request via Redis (multi-instance) rather
      *   than directly over the local WebSocket connection
@@ -279,9 +312,17 @@ export class PluginBridge {
 
             // register the task for result correlation, then publish the request via Redis
             this.pendingTasks.set(task.id, task);
-            void redisBridge.sendTaskRequest(userToken, task.toRequest(), (response) =>
-                this.handlePluginTaskResponse(response)
-            );
+            void redisBridge
+                .sendTaskRequest(userToken, task.toRequest(), (response) => this.handlePluginTaskResponse(response))
+                .then((receiverCount) => {
+                    // fail fast when no instance received the request (no connection with matching user token in any instance)
+                    if (receiverCount === 0) {
+                        this.rejectPendingTask(task.id, new Error(PluginBridge.MULTIUSER_CONNECTION_ERROR_MESSAGE));
+                    }
+                })
+                .catch((error) => {
+                    this.rejectPendingTask(task.id, error instanceof Error ? error : new Error(String(error)));
+                });
 
             // on timeout, release the response-channel subscription, since no response
             // will arrive to trigger its self-unsubscribe.
@@ -300,14 +341,13 @@ export class PluginBridge {
 
         // Set up a timeout to reject the task if no response is received
         const timeoutHandle = setTimeout(() => {
-            const pendingTask = this.pendingTasks.get(task.id);
-            if (pendingTask) {
-                this.pendingTasks.delete(task.id);
-                this.taskTimeouts.delete(task.id);
-                onTimeout?.();
-                pendingTask.rejectWithError(
+            if (
+                this.rejectPendingTask(
+                    task.id,
                     new Error(`Task ${task.id} timed out after ${this.taskTimeoutSecs} seconds`)
-                );
+                )
+            ) {
+                onTimeout?.();
             }
         }, this.taskTimeoutSecs * 1000);
 

mcp/packages/server/src/RedisBridge.ts

@@ -91,12 +91,16 @@ export class RedisBridge {
      * @param userToken - The user token identifying the target plugin's request channel
      * @param request - The serialized plugin task request, passed through verbatim
      * @param onResponse - Handler invoked with the response when it arrives
+     * @returns The number of instances that received the request. A count of 0 means no
+     *   instance is subscribed to the token's request channel (i.e. the plugin is not
+     *   connected anywhere); the request was dropped, no response will ever arrive, and
+     *   the response subscription has already been released.
      */
     async sendTaskRequest(
         userToken: string,
         request: PluginTaskRequest,
         onResponse: TaskResponseHandler
-    ): Promise<void> {
+    ): Promise<number> {
         const responseChannel = this.responseChannel(request.id);
         const requestChannel = this.requestChannel(userToken);
 
@@ -113,7 +117,19 @@ export class RedisBridge {
 
         await this.subscriber.subscribe(responseChannel);
         // publish only once the response subscription is confirmed
-        await this.publisher.publish(requestChannel, JSON.stringify(request));
+        let receiverCount: number;
+        try {
+            receiverCount = await this.publisher.publish(requestChannel, JSON.stringify(request));
+        } catch (error) {
+            // the request was never delivered, so no response can arrive
+            await this.unsubscribeFromResponse(request.id);
+            throw error;
+        }
+        if (receiverCount === 0) {
+            // no subscriber received the request, so no response can arrive
+            await this.unsubscribeFromResponse(request.id);
+        }
+        return receiverCount;
     }
 
     /**