{"record":{"id":"ec13f492ab6b482f","repo":"mastra-ai/mastra","slug":"unixsocketpubsub-is-closed","errorCode":null,"errorMessage":"UnixSocketPubSub is closed","messagePattern":"UnixSocketPubSub is closed","errorType":"exception","errorClass":"Error","httpStatus":null,"severity":"error","filePath":"packages/core/src/events/unix-socket-pubsub.ts","lineNumber":285,"sourceCode":"\n    for (const client of [...this.#brokerClients.values()]) {\n      this.#removeBrokerClient(client);\n    }\n\n    if (this.#server) {\n      await new Promise<void>(resolve => this.#server?.close(() => resolve()));\n      this.#server = undefined;\n    }\n\n    if (this.#isBroker) {\n      await unlink(this.socketPath).catch(() => {});\n    }\n    this.#isBroker = false;\n  }\n\n  async #ensureStarted(forceReconnect = false): Promise<void> {\n    if (this.#closed) {\n      throw new Error('UnixSocketPubSub is closed');\n    }\n    if (!forceReconnect && (this.#isBroker || (this.#clientSocket && !this.#clientSocket.destroyed))) {\n      return;\n    }\n    if (this.#starting) {\n      return this.#starting;\n    }\n\n    this.#starting = this.#start(forceReconnect).finally(() => {\n      this.#starting = undefined;\n    });\n    return this.#starting;\n  }\n\n  async #start(forceReconnect: boolean): Promise<void> {\n    if (forceReconnect) {\n      this.#clientSocket?.destroy();\n      this.#clientSocket = undefined;","sourceCodeStart":267,"sourceCodeEnd":303,"githubUrl":"https://github.com/mastra-ai/mastra/blob/75dd419e613fe9c39f846ffc500716141b74fda6/packages/core/src/events/unix-socket-pubsub.ts#L267-L303","documentation":"#ensureStarted is the gate for publish, subscribe, and the internal send/start paths. Once the instance has been closed (this.#closed set by close()), any attempt to start or use it throws this error — a closed UnixSocketPubSub cannot be restarted.","triggerScenarios":"Calling publish() or subscribe() after close() has resolved; a pending #recoverClientConnectionLoop or #sendToBroker tick firing after close(); #ensureStarted(forceReconnect) invoked when this.#closed is true.","commonSituations":"Shutdown ordering bug where close() runs while producers still emit, a long-lived publish timer not cancelled before close, reconnect loop racing application shutdown, using a stale reference to an already-closed instance.","solutions":["Reorder shutdown: cancel all publishers/subscribers and timers before calling close()","Track instance lifecycle and create a new UnixSocketPubSub instance if it must be used again after close","Guard publish/subscribe calls with an isClosed check in application code","In catch blocks for this error, stop retrying and fall back to local handling"],"exampleFix":"// before\nawait pubsub.close();\nawait pubsub.publish(\"events\", payload); // throws\n// after\nawait pubsub.publish(\"events\", payload);\nawait pubsub.close();","handlingStrategy":"try-catch","validationCode":"if (pubsub.isClosed?.()) {\n  throw new Error('Skipping publish: pubsub already closed');\n}\nawait pubsub.publish(topic, event);","typeGuard":"function isUsable(ps: UnixSocketPubSub): boolean {\n  return !ps.isClosed?.();\n}","tryCatchPattern":"try {\n  await pubsub.publish(topic, event);\n} catch (e) {\n  if (e.message === 'UnixSocketPubSub is closed') {\n    logger.warn('Publish after close; dropping event', { topic });\n  } else throw e;\n}","preventionTips":["Cancel publish timers/producers before calling close()","Await all in-flight publish/subscribe promises before close","Wrap lifecycle in a manager that forbids operations after close","After close, always create a new instance rather than reusing the old one"],"tags":["pubsub","lifecycle","shutdown","unix-socket"],"backgroundTag":"pubsub-closed","analyzedSha":"75dd419e613fe9c39f846ffc500716141b74fda6","analyzedAt":"2026-08-30T00:15:31.844Z","schemaVersion":2},"datasetVersion":"2026-08-30T03:17:51.788Z"}