What causes "UnixSocketPubSub is closed" in mastra-ai/mastra?

Reorder shutdown: cancel all publishers/subscribers and timers before calling close()

mastra-ai/mastra · error · Error

UnixSocketPubSub is closed

Error message

UnixSocketPubSub is closed

What it means

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

Source

Thrown at packages/core/src/events/unix-socket-pubsub.ts:285

    for (const client of [...this.#brokerClients.values()]) {
      this.#removeBrokerClient(client);
    }

    if (this.#server) {
      await new Promise<void>(resolve => this.#server?.close(() => resolve()));
      this.#server = undefined;
    }

    if (this.#isBroker) {
      await unlink(this.socketPath).catch(() => {});
    }
    this.#isBroker = false;
  }

  async #ensureStarted(forceReconnect = false): Promise<void> {
    if (this.#closed) {
      throw new Error('UnixSocketPubSub is closed');
    }
    if (!forceReconnect && (this.#isBroker || (this.#clientSocket && !this.#clientSocket.destroyed))) {
      return;
    }
    if (this.#starting) {
      return this.#starting;
    }

    this.#starting = this.#start(forceReconnect).finally(() => {
      this.#starting = undefined;
    });
    return this.#starting;
  }

  async #start(forceReconnect: boolean): Promise<void> {
    if (forceReconnect) {
      this.#clientSocket?.destroy();
      this.#clientSocket = undefined;

View on GitHub (pinned to 75dd419e61)

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

Example fix

// before
await pubsub.close();
await pubsub.publish("events", payload); // throws
// after
await pubsub.publish("events", payload);
await pubsub.close();

Defensive patterns

Strategy: try-catch

Validate before calling

if (pubsub.isClosed?.()) {
  throw new Error('Skipping publish: pubsub already closed');
}
await pubsub.publish(topic, event);

Type guard

function isUsable(ps: UnixSocketPubSub): boolean {
  return !ps.isClosed?.();
}

Try / catch

try {
  await pubsub.publish(topic, event);
} catch (e) {
  if (e.message === 'UnixSocketPubSub is closed') {
    logger.warn('Publish after close; dropping event', { topic });
  } else throw e;
}

Prevention

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

When it happens

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

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

Related errors

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