mastra-ai/mastra · error · Error

UnixSocketPubSub does not support grouped subscriptions yet

Error message

UnixSocketPubSub does not support grouped subscriptions yet

What it means

UnixSocketPubSub does not implement consumer-group semantics, where multiple subscribers share messages from a topic. Passing options.group to subscribe() throws immediately rather than silently ignoring the grouping option.

Solutions

  1. Remove the group option and subscribe without it
  2. If load balancing is required, switch to a pubsub implementation that supports grouped subscriptions
  3. Implement app-level fan-out partitioning (e.g. route by event key in your own callback)
  4. Check the backend's feature matrix before relying on group semantics

Example fix

// before
await pubsub.subscribe("jobs", handler, { group: "workers" });
// after
await pubsub.subscribe("jobs", handler);
Defensive patterns

Strategy: fallback

Validate before calling

if ('group' in options && options.group) {
  throw new Error('Backend does not support grouped subscriptions');
}
await pubsub.subscribe(topic, cb);

Try / catch

try {
  await pubsub.subscribe(topic, cb, options);
} catch (e) {
  if (e.message.includes('grouped subscriptions')) {
    await pubsub.subscribe(topic, cb); // fall back to ungrouped
  } else throw e;
}

Prevention

When it happens

Trigger: Calling pubsub.subscribe(topic, callback, { group: "workers" }) — any non-empty group option in SubscribeOptions on a UnixSocketPubSub instance, typically from setupPair.

Common situations: Code shared between a pubsub backend that supports groups (e.g. Redis-based) and this Unix-socket backend, load-balancing multiple workers over one topic, copy-pasted subscription code.

Related errors


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

Appendix: source

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

      this.#deliverLocal(topic, localEvent);
      return;
    }

    if (this.#isBroker) {
      await this.#publishFromBroker(topic, event, undefined, options?.localOnly);
      return;
    }

    const socket = this.#clientSocket;
    if (!socket || socket.destroyed) {
      await this.#ensureStarted(true);
    }
    await this.#sendToBroker({ type: 'publish', topic, event, localOnly: options?.localOnly });
  }

  async subscribe(topic: string, cb: EventCallback, options?: SubscribeOptions): Promise<void> {
    if (options?.group) {
      throw new Error('UnixSocketPubSub does not support grouped subscriptions yet');
    }

    const callbacks = this.#callbacks.get(topic) ?? new Set<EventCallback>();
    const hadCallback = callbacks.has(cb);
    const wasConnected = Boolean(this.#clientSocket && !this.#clientSocket.destroyed);
    callbacks.add(cb);
    this.#callbacks.set(topic, callbacks);

    try {
      await this.#ensureStarted();
      if (!this.#isBroker && !hadCallback && wasConnected) {
        await this.#sendSubscribeToBroker(topic);
      }
    } catch (error) {
      if (!hadCallback) {
        callbacks.delete(cb);
        if (callbacks.size === 0) {
          this.#callbacks.delete(topic);

View on GitHub (pinned to 75dd419e61)