{"record":{"id":"313dcbeff789c539","repo":"mastra-ai/mastra","slug":"unixsocketpubsub-is-not-connected-to-a-broker","errorCode":null,"errorMessage":"UnixSocketPubSub is not connected to a broker","messagePattern":"UnixSocketPubSub is not connected to a broker","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"packages/core/src/events/unix-socket-pubsub.ts","lineNumber":760,"sourceCode":"        await new Promise(resolve => setTimeout(resolve, 10 * (attempt + 1)));\n      }\n    }\n  }\n\n  async #sendToActiveBroker(frame: ClientFrame) {\n    const socket = this.#clientSocket;\n    if (!socket || socket.destroyed) {\n      await this.#ensureStarted(true);\n    }\n    if (this.#isBroker) {\n      await this.#handlePromotedBrokerFrame(frame);\n      return;\n    }\n    const activeSocket = this.#clientSocket;\n    if (!activeSocket || activeSocket.destroyed) {\n      // NOTE: keep this exact message in sync with the transient-error\n      // classifier in #sendToBroker (search for 'not connected to a broker').\n      throw new Error('UnixSocketPubSub is not connected to a broker');\n    }\n    await writeFrame(activeSocket, frame);\n  }\n\n  async #handlePromotedBrokerFrame(frame: ClientFrame) {\n    if (frame.type === 'subscribe') {\n      this.#settleSubscribeWaiters(frame.topic);\n    } else if (frame.type === 'publish') {\n      await this.#publishFromBroker(frame.topic, frame.event);\n    }\n  }\n}\n","sourceCodeStart":742,"sourceCodeEnd":773,"githubUrl":"https://github.com/mastra-ai/mastra/blob/75dd419e613fe9c39f846ffc500716141b74fda6/packages/core/src/events/unix-socket-pubsub.ts#L742-L773","documentation":"UnixSocketPubSub publishes/subscribes over a local Unix domain socket broker. publish() (and similar send paths) checks the internal client socket before writing; if the socket was never connected or has been destroyed, the library throws this error rather than silently dropping the message. It is an internal invariant guard that doubles as a signal for the transient-error classifier in #sendToBroker.","triggerScenarios":"Calling publish/subscribe on a UnixSocketPubSub instance whose #clientSocket is undefined or has .destroyed set — e.g. calling publish() before connect() completes, after the broker process exited and killed the socket, or after an explicit disconnect().","commonSituations":"Publishing during application startup before the pub/sub client has finished connecting; a crashed broker daemon leaving a destroyed socket; unit tests instantiating UnixSocketPubSub without a running broker; race between broker restart and in-flight publish calls.","solutions":["Ensure connect() is awaited and completed before calling publish or subscribe","Add reconnect logic: catch this error, call connect() again, then retry the publish","Check broker/socket lifecycle — make sure the broker process is running and the socket path is valid","If publishing on shutdown, guard calls with a connected-state check on the instance"],"exampleFix":"// before\nawait pubsub.publish({ topic: 'events', data });\n// after\nif (!pubsub.isConnected()) {\n  await pubsub.connect();\n}\nawait pubsub.publish({ topic: 'events', data });","handlingStrategy":"retry","validationCode":"function canPublish(pubsub) {\n  return pubsub.isConnected?.() === true;\n}","typeGuard":"function isSocketLive(socket) {\n  return socket != null && !socket.destroyed;\n}","tryCatchPattern":"try {\n  await pubsub.publish(frame);\n} catch (e) {\n  if (e.message.includes('not connected to a broker')) {\n    await pubsub.connect();\n    await pubsub.publish(frame);\n  } else {\n    throw e;\n  }\n}","preventionTips":["Always await connect() before first publish","Add reconnect-with-backoff on socket close/destroy events","Suppress publishes during shutdown once disconnect() has been called","In tests, start the broker fixture before instantiating the pub/sub client"],"tags":["pubsub","ipc","connection","lifecycle"],"backgroundTag":"pubsub-not-connected","analyzedSha":"75dd419e613fe9c39f846ffc500716141b74fda6","analyzedAt":"2026-08-30T00:15:31.844Z","schemaVersion":2},"datasetVersion":"2026-08-30T03:17:51.788Z"}