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.

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)

Solutions

Remove the group option and subscribe without it
If load balancing is required, switch to a pubsub implementation that supports grouped subscriptions
Implement app-level fan-out partitioning (e.g. route by event key in your own callback)
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

Check backend capability before passing SubscribeOptions.group
Abstract subscriptions behind an interface that only exposes group on supporting backends
Add a unit test asserting group options are only sent to group-capable pubsub implementations
Document group support per pubsub backend in your infra docs

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.