hashicorp/nomad · error

subscription closed

Error message

subscription closed

What it means

eventBufferSubscription.Next blocks on a select over the caller's context, the link's force-close channel, and nextCh. When the buffer is torn down (shutdown or a competing close path) forceClose fires and Next returns the literal error "subscription closed" to signal the iterator is permanently finished. Callers must stop consuming; retrying will keep failing.

Source

Thrown at nomad/stream/event_buffer.go:264

		link: &bufferLink{
			nextCh:    make(chan struct{}),
			droppedCh: make(chan struct{}),
		},
		Events:    events,
		createdAt: time.Now(),
	}
}

// Next return the next buffer item in the buffer. It may block until ctx is
// cancelled or until the next item is published.
func (i *bufferItem) Next(ctx context.Context, forceClose <-chan struct{}) (*bufferItem, error) {
	// See if there is already a next value, block if so. Note we don't rely on
	// state change (chan nil) as that's not threadsafe but detecting close is.
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case <-forceClose:
		return nil, fmt.Errorf("subscription closed")
	case <-i.link.nextCh:
	}

	// Check if the reader is too slow and the event buffer as discarded the event
	// This must happen after the above select to prevent a random selection
	// between linkCh and droppedCh
	select {
	case <-i.link.droppedCh:
		return nil, fmt.Errorf("event dropped from buffer")
	default:
	}

	// If channel closed, there must be a next item to read
	nextRaw := i.link.next.Load()
	if nextRaw == nil {
		// shouldn't be possible
		return nil, errors.New("invalid next item")
	}

View on GitHub (pinned to 482b49bf1a)

Solutions

1. Stop the consumption loop and unwind the stream — the subscription cannot be resumed.
2. On the client, reconnect with a fresh Subscribe and resume from the last observed index (falling back to head if 'requested index not in buffer' occurs).
3. Ensure your loop exits on this error instead of busy-retrying Next().
4. Handle ctx cancellation alongside it so shutdowns propagate cleanly.

Example fix

// before
for { ev, err := sub.Next(ctx); if err != nil { continue } }
// after
for {
  ev, err := sub.Next(ctx)
  if err != nil {
    if err.Error() == "subscription closed" { return err } // terminal
    if ctx.Err() != nil { return ctx.Err() }
    return err
  }
  handle(ev)
}

func isSubscriptionClosed(err error) bool { return err != nil && err.Error() == "subscription closed" }

for {
  ev, err := sub.Next(ctx)
  if isSubscriptionClosed(err) || ctx.Err() != nil {
    return reconnectWithFreshSubscribe(ctx) // terminal: never call Next again
  }
  handle(ev)
}

When it happens

Trigger: Calling Next() on a subscription after the event buffer was closed — e.g. server shutdown, the broker stopping, or EndSubscription/close path triggered — while the consumer was blocked waiting for the next event.

Common situations: Nomad server restart or leader change while an HTTP event stream (job events, exec output) is open; client-side code looping over Next() without checking for terminal errors.

Related errors

• requested index not in buffer
• event dropped from buffer
• users: release of unused uid/gid
• missing AllocID
• <remote node streaming RPC error relayed from ack.Error>

AI-assisted analysis of hashicorp/nomad@482b49bf1a (2026-09-04). Data as JSON: /api/errors/00bd85a4d306a1cf. Report an issue: GitHub.