mastra-ai/mastra · error

SignalsPubSub is closed

Error message

SignalsPubSub is closed

What it means

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).

Source

Thrown at mastracode/sdk/src/utils/signals-pubsub.ts:75

  }

  async flush(): Promise<void> {
    await Promise.all([...this.#sockets.values()].map(s => s.flush()));
  }

  async close(): Promise<void> {
    this.#closed = true;
    await Promise.allSettled([...this.#sockets.values()].map(s => s.close()));
    this.#sockets.clear();
  }

  /** Get the underlying socket for a topic (for testing/inspection). */
  getSocket(topic: string): UnixSocketPubSub | undefined {
    return this.#sockets.get(this.#topicKey(topic));
  }

  async #getOrCreate(topic: string): Promise<UnixSocketPubSub> {
    if (this.#closed) throw new Error('SignalsPubSub is closed');
    const key = this.#topicKey(topic);
    const existing = this.#sockets.get(key);
    if (existing) return existing;
    // Deduplicate concurrent callers so only one socket is created per topic.
    let inflight = this.#pending.get(key);
    if (!inflight) {
      inflight = this.#initSocket(topic, key);
      this.#pending.set(key, inflight);
    }
    const socket = await inflight;
    if (this.#closed) throw new Error('SignalsPubSub is closed');
    return socket;
  }

  async #initSocket(topic: string, key: string): Promise<UnixSocketPubSub> {
    try {
      const socketPath = await this.#socketPath(topic);
      if (this.#closed) throw new Error('SignalsPubSub is closed');

View on GitHub (pinned to 75dd419e61)

Solutions

1. Do not call publish/subscribe after close(); guard your shutdown ordering so event emission finishes before teardown calls close().
2. 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.
3. If the instance was closed unexpectedly, create a fresh one via createSignalsPubSub(resourceId) and re-subscribe your handlers.
4. Audit for shared/singleton instances: ensure only one owner calls close() and others are notified.

Example fix

// before
await signals.publish(topic, event); // may throw after close()
// after
if (isShuttingDown) return; // skip emit during teardown
try {
  await signals.publish(topic, event);
} catch (err) {
  if (err instanceof Error && err.message === 'SignalsPubSub is closed') return; // drop late event
  throw err;
}

function isSignalsPubSubClosedError(err) {
  return err instanceof Error && err.message === 'SignalsPubSub is closed';
}

try {
  await signals.publish(topic, event);
} catch (err) {
  if (isSignalsPubSubClosedError(err)) {
    // instance closed: drop the event or recreate via createSignalsPubSub(resourceId)
    return;
  }
  throw err;
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• UnixSocketPubSub is closed
• UnixSocketPubSub does not support grouped subscriptions yet
• Stale broker election lock removed
• Broker election in progress by another process
• UnixSocketPubSub is not connected to a broker

AI-assisted analysis of mastra-ai/mastra@75dd419e61 (2026-08-30). Data as JSON: /api/errors/6dd4c37817f40b34. Report an issue: GitHub.