{"record":{"id":"6dd4c37817f40b34","repo":"mastra-ai/mastra","slug":"signalspubsub-is-closed","errorCode":null,"errorMessage":"SignalsPubSub is closed","messagePattern":"SignalsPubSub is closed","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"mastracode/sdk/src/utils/signals-pubsub.ts","lineNumber":75,"sourceCode":"  }\n\n  async flush(): Promise<void> {\n    await Promise.all([...this.#sockets.values()].map(s => s.flush()));\n  }\n\n  async close(): Promise<void> {\n    this.#closed = true;\n    await Promise.allSettled([...this.#sockets.values()].map(s => s.close()));\n    this.#sockets.clear();\n  }\n\n  /** Get the underlying socket for a topic (for testing/inspection). */\n  getSocket(topic: string): UnixSocketPubSub | undefined {\n    return this.#sockets.get(this.#topicKey(topic));\n  }\n\n  async #getOrCreate(topic: string): Promise<UnixSocketPubSub> {\n    if (this.#closed) throw new Error('SignalsPubSub is closed');\n    const key = this.#topicKey(topic);\n    const existing = this.#sockets.get(key);\n    if (existing) return existing;\n    // Deduplicate concurrent callers so only one socket is created per topic.\n    let inflight = this.#pending.get(key);\n    if (!inflight) {\n      inflight = this.#initSocket(topic, key);\n      this.#pending.set(key, inflight);\n    }\n    const socket = await inflight;\n    if (this.#closed) throw new Error('SignalsPubSub is closed');\n    return socket;\n  }\n\n  async #initSocket(topic: string, key: string): Promise<UnixSocketPubSub> {\n    try {\n      const socketPath = await this.#socketPath(topic);\n      if (this.#closed) throw new Error('SignalsPubSub is closed');","sourceCodeStart":57,"sourceCodeEnd":93,"githubUrl":"https://github.com/mastra-ai/mastra/blob/75dd419e613fe9c39f846ffc500716141b74fda6/mastracode/sdk/src/utils/signals-pubsub.ts#L57-L93","documentation":"SignalsPubSub coordinates cross-process signals over one Unix socket per topic. Once close() has been called, the instance is permanently closed and every publish/subscribe entry point first checks the #closed flag and throws 'SignalsPubSub is closed' to prevent use-after-close. This synchronous check in #getOrCreate catches calls made after close() has fully completed (or at least after the flag was set).","triggerScenarios":"Calling publish() or subscribe() on a SignalsPubSub instance after close() has been invoked on it — e.g. publishing a signal event or subscribing to a thread-stream topic after shutdown.","commonSituations":"Publishing an event during process shutdown after a cleanup handler already closed the pubsub; holding a cached SignalsPubSub reference in a singleton that another module closed; race between a stream teardown calling close() and a late signal publish; tests that close fixtures but still emit on them.","solutions":["Do not call publish/subscribe after close(); guard your shutdown ordering so event emission finishes before teardown calls close().","Check a lifecycle flag in your app (e.g. this shuttingDown) before publishing, or wrap the call in try/catch for this error and drop the event.","If the instance was closed unexpectedly, create a fresh one via createSignalsPubSub(resourceId) and re-subscribe your handlers.","Audit for shared/singleton instances: ensure only one owner calls close() and others are notified."],"exampleFix":"// before\nawait signals.publish(topic, event); // may throw after close()\n// after\nif (isShuttingDown) return; // skip emit during teardown\ntry {\n  await signals.publish(topic, event);\n} catch (err) {\n  if (err instanceof Error && err.message === 'SignalsPubSub is closed') return; // drop late event\n  throw err;\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":"function isSignalsPubSubClosedError(err) {\n  return err instanceof Error && err.message === 'SignalsPubSub is closed';\n}","tryCatchPattern":"try {\n  await signals.publish(topic, event);\n} catch (err) {\n  if (isSignalsPubSubClosedError(err)) {\n    // instance closed: drop the event or recreate via createSignalsPubSub(resourceId)\n    return;\n  }\n  throw err;\n}","preventionTips":["Establish a single owner responsible for calling close(); never share the close responsibility across modules.","Track shutdown state in your app and stop emitting before teardown closes the pubsub.","Do not cache SignalsPubSub instances in globals that outlive their owner's lifecycle.","In tests, create a fresh pubsub per test and close it only in afterEach."],"tags":["pubsub","lifecycle","unix-socket","use-after-close"],"backgroundTag":"pubsub-closed","analyzedSha":"75dd419e613fe9c39f846ffc500716141b74fda6","analyzedAt":"2026-08-30T00:15:31.844Z","schemaVersion":2},"datasetVersion":"2026-08-30T03:17:51.788Z"}