shared_poll.go

package centrifuge

import (
	"bytes"
	"context"
	"errors"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/centrifugal/centrifuge/internal/epoch"
	"github.com/centrifugal/protocol"
	"github.com/cespare/xxhash/v2"
	fdelta "github.com/shadowspore/fossil-delta"
)

// notifChCapacity bounds the per-channel notification queue used by
// SharedPollManager.notify. Each entry is one tracked-key notification waiting
// to be batched into the next backend poll. Sized to absorb typical bursts
// (large reconnect waves, cache-invalidation storms). When the buffer is full,
// extra notifications are dropped and counted by the droppedNotifyCount metric
// — the next timer-based poll cycle covers anything that was lost.
const notifChCapacity = 4096

// SharedPollManager manages all shared poll channels for a Node.
// One instance per Node. Thread-safe.
type SharedPollManager struct {
	node       *Node
	shutdownCh chan struct{}
	wg         sync.WaitGroup // tracks running refresh workers for graceful shutdown
	sem        chan struct{}  // shared concurrency limiter for backend calls
	epoch      string         // random string generated at startup, used for versionless mode

	mu       sync.RWMutex
	channels map[string]*sharedPollChannelState

	brokerSubMu    sync.RWMutex
	brokerSubChans map[string]Broker              // key-channel → subscribed broker
	brokerSubKeys  map[string]map[string]struct{} // base channel → set of key-channels
}

type sharedPollChannelState struct {
	mu sync.Mutex // protects all fields below

	// opts is captured at first track for this channel state and treated as
	// immutable for the lifetime of the state. The refresh worker reads opts
	// (RefreshInterval, batching, KeepLatestData, versioned/versionless mode, …)
	// without re-resolving — runtime config changes via GetSharedPollChannelOptions
	// only take effect after the channel state fully drains and is recreated
	// (i.e., after all subscribers leave and the shutdown delay expires).
	opts SharedPollChannelOptions

	// epoch is a random string generated when this channel state is created.
	// Clients compare epochs on reconnect and reset stored versions when
	// the epoch changes. This covers channel state recreation (e.g., after
	// shutdown delay expires and all items are cleaned up) without requiring
	// a full server restart.
	epoch string

	// itemIndex: key → tracked entry (version, data).
	itemIndex map[string]*sharedPollTrackedEntry

	// versionCounter is a monotonic counter for synthetic versions (versionless mode).
	versionCounter uint64

	// notifCh receives individual key notifications from SharedPollNotify.
	// Buffered, non-blocking send — drops if full and increments the
	// droppedNotifyCount metric. Capacity is sized for typical bursts; under
	// extreme load drops are surfaced via the metric and the next timer-based
	// poll closes the gap.
	notifCh chan string

	// refreshWorker lifecycle.
	workerRunning bool
	workerCancel  context.CancelFunc
	workerCtx     context.Context // checked by track() to detect cancelled worker
	workerGen     uint64          // incremented on each worker start, checked on exit
	// shutdownTimer delays channel cleanup after last item removed.
	shutdownTimer *time.Timer
	removed       bool // set by shutdown timer under mu; track() checks this
}

type sharedPollTrackedEntry struct {
	version          uint64 // last version from backend (or synthetic in versionless mode)
	data             []byte // only when KeepLatestData is true
	dataHash         uint64 // xxhash64 hash of data, only used in versionless mode
	freshFromPublish bool   // set by SharedPollPublish, cleared each timer poll cycle
	needsBroadcast   bool   // set when a version=0 subscriber joins an existing key; cleared after broadcast

	// subscribeReady is non-nil if the broker.Subscribe for this key is
	// currently in flight (called by an earlier track*() that already
	// installed the entry but released s.mu before calling the broker).
	// Concurrent track*() callers for the same key must wait on this chan
	// before reporting success — otherwise a subsequent subscribe failure
	// + rollback would orphan their caller's client in the keyed hub with
	// no broker subscription. The chan is closed (and nilled under s.mu)
	// once subscribe completes; the result is recorded in subscribeErr
	// just before close, so the chan-close memory barrier makes it visible
	// to waiters.
	subscribeReady chan struct{}
	subscribeErr   error

	// pendingHubJoin counts trackKeys callers that have successfully
	// reserved this entry but have NOT yet finalized — either by joining
	// the hub via keyedManager.addSubscribers or by rolling back. While
	// pendingHubJoin > 0, untrack must NOT delete the entry: a concurrent
	// trackKeys caller may have returned success and be about to join the
	// hub, and deleting now would silently orphan that caller (in hub,
	// no itemIndex, no broker subscription). Each trackKeys call
	// increments under s.mu and returns a release closure that decrements
	// under s.mu and runs untrack-style cleanup if the counter and hub
	// subscriber count are both zero.
	pendingHubJoin int
}

func xxHash64(data []byte) uint64 {
	return xxhash.Sum64(data)
}

// sharedPollKeyChannel builds a PUB/SUB channel name scoped to a specific key.
// Format: "<len>:<channel><key>" — length-prefix encoding, safe for any broker
// (no null bytes, no special characters in the framing).
func sharedPollKeyChannel(channel, key string) string {
	return strconv.Itoa(len(channel)) + ":" + channel + key
}

// parseSharedPollKeyChannel splits a key-scoped PUB/SUB channel into base channel and key.
// Returns ("", "") if the channel is not in the expected length-prefix format.
func parseSharedPollKeyChannel(keyCh string) (string, string) {
	i := strings.IndexByte(keyCh, ':')
	if i <= 0 {
		return "", ""
	}
	n, err := strconv.Atoi(keyCh[:i])
	if err != nil || n < 0 || i+1+n > len(keyCh) {
		return "", ""
	}
	return keyCh[i+1 : i+1+n], keyCh[i+1+n:]
}

func newSharedPollManager(node *Node) *SharedPollManager {
	concurrencyLimit := node.config.SharedPoll.ConcurrencyLimit
	if concurrencyLimit <= 0 {
		concurrencyLimit = 64
	}
	return &SharedPollManager{
		node:           node,
		shutdownCh:     make(chan struct{}),
		channels:       make(map[string]*sharedPollChannelState),
		sem:            make(chan struct{}, concurrencyLimit),
		brokerSubChans: make(map[string]Broker),
		brokerSubKeys:  make(map[string]map[string]struct{}),
		epoch:          epoch.Generate(),
	}
}

// noopReleaseTrack is returned from track/trackKeys on early-exit paths
// (shutdown, etc.) where no reservation was acquired. Lets callers
// unconditionally invoke release() without nil-checking.
func noopReleaseTrack() {}

// track registers an item in the shared poll channel state and ensures
// a refresh worker is running. Hub registration (addSubscriber) is handled
// by the generic keyed layer in handleTrack — NOT here.
//
// On success returns a release closure that the caller MUST invoke after
// either joining the hub via keyedManager.addSubscribers or definitively
// abandoning the track (rollback path). The closure decrements the
// pendingHubJoin reservation and cleans up the entry if it has no
// remaining holders. Without this contract a concurrent rollback in
// another goroutine could orphan an in-flight hub join — see
// pendingHubJoin doc.
func (m *SharedPollManager) track(channel string, opts SharedPollChannelOptions, key string) (bool, uint64, func(), error) {
	// Check global shutdown.
	select {
	case <-m.shutdownCh:
		return false, 0, noopReleaseTrack, nil
	default:
	}

	// Get or create channel state.
	m.mu.Lock()
	s, ok := m.channels[channel]
	if !ok {
		s = &sharedPollChannelState{
			opts:      opts,
			epoch:     initialChannelEpoch(opts),
			itemIndex: make(map[string]*sharedPollTrackedEntry),
			notifCh:   make(chan string, notifChCapacity),
		}
		m.channels[channel] = s
	}
	m.mu.Unlock()

	s.mu.Lock()
	// Re-check shutdown under state lock.
	select {
	case <-m.shutdownCh:
		s.mu.Unlock()
		return false, 0, noopReleaseTrack, nil
	default:
	}

	// If this state was removed by shutdown timer, replace it and loop
	// until we hold the lock on a non-removed state.
	//
	// Re-check m.channels[channel] under m.mu and reuse any fresh state a
	// concurrent caller installed — an unconditional overwrite would
	// orphan that caller's worker (nothing outside m.channels keeps a
	// reference for cancellation, so the lost worker would stay parked in
	// select with an uncancelled context, and m.close()'s wg.Wait would
	// hang forever).
	//
	// Loop, because the cur we ended up with may itself be removed by a
	// third goroutine between when it was placed and when we acquire
	// cur.mu — that would otherwise fall through to add a key + restart
	// the worker on an about-to-be-finalized state, leaking the new
	// worker the same way.
	for s.removed {
		s.mu.Unlock()
		m.mu.Lock()
		cur, ok := m.channels[channel]
		if !ok || cur == s {
			cur = &sharedPollChannelState{
				opts:      opts,
				epoch:     initialChannelEpoch(opts),
				itemIndex: make(map[string]*sharedPollTrackedEntry),
				notifCh:   make(chan string, notifChCapacity),
			}
			m.channels[channel] = cur
		}
		s = cur
		m.mu.Unlock()
		s.mu.Lock()
		select {
		case <-m.shutdownCh:
			s.mu.Unlock()
			return false, 0, noopReleaseTrack, nil
		default:
		}
	}

	// Cancel any pending shutdown timer.
	s.cancelShutdown()

	// Get or create itemIndex entry. New entries start at version=0.
	// Client-provided versions never enter itemIndex.
	isNewKey := s.itemIndex[key] == nil
	if isNewKey {
		s.itemIndex[key] = &sharedPollTrackedEntry{}
	}
	entry := s.itemIndex[key]
	entryVersion := entry.version
	// Reserve a hub-join slot. Decremented by the returned release closure
	// or by the internal failure paths below.
	entry.pendingHubJoin++

	// Ensure refresh worker is running.
	startWorker := false
	if !s.workerRunning {
		startWorker = true
	} else if s.workerCtx != nil {
		select {
		case <-s.workerCtx.Done():
			startWorker = true
		default:
		}
	}
	if startWorker {
		s.workerRunning = true
		s.workerGen++
		ctx, cancel := context.WithCancel(context.Background())
		s.workerCancel = cancel
		s.workerCtx = ctx
		gen := s.workerGen
		m.wg.Add(1)
		go s.runRefreshWorker(ctx, m.node, channel, gen, m)
	}

	// Identify ownership: either we OWN the in-flight subscribe for this
	// key (isNewKey + PublishEnabled), or another track*() owns it and we
	// must wait. ownEntry / waitCh capture pointers, not keys — between
	// here and the broker.Subscribe result a concurrent untrack+retrack
	// can replace s.itemIndex[key] with a different entry, and we must
	// only close/delete the entry WE created.
	//
	// Use s.opts.PublishEnabled (frozen at channel-state creation) rather
	// than the caller's opts.PublishEnabled. If the caller-passed opts
	// drift from the channel's first-tracker opts, gating on the caller's
	// flag would skip broker.Subscribe for a key on a PublishEnabled
	// channel — publish() then routes through the broker per s.opts but
	// this node is not subscribed for the key, so cross-node publications
	// silently miss local subscribers.
	publishEnabled := s.opts.PublishEnabled
	var ownEntry *sharedPollTrackedEntry
	var waitEntry *sharedPollTrackedEntry
	var waitCh chan struct{}
	if isNewKey {
		if publishEnabled {
			ownEntry = entry
			ownEntry.subscribeReady = make(chan struct{})
		}
	} else if entry.subscribeReady != nil {
		waitEntry = entry
		waitCh = entry.subscribeReady
	}
	s.mu.Unlock()

	reservations := []reservation{{key: key, entry: entry}}

	// Wait for the concurrent in-flight subscribe (if any). The chan close
	// is a memory barrier for waitEntry.subscribeErr.
	if waitCh != nil {
		<-waitCh
		if waitEntry.subscribeErr != nil {
			// The owner of the in-flight subscribe failed and rolled back
			// its entry. We must fail too — surfacing success would orphan
			// our caller in the keyed hub with no broker subscription.
			// Decrement our reservation (the wait entry; ownEntry is nil
			// in the wait branch). No synchronous channel cleanup needed
			// — the owner's failure path already handled that.
			m.releasePendingHubJoin(s, reservations)
			return false, 0, noopReleaseTrack, waitEntry.subscribeErr
		}
	}

	if ownEntry != nil {
		err := m.subscribeToBrokerKeys(channel, []string{key})
		s.mu.Lock()
		// Always notify waiters BEFORE touching itemIndex — the waiters
		// are waiting on ownEntry.subscribeReady, not on whatever is
		// currently in s.itemIndex[key] (a concurrent untrack+retrack may
		// have installed a different entry by now).
		ownEntry.subscribeErr = err
		close(ownEntry.subscribeReady)
		ownEntry.subscribeReady = nil
		if err != nil {
			// Synchronous cleanup of our owned entry (broker.Subscribe
			// failed so brokerSubChans has no record). Pointer identity
			// check protects against a concurrent untrack+retrack that
			// replaced s.itemIndex[key] with a different entry.
			cleanedUp := false
			if s.itemIndex[key] == ownEntry {
				delete(s.itemIndex, key)
				cleanedUp = true
			}
			empty := cleanedUp && len(s.itemIndex) == 0
			if empty && s.workerCancel != nil {
				s.workerCancel()
				s.removed = true
			}
			s.mu.Unlock()
			if empty {
				m.mu.Lock()
				if m.channels[channel] == s {
					delete(m.channels, channel)
				}
				m.mu.Unlock()
			}
			// Decrement reservations. ownEntry's counter is moot (entry
			// deleted) but the call is symmetric and harmless.
			m.releasePendingHubJoin(s, reservations)
			return false, 0, noopReleaseTrack, err
		}
		s.mu.Unlock()
	}

	release := func() {
		m.releaseTrackReservations(channel, s, reservations)
	}
	return isNewKey, entryVersion, release, nil
}

// trackKeyResult holds the outcome for a single key tracked via trackKeys.
type trackKeyResult struct {
	isNew        bool
	entryVersion uint64
}

// ownedKey pairs a key with the entry pointer this trackKeys call
// installed for it. Pointer identity matters: a concurrent untrack +
// retrack can replace s.itemIndex[key] with a different entry while our
// broker.Subscribe is in flight, and we must only close/delete the entry
// WE own — touching another goroutine's entry would either double-close
// its subscribeReady chan (panic) or erase its registration.
type ownedKey struct {
	key   string
	entry *sharedPollTrackedEntry
}

// reservation pairs a key with the entry pointer whose pendingHubJoin
// this trackKeys call incremented. Used to undo the increment (and
// possibly clean up the entry) on rollback or release. Pointer identity
// matters: a concurrent untrack+retrack may have installed a different
// entry by the time we run cleanup, and decrementing the wrong entry
// would underflow it or interfere with another caller's bookkeeping.
type reservation struct {
	key   string
	entry *sharedPollTrackedEntry
}

// releaseTrackReservations decrements pendingHubJoin on each reserved
// entry and performs untrack-style cleanup for any entry whose counter
// reaches zero AND has no remaining hub subscribers. The hub-count check
// closes the inter-client orphan race: another in-flight trackKeys
// caller (waiter who returned success and is about to call
// addSubscribers) keeps the entry alive via its own reservation until
// it has joined the hub.
//
// Pointer identity (s.itemIndex[key] == r.entry) is required before
// deleting — a concurrent untrack+retrack may have installed a different
// entry, and deleting now would erase that caller's freshly installed
// entry.
//
// Used by the release closure returned to successful trackKeys callers,
// invoked after either addSubscribers (entry stays via hub.count>0) or
// rollback (entry deletes if no other holders).
func (m *SharedPollManager) releaseTrackReservations(channel string, s *sharedPollChannelState, reservations []reservation) {
	if len(reservations) == 0 {
		return
	}
	hub := m.node.keyedManager.getHub(channel)

	var cleanupKeys []string
	s.mu.Lock()
	for _, r := range reservations {
		if r.entry.pendingHubJoin > 0 {
			r.entry.pendingHubJoin--
		}
		if r.entry.pendingHubJoin > 0 {
			continue
		}
		if s.itemIndex[r.key] != r.entry {
			continue
		}
		if hub != nil && hub.subscriberCount(r.key) > 0 {
			continue
		}
		delete(s.itemIndex, r.key)
		cleanupKeys = append(cleanupKeys, r.key)
	}
	empty := len(s.itemIndex) == 0
	publishEnabled := s.opts.PublishEnabled
	s.mu.Unlock()

	if publishEnabled && len(cleanupKeys) > 0 {
		m.unsubscribeFromBrokerKeys(channel, cleanupKeys)
	}
	if empty {
		s.scheduleShutdown(m, channel, s.opts.ChannelShutdownDelay)
	}
}

// releasePendingHubJoin decrements pendingHubJoin on each reservation
// without performing any entry or channel-state cleanup. Used by
// trackKeys / track internal failure paths (broker.Subscribe failure or
// wait-failure) where the failing path already runs its own synchronous
// cleanup — we just need to drop the reservations we held on wait
// entries so other concurrent trackKeys callers see a correct count.
// Owned entries are deleted synchronously by the failing path; their
// counter goes away with the entry, so the decrement is harmless.
func (m *SharedPollManager) releasePendingHubJoin(s *sharedPollChannelState, reservations []reservation) {
	if len(reservations) == 0 {
		return
	}
	s.mu.Lock()
	for _, r := range reservations {
		if r.entry.pendingHubJoin > 0 {
			r.entry.pendingHubJoin--
		}
	}
	s.mu.Unlock()
}

// trackKeys registers multiple items in the shared poll channel state in
// one batch. It subscribes to all new keys with a single broker.Subscribe
// call. On success returns a release closure the caller MUST invoke
// after either joining the hub via keyedManager.addSubscribers or
// definitively abandoning the track (rollback path). See track() and
// pendingHubJoin doc for the orphan race this guards against.
func (m *SharedPollManager) trackKeys(channel string, opts SharedPollChannelOptions, keys []string) ([]trackKeyResult, func(), error) {
	if len(keys) == 0 {
		return nil, noopReleaseTrack, nil
	}

	// Check global shutdown.
	select {
	case <-m.shutdownCh:
		return make([]trackKeyResult, len(keys)), noopReleaseTrack, nil
	default:
	}

	// Get or create channel state.
	m.mu.Lock()
	s, ok := m.channels[channel]
	if !ok {
		s = &sharedPollChannelState{
			opts:      opts,
			epoch:     initialChannelEpoch(opts),
			itemIndex: make(map[string]*sharedPollTrackedEntry),
			notifCh:   make(chan string, notifChCapacity),
		}
		m.channels[channel] = s
	}
	m.mu.Unlock()

	s.mu.Lock()
	// Re-check shutdown under state lock.
	select {
	case <-m.shutdownCh:
		s.mu.Unlock()
		return make([]trackKeyResult, len(keys)), noopReleaseTrack, nil
	default:
	}

	// If this state was removed by shutdown timer, replace it. See the
	// matching loop in track() for the race rationale: both an
	// unconditional overwrite and a single-shot re-check would still
	// orphan a worker — a fresh state we land on may itself be in the
	// process of being shut down by a third goroutine.
	for s.removed {
		s.mu.Unlock()
		m.mu.Lock()
		cur, ok := m.channels[channel]
		if !ok || cur == s {
			cur = &sharedPollChannelState{
				opts:      opts,
				epoch:     initialChannelEpoch(opts),
				itemIndex: make(map[string]*sharedPollTrackedEntry),
				notifCh:   make(chan string, notifChCapacity),
			}
			m.channels[channel] = cur
		}
		s = cur
		m.mu.Unlock()
		s.mu.Lock()
		select {
		case <-m.shutdownCh:
			s.mu.Unlock()
			return make([]trackKeyResult, len(keys)), noopReleaseTrack, nil
		default:
		}
	}

	// Cancel any pending shutdown timer.
	s.cancelShutdown()

	// Register all keys and collect results, owned-new-entry pointers,
	// reservation list (for pendingHubJoin bookkeeping), and any in-flight
	// subscribe chans we must wait on. We capture entry pointers (not key
	// strings) because a concurrent untrack+retrack can replace
	// s.itemIndex[key] with a different entry between here and when our
	// broker.Subscribe completes — close/delete must only touch the
	// entries WE installed, and pending-counter decrement must only touch
	// the entry we incremented.
	//
	// Use s.opts.PublishEnabled (frozen at channel-state creation) rather
	// than the caller's opts.PublishEnabled. See track() for the drift
	// rationale — gating broker.Subscribe on the caller's flag would
	// silently skip subscriptions on a PublishEnabled channel and miss
	// cross-node publications for that key.
	publishEnabled := s.opts.PublishEnabled
	results := make([]trackKeyResult, len(keys))
	var owned []ownedKey
	reservations := make([]reservation, 0, len(keys))
	// reservedSet dedups: if the same key appears multiple times in the
	// keys slice, we reserve once per unique entry. Otherwise duplicate
	// keys would over-increment pendingHubJoin and the matching release
	// would underflow (or worse, decrement another caller's count).
	reservedSet := make(map[*sharedPollTrackedEntry]struct{}, len(keys))
	// ownedSet identifies entries this call installed so a duplicate key
	// later in the same `keys` slice does not get classified as someone
	// else's in-flight subscribe — which would have us wait on our own
	// subscribeReady chan and deadlock.
	var ownedSet map[*sharedPollTrackedEntry]struct{}
	type waitInfo struct {
		entry *sharedPollTrackedEntry
		ch    chan struct{}
	}
	var waits []waitInfo
	for i, key := range keys {
		entry := s.itemIndex[key]
		isNewKey := entry == nil
		if isNewKey {
			entry = &sharedPollTrackedEntry{}
			if publishEnabled {
				entry.subscribeReady = make(chan struct{})
			}
			s.itemIndex[key] = entry
			owned = append(owned, ownedKey{key: key, entry: entry})
			if ownedSet == nil {
				ownedSet = make(map[*sharedPollTrackedEntry]struct{}, len(keys))
			}
			ownedSet[entry] = struct{}{}
		} else if entry.subscribeReady != nil {
			if _, mine := ownedSet[entry]; !mine {
				waits = append(waits, waitInfo{entry: entry, ch: entry.subscribeReady})
			}
		}
		if _, dup := reservedSet[entry]; !dup {
			entry.pendingHubJoin++
			reservations = append(reservations, reservation{key: key, entry: entry})
			reservedSet[entry] = struct{}{}
		}
		results[i] = trackKeyResult{
			isNew:        isNewKey,
			entryVersion: entry.version,
		}
	}

	// Ensure refresh worker is running.
	startWorker := false
	if !s.workerRunning {
		startWorker = true
	} else if s.workerCtx != nil {
		select {
		case <-s.workerCtx.Done():
			startWorker = true
		default:
		}
	}
	if startWorker {
		s.workerRunning = true
		s.workerGen++
		ctx, cancel := context.WithCancel(context.Background())
		s.workerCancel = cancel
		s.workerCtx = ctx
		gen := s.workerGen
		m.wg.Add(1)
		go s.runRefreshWorker(ctx, m.node, channel, gen, m)
	}

	s.mu.Unlock()

	// Wait for any concurrent in-flight broker subscribes for keys we did
	// not create. The chan close acts as a memory barrier for subscribeErr.
	// If any in-flight subscribe failed, our track fails too — surfacing
	// success while a key's broker subscription is missing would orphan
	// the caller's client in the keyed hub.
	for _, w := range waits {
		<-w.ch
		if w.entry.subscribeErr != nil {
			// Roll back the entries WE created so subsequent callers can
			// retry from scratch (synchronous channel-state cleanup), and
			// decrement our pending counter on wait entries.
			m.rollbackOwnedKeys(channel, s, owned, w.entry.subscribeErr)
			m.releasePendingHubJoin(s, reservations)
			return nil, noopReleaseTrack, w.entry.subscribeErr
		}
	}

	if len(owned) > 0 && publishEnabled {
		ownedKeyStrs := make([]string, len(owned))
		for i, o := range owned {
			ownedKeyStrs[i] = o.key
		}
		err := m.subscribeToBrokerKeys(channel, ownedKeyStrs)
		s.mu.Lock()
		// Always notify waiters BEFORE touching itemIndex — they're
		// waiting on our captured ownEntry pointers, not on whatever the
		// map currently holds (a concurrent untrack+retrack may have
		// installed a different entry).
		for _, o := range owned {
			o.entry.subscribeErr = err
			close(o.entry.subscribeReady)
			o.entry.subscribeReady = nil
		}
		if err != nil {
			// Synchronous cleanup of our owned entries (broker.Subscribe
			// failed; brokerSubChans has no record). Pointer identity
			// check guards against concurrent untrack+retrack.
			for _, o := range owned {
				if s.itemIndex[o.key] == o.entry {
					delete(s.itemIndex, o.key)
				}
			}
			empty := len(s.itemIndex) == 0
			if empty && s.workerCancel != nil {
				s.workerCancel()
				s.removed = true
			}
			s.mu.Unlock()
			if empty {
				m.mu.Lock()
				if m.channels[channel] == s {
					delete(m.channels, channel)
				}
				m.mu.Unlock()
			}
			m.releasePendingHubJoin(s, reservations)
			return nil, noopReleaseTrack, err
		}
		s.mu.Unlock()
	}

	release := func() {
		m.releaseTrackReservations(channel, s, reservations)
	}
	return results, release, nil
}

// rollbackOwnedKeys signals failure on each owned entry's subscribeReady
// chan and deletes the entry from itemIndex when it is still the installed
// one. Used when a concurrent in-flight subscribe (for a key we did NOT
// own) failed — we must abandon any entries we ourselves installed in this
// same trackKeys call so subsequent track*() callers retry from scratch.
func (m *SharedPollManager) rollbackOwnedKeys(channel string, s *sharedPollChannelState, owned []ownedKey, err error) {
	if len(owned) == 0 {
		return
	}
	s.mu.Lock()
	for _, o := range owned {
		if o.entry.subscribeReady != nil {
			o.entry.subscribeErr = err
			close(o.entry.subscribeReady)
			o.entry.subscribeReady = nil
		}
		if s.itemIndex[o.key] == o.entry {
			delete(s.itemIndex, o.key)
		}
	}
	empty := len(s.itemIndex) == 0
	if empty && s.workerCancel != nil {
		s.workerCancel()
		s.removed = true
	}
	s.mu.Unlock()
	if empty {
		m.mu.Lock()
		if m.channels[channel] == s {
			delete(m.channels, channel)
		}
		m.mu.Unlock()
	}
}

// untrack removes an item from shared poll tracking when no connections
// remain. Called after the client has been removed from the hub via
// keyedHub.removeSubscriber.
//
// The pendingHubJoin guard prevents deleting an entry while a concurrent
// trackKeys caller still holds a reservation — that caller may have
// returned success and be about to join the hub, and deleting now would
// silently orphan it (in hub, no itemIndex, no broker subscription). See
// pendingHubJoin doc on sharedPollTrackedEntry.
func (m *SharedPollManager) untrack(channel string, key string) {
	m.mu.RLock()
	s, ok := m.channels[channel]
	m.mu.RUnlock()
	if !ok {
		return
	}

	hub := m.node.keyedManager.getHub(channel)

	s.mu.Lock()
	if hub != nil && hub.subscriberCount(key) > 0 {
		s.mu.Unlock()
		return
	}
	if entry := s.itemIndex[key]; entry != nil && entry.pendingHubJoin > 0 {
		s.mu.Unlock()
		return
	}
	delete(s.itemIndex, key)
	empty := len(s.itemIndex) == 0
	publishEnabled := s.opts.PublishEnabled
	s.mu.Unlock()

	if publishEnabled {
		m.unsubscribeFromBrokerKeys(channel, []string{key})
	}
	if empty {
		s.scheduleShutdown(m, channel, s.opts.ChannelShutdownDelay)
	}
}

// warmKeyData holds cached data for a warm key that can be delivered directly.
type warmKeyData struct {
	key             string
	internalVersion uint64                // synthetic/real version for per-connection dedup
	pub             *protocol.Publication // publication with wire version
}

// getWarmKeyData returns cached data for warm keys. Only returns entries when
// KeepLatestData is enabled and the entry has data (version > 0).
func (m *SharedPollManager) getWarmKeyData(channel string, keys []string) []warmKeyData {
	m.mu.RLock()
	s, ok := m.channels[channel]
	m.mu.RUnlock()
	if !ok {
		return nil
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.opts.KeepLatestData || s.opts.isVersionless() {
		return nil
	}
	var result []warmKeyData
	for _, key := range keys {
		entry := s.itemIndex[key]
		if entry == nil || entry.version == 0 || entry.data == nil {
			continue
		}
		result = append(result, warmKeyData{
			key:             key,
			internalVersion: entry.version,
			pub: &protocol.Publication{
				Key:     key,
				Data:    entry.data,
				Version: entry.version,
			},
		})
	}
	return result
}

// markNeedsBroadcast flags existing keys so polls re-broadcast their data even
// if unchanged. Also triggers a notify for keys that have data (version > 0) and
// were not already flagged — this provides near-immediate delivery via the
// notification fast path instead of waiting for the next timer poll. The
// at-most-once notify per key ensures mass reconnect (1000 clients, same key)
// causes only one backend call per "wave", not one per client.
//
// Must be called AFTER addSubscriber so that broadcasts from the triggered notify
// (or a concurrent timer poll) can reach the subscribing client.
func (m *SharedPollManager) markNeedsBroadcast(channel string, keys []string) {
	m.mu.RLock()
	s, ok := m.channels[channel]
	m.mu.RUnlock()
	if !ok {
		return
	}
	s.mu.Lock()
	var toNotify []string
	for _, key := range keys {
		if entry := s.itemIndex[key]; entry != nil {
			if !entry.needsBroadcast {
				entry.needsBroadcast = true
				// Only notify keys that already have data — version=0 keys are
				// being handled by an in-flight cold key notify.
				if entry.version > 0 {
					toNotify = append(toNotify, key)
				}
			}
		}
	}
	s.mu.Unlock()

	// Notify outside the lock — triggers backend call for near-immediate delivery.
	// Keys are combined by the worker's batch dedup when batching is configured.
	for _, key := range toNotify {
		m.notify(channel, key)
	}
}

// notify sends a key notification to the channel's notification channel.
// Non-blocking: drops the notification if the buffer is full.
// Silently ignores unknown channels (channel must already be tracked).
//
// The read lock is held across the send so that a concurrent shutdown cannot
// replace m.channels[channel] with a fresh state while we send into the now-
// orphaned notifCh (whose worker has been cancelled and will never drain it).
// The send is non-blocking (select with default) and the metric increments are
// cheap, so holding the RLock here does not introduce contention.
func (m *SharedPollManager) notify(channel string, key string) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	s, ok := m.channels[channel]
	if !ok {
		return
	}
	// Non-blocking send — drop if full.
	select {
	case s.notifCh <- key:
		m.node.metrics.getSharedPollChannelCached(channel).notifyCount.Inc()
	default:
		m.node.metrics.getSharedPollChannelCached(channel).droppedNotifyCount.Inc()
	}
}

// getCachedData returns cached publications for items where the server has a newer
// version than the client. Returns nil when nothing to return (omitted from protobuf).
// Only returns data when KeepLatestData is enabled for the channel.
func (m *SharedPollManager) getCachedData(channel string, items []TrackItem) []*protocol.Publication {
	m.mu.RLock()
	s, ok := m.channels[channel]
	m.mu.RUnlock()
	if !ok {
		return nil
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.opts.KeepLatestData || s.opts.isVersionless() {
		return nil
	}
	var pubs []*protocol.Publication
	for _, item := range items {
		entry := s.itemIndex[item.Key]
		if entry == nil || entry.version == 0 || entry.data == nil {
			continue
		}
		if entry.version <= item.Version {
			continue
		}
		pubs = append(pubs, &protocol.Publication{
			Key:     item.Key,
			Data:    entry.data,
			Version: entry.version,
		})
	}
	return pubs
}

// hasChannel reports whether the SharedPollManager has state for the given channel.
func (m *SharedPollManager) hasChannel(channel string) bool {
	m.mu.RLock()
	_, ok := m.channels[channel]
	m.mu.RUnlock()
	return ok
}

// initialChannelEpoch returns the epoch a freshly-created sharedPollChannelState
// starts with. For versionless mode the server generates a per-channel epoch
// (changes when state is recreated). For versioned mode the epoch is supplied
// by the publisher on each publish/refresh, so initial state is empty and
// gets populated via flipEpoch on the first incoming publish.
func initialChannelEpoch(opts SharedPollChannelOptions) string {
	if opts.isVersionless() {
		return epoch.Generate()
	}
	return ""
}

// Epoch returns the epoch string for subscribe replies.
//
// Versionless channels: server-generated, stable for the lifetime of channel
// state. Changes when state is recreated after shutdown delay expires.
//
// Versioned channels: publisher-supplied, set on first publish/refresh.
// Changes (epoch flip) trigger unsubscribe of all current subscribers with
// insufficient-state code so they re-track from version 0 on resubscribe.
func (m *SharedPollManager) Epoch(channel string, isVersionless bool) string {
	m.mu.RLock()
	s, ok := m.channels[channel]
	m.mu.RUnlock()
	if !ok {
		if isVersionless {
			return m.epoch
		}
		return ""
	}
	s.mu.Lock()
	e := s.epoch
	s.mu.Unlock()
	return e
}

// flipEpochAndCollectClients atomically updates the stored channel epoch,
// resets all per-key state, and collects the *Client references currently
// subscribed (via the keyed hub). Caller must invoke Client.Unsubscribe on
// each returned client with no locks held — calling Unsubscribe under
// s.mu would deadlock since Unsubscribe's cleanup path takes s.mu via
// untrack.
//
// All state mutation (epoch + entries) and the client snapshot are taken
// under s.mu to prevent a window where a client subscribing concurrently
// with the flip could be unsubbed despite having seen the new epoch in
// its subscribe reply.
//
// If newEpoch == s.epoch, returns nil and performs no mutation.
func (s *sharedPollChannelState) flipEpochAndCollectClients(hub *keyedHub, newEpoch string) []*Client {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.epoch == newEpoch {
		return nil
	}
	s.epoch = newEpoch
	for _, e := range s.itemIndex {
		e.version = 0
		e.data = nil
		e.dataHash = 0
		e.freshFromPublish = false
		e.needsBroadcast = false
	}
	if hub == nil {
		return nil
	}
	// Nested s.mu -> h.mu is consistent with existing patterns in this file
	// (e.g. SharedPollRevokeKeys uses removeAllSubscribers under s.mu).
	return hub.collectAllClients()
}

// brokerChannelsSnapshot returns the key channels currently subscribed at the
// given broker. Key channels are subscribed at broker level and tracked only
// here — not in the Hub — so a PUB/SUB reconnect would silently drop them:
// the reconnect resubscribes Hub().Channels() and nothing else. The Redis
// PUB/SUB loops call this (via Node) to restore key channels together with
// Hub channels after re-establishing a connection.
//
// Brokers are matched by interface equality. That is well defined for every
// built-in broker (all pointer types); a Broker implemented as a
// non-comparable value type — a struct with a map or slice field, used by
// value — would panic on the comparison below.
func (m *SharedPollManager) brokerChannelsSnapshot(b Broker) []string {
	m.brokerSubMu.RLock()
	defer m.brokerSubMu.RUnlock()
	channels := make([]string, 0, len(m.brokerSubChans))
	for kc, broker := range m.brokerSubChans {
		if broker == b {
			channels = append(channels, kc)
		}
	}
	return channels
}

// close stops all refresh workers and waits for them to finish.
func (m *SharedPollManager) close() {
	close(m.shutdownCh)
	m.mu.Lock()
	for _, ch := range m.channels {
		ch.mu.Lock()
		if ch.workerCancel != nil {
			ch.workerCancel()
		}
		if ch.shutdownTimer != nil {
			ch.shutdownTimer.Stop()
		}
		ch.mu.Unlock()
	}
	m.mu.Unlock()
	m.wg.Wait()

	// Unsubscribe all broker subscriptions, batched by broker.
	m.brokerSubMu.Lock()
	brokerChannels := make(map[Broker][]string)
	for kc, broker := range m.brokerSubChans {
		brokerChannels[broker] = append(brokerChannels[broker], kc)
	}
	for broker, channels := range brokerChannels {
		if err := broker.Unsubscribe(channels...); err != nil {
			m.node.logger.log(newLogEntry(LogLevelError, "error unsubscribing from broker for shared poll", map[string]any{"error": err.Error()}))
		}
	}
	m.brokerSubChans = make(map[string]Broker)
	m.brokerSubKeys = make(map[string]map[string]struct{})
	m.brokerSubMu.Unlock()
}

func (m *SharedPollManager) subscribeToBrokerKeys(channel string, keys []string) error {
	if len(keys) == 0 {
		return nil
	}
	broker := m.node.getBroker(channel)
	keyChannels := make([]string, len(keys))
	for i, key := range keys {
		keyChannels[i] = sharedPollKeyChannel(channel, key)
	}
	m.node.metrics.incActionCount("broker_subscribe", channel)
	// Per-channel sharded lock — mirrors Node.addSubscription so that a
	// concurrent shared-poll unsubscribe (queued via subDissolver) cannot
	// race between our broker.Subscribe and brokerSubChans update and
	// silently unsubscribe the key at the broker after we've recorded it
	// as subscribed. Without this, the queued unsub's filter check sees
	// the key absent (because we hadn't installed our entry yet when it
	// ran), proceeds to broker.Unsubscribe(kc) after we have already
	// re-subscribed, and then deletes brokerSubChans[kc] — leaving the
	// new tracker in the hub with no broker subscription.
	mu := m.node.subLock(channel)
	mu.Lock()
	defer mu.Unlock()
	if err := broker.Subscribe(keyChannels...); err != nil {
		return err
	}
	m.brokerSubMu.Lock()
	if m.brokerSubKeys[channel] == nil {
		m.brokerSubKeys[channel] = make(map[string]struct{})
	}
	for _, kc := range keyChannels {
		m.brokerSubChans[kc] = broker
		m.brokerSubKeys[channel][kc] = struct{}{}
	}
	m.brokerSubMu.Unlock()
	return nil
}

func (m *SharedPollManager) unsubscribeFromBrokerKeys(channel string, keys []string) {
	if len(keys) == 0 {
		return
	}
	keyChannels := make([]string, len(keys))
	for i, key := range keys {
		keyChannels[i] = sharedPollKeyChannel(channel, key)
	}
	_ = m.node.subDissolver.Submit(func() error {
		m.node.metrics.incActionCount("broker_unsubscribe", channel)
		// Hold the per-channel sharded lock across the filter recheck, the
		// broker.Unsubscribe call, and the brokerSubChans update — so a
		// concurrent retrack (which also takes the lock in
		// subscribeToBrokerKeys) cannot squeeze a fresh broker.Subscribe
		// between our filter check and our Unsubscribe. Mirrors the
		// Node.removeSubscription pattern.
		mu := m.node.subLock(channel)
		mu.Lock()
		defer mu.Unlock()
		// Filter out keys that were re-tracked between queueing and execution.
		m.mu.RLock()
		s, chExists := m.channels[channel]
		m.mu.RUnlock()
		var toUnsub []string
		if chExists {
			s.mu.Lock()
			for i, kc := range keyChannels {
				if _, tracked := s.itemIndex[keys[i]]; !tracked {
					toUnsub = append(toUnsub, kc)
				}
			}
			s.mu.Unlock()
		} else {
			toUnsub = keyChannels
		}
		if len(toUnsub) == 0 {
			return nil
		}
		m.brokerSubMu.RLock()
		broker, ok := m.brokerSubChans[toUnsub[0]]
		m.brokerSubMu.RUnlock()
		if !ok {
			return nil
		}
		if err := broker.Unsubscribe(toUnsub...); err != nil {
			time.Sleep(500 * time.Millisecond)
			return err
		}
		m.brokerSubMu.Lock()
		for _, kc := range toUnsub {
			delete(m.brokerSubChans, kc)
		}
		if ks, ok := m.brokerSubKeys[channel]; ok {
			for _, kc := range toUnsub {
				delete(ks, kc)
			}
			if len(ks) == 0 {
				delete(m.brokerSubKeys, channel)
			}
		}
		m.brokerSubMu.Unlock()
		return nil
	})
}

// unsubscribeAllBrokerKeys unsubscribes all key-channels for a base channel.
// Used on channel shutdown.
func (m *SharedPollManager) unsubscribeAllBrokerKeys(channel string) {
	_ = m.node.subDissolver.Submit(func() error {
		m.node.metrics.incActionCount("broker_unsubscribe", channel)
		// Per-channel sharded lock — same rationale as unsubscribeFromBrokerKeys.
		// A concurrent track() that re-creates the channel state and issues a
		// fresh broker.Subscribe could otherwise race the shutdown-time
		// Unsubscribe and end up with a tracked key but no broker subscription.
		mu := m.node.subLock(channel)
		mu.Lock()
		defer mu.Unlock()
		// Check if channel was re-created while shutting down.
		m.mu.RLock()
		_, stillActive := m.channels[channel]
		m.mu.RUnlock()
		if stillActive {
			return nil
		}
		m.brokerSubMu.RLock()
		ks, ok := m.brokerSubKeys[channel]
		if !ok || len(ks) == 0 {
			m.brokerSubMu.RUnlock()
			return nil
		}
		keyChannels := make([]string, 0, len(ks))
		for kc := range ks {
			keyChannels = append(keyChannels, kc)
		}
		var broker Broker
		for _, kc := range keyChannels {
			if b, ok := m.brokerSubChans[kc]; ok {
				broker = b
				break
			}
		}
		m.brokerSubMu.RUnlock()
		if broker == nil {
			return nil
		}
		if err := broker.Unsubscribe(keyChannels...); err != nil {
			time.Sleep(500 * time.Millisecond)
			return err
		}
		m.brokerSubMu.Lock()
		for _, kc := range keyChannels {
			delete(m.brokerSubChans, kc)
		}
		if ks, ok := m.brokerSubKeys[channel]; ok {
			for _, kc := range keyChannels {
				delete(ks, kc)
			}
			if len(ks) == 0 {
				delete(m.brokerSubKeys, channel)
			}
		}
		m.brokerSubMu.Unlock()
		return nil
	})
}

func (m *SharedPollManager) publish(ctx context.Context, channel string, key string, version uint64, epoch string, data []byte) error {
	// Resolve channel options. Prefer the running channel state's opts
	// (immutable for its lifetime). Fall back to the config callback so a
	// publisher node that has never tracked the channel still gets the
	// correct routing decision.
	var opts SharedPollChannelOptions
	var hasOpts bool

	m.mu.RLock()
	s, ok := m.channels[channel]
	m.mu.RUnlock()
	if ok {
		s.mu.Lock()
		opts = s.opts
		s.mu.Unlock()
		hasOpts = true
	} else if m.node.config.SharedPoll.GetSharedPollChannelOptions != nil {
		opts, hasOpts = m.node.config.SharedPoll.GetSharedPollChannelOptions(channel)
	}
	if !hasOpts {
		return errors.New("SharedPollPublish: channel not configured for shared poll")
	}
	if opts.isVersionless() {
		return errors.New("SharedPollPublish not supported in versionless refresh mode")
	}

	// Routing is purely a function of PublishEnabled — not of local
	// subscriber state. When PublishEnabled is true we always go through
	// the broker so cross-node subscribers receive the publication,
	// regardless of whether this node has tracked the key (and regardless
	// of any in-flight broker.Subscribe). For shared-poll keyed channels
	// the broker's Publication.Epoch carries the publisher's per-channel
	// epoch (the wire field is shared with stream channels' stream-position
	// epoch — semantics differ per channel type, but the wire is the same).
	if opts.PublishEnabled {
		broker := m.node.getBroker(channel)
		keyCh := sharedPollKeyChannel(channel, key)
		_, err := broker.Publish(keyCh, data, PublishOptions{Key: key, Version: version, Epoch: epoch})
		return err
	}
	// Local-only mode: apply directly. Only this node's subscribers see it.
	m.handlePublishedData(channel, key, version, epoch, data)
	return nil
}

// stats returns the number of active channels and total tracked keys.
func (m *SharedPollManager) stats() (int, int) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	numChannels := len(m.channels)
	numKeys := 0
	for _, s := range m.channels {
		s.mu.Lock()
		numKeys += len(s.itemIndex)
		s.mu.Unlock()
	}
	return numChannels, numKeys
}

func (m *SharedPollManager) handlePublishedData(channel string, key string, version uint64, epoch string, data []byte) {
	m.mu.RLock()
	s, ok := m.channels[channel]
	m.mu.RUnlock()
	if !ok {
		return
	}
	hub := m.node.keyedManager.getHub(channel)
	if hub == nil {
		return
	}

	if s.opts.isVersionless() {
		// SharedPollPublish should not reach here in versionless mode
		// (rejected by publish()), but guard defensively.
		return
	}

	// Epoch comparison runs *before* the entry-lookup early-return so the
	// channel's stored epoch stays current even on nodes that don't track
	// this specific key. Otherwise the first subscriber to ever track a
	// key on a quiet node would trigger an unnecessary unsubscribe cycle
	// when the cold-key auto-poll detects the existing publisher epoch.
	if clients := s.flipEpochAndCollectClients(hub, epoch); len(clients) > 0 {
		for _, c := range clients {
			c.Unsubscribe(channel, unsubscribeInsufficientState)
		}
	}

	s.mu.Lock()
	entry := s.itemIndex[key]
	if entry == nil {
		s.mu.Unlock()
		return // Key not tracked on this node.
	}
	if version <= entry.version {
		s.mu.Unlock()
		m.node.metrics.getSharedPollPublishCached(channel).skipped.Inc()
		return
	}
	prevVersion := entry.version
	entry.version = version
	entry.freshFromPublish = true
	var prevData []byte
	if s.opts.KeepLatestData {
		prevData = entry.data
		entry.data = data
	}
	s.mu.Unlock()

	m.node.metrics.getSharedPollPublishCached(channel).applied.Inc()

	pub := &protocol.Publication{Key: key, Data: data, Version: version, Epoch: epoch}
	prep := buildPreparedPollData(pub, prevData, prevVersion)
	hub.broadcastToKey(channel, key, version, pub, prep)
}

// SharedPollRevokeKeys removes items for matching connections. Sends
// Publication{Removed: true} to affected subscribers and cleans up hub + itemIndex.
func (m *SharedPollManager) SharedPollRevokeKeys(channel string, keys []string, users []string, excludeUsers []string) {
	m.mu.RLock()
	s, ok := m.channels[channel]
	m.mu.RUnlock()
	if !ok {
		return
	}

	hub := m.node.keyedManager.getHub(channel)
	if hub == nil {
		return
	}

	// Broadcast removals to affected subscribers.
	for _, key := range keys {
		if len(users) == 0 && len(excludeUsers) == 0 {
			hub.broadcastRemoval(channel, key)
		} else {
			hub.broadcastRemovalToUsers(channel, key, users, excludeUsers)
		}
	}

	// Clean up hub and itemIndex. Track keys we actually delete from
	// itemIndex so we can drop their broker subscriptions too — otherwise
	// the broker keeps pushing cross-node publications that handlePublishedData
	// silently no-ops (entry == nil), wasting traffic until full channel
	// shutdown.
	var brokerCleanupKeys []string
	s.mu.Lock()
	publishEnabled := s.opts.PublishEnabled
	for _, key := range keys {
		if len(users) == 0 && len(excludeUsers) == 0 {
			hub.removeAllSubscribers(key)
		} else {
			hub.removeSubscribersForUsers(key, users, excludeUsers)
		}
		// Remove from itemIndex if no subscribers remain AND no in-flight
		// trackKeys reservation is holding the key alive. Mirrors the
		// untrack guard so a concurrent track caller about to addSubscribers
		// is not orphaned.
		if hub.subscriberCount(key) == 0 {
			if entry := s.itemIndex[key]; entry != nil && entry.pendingHubJoin == 0 {
				delete(s.itemIndex, key)
				if publishEnabled {
					brokerCleanupKeys = append(brokerCleanupKeys, key)
				}
			}
		}
	}
	empty := len(s.itemIndex) == 0
	s.mu.Unlock()

	if len(brokerCleanupKeys) > 0 {
		m.unsubscribeFromBrokerKeys(channel, brokerCleanupKeys)
	}
	if empty {
		s.scheduleShutdown(m, channel, s.opts.ChannelShutdownDelay)
	}
}

// scheduleShutdown starts a delayed cleanup timer.
// A zero delay means default (1s). Use a negative delay (e.g. -1) for immediate shutdown.
func (s *sharedPollChannelState) scheduleShutdown(m *SharedPollManager, channel string, delay time.Duration) {
	s.mu.Lock()

	if s.shutdownTimer != nil {
		s.shutdownTimer.Stop()
	}

	if delay == 0 {
		delay = 1 * time.Second
	}

	if delay < 0 {
		// Immediate shutdown: mirror the timer body. Release s.mu before
		// touching m.mu — lock ordering is m.mu → s.mu everywhere.
		s.doShutdownLocked(m, channel)
		s.mu.Unlock()
		s.finalizeShutdown(m, channel)
		return
	}

	s.shutdownTimer = time.AfterFunc(delay, func() {
		select {
		case <-m.shutdownCh:
			return
		default:
		}
		s.mu.Lock()
		if len(s.itemIndex) > 0 {
			s.shutdownTimer = nil
			s.mu.Unlock()
			return
		}
		s.doShutdownLocked(m, channel)
		s.mu.Unlock()
		s.finalizeShutdown(m, channel)
	})
	s.mu.Unlock()
}

// doShutdownLocked marks the state as removed and cancels the worker. Caller
// must hold s.mu and must not call any m.mu operation while holding s.mu —
// lock ordering is m.mu → s.mu. The follow-up m.mu work is done in
// finalizeShutdown after the caller releases s.mu.
func (s *sharedPollChannelState) doShutdownLocked(m *SharedPollManager, channel string) {
	if s.workerCancel != nil {
		s.workerCancel()
	}
	s.removed = true
	s.shutdownTimer = nil
}

// finalizeShutdown removes the channel from the manager and unsubscribes broker
// keys. Must be called with s.mu released (acquires m.mu and may call back into
// keyedManager / dissolver). Safe to call when m.channels[channel] no longer
// points to s — that branch is a no-op.
//
// The keyedManager call is removeChannelIfEmpty (not removeChannel) to avoid
// a race: a new client whose handleTrack ran concurrently with our shutdown
// may have already added itself as a subscriber via
// keyedManager.addSubscribers, populating the channel's hub. An
// unconditional removeChannel would delete that state and orphan the new
// subscriber from future broadcasts (which look up via getHub).
func (s *sharedPollChannelState) finalizeShutdown(m *SharedPollManager, channel string) {
	m.mu.Lock()
	if m.channels[channel] == s {
		delete(m.channels, channel)
		m.mu.Unlock()
		m.node.keyedManager.removeChannelIfEmpty(channel)
	} else {
		m.mu.Unlock()
	}
	m.unsubscribeAllBrokerKeys(channel)
}

// cancelShutdown stops a pending shutdown timer. Caller must hold s.mu.
func (s *sharedPollChannelState) cancelShutdown() {
	if s.shutdownTimer != nil {
		s.shutdownTimer.Stop()
		s.shutdownTimer = nil
	}
}

func (s *sharedPollChannelState) runRefreshWorker(ctx context.Context, node *Node, channel string, gen uint64, m *SharedPollManager) {
	defer m.wg.Done()
	defer func() {
		s.mu.Lock()
		if s.workerGen == gen {
			s.workerRunning = false
		}
		s.mu.Unlock()
	}()

	interval := s.opts.RefreshInterval
	if interval <= 0 {
		interval = 10 * time.Second
	}
	timer := time.NewTimer(interval)
	defer timer.Stop()

	// Notification batching state.
	batchMaxSize := s.opts.NotificationBatchMaxSize
	batchMaxDelay := s.opts.NotificationBatchMaxDelay
	// When size-based batching is configured without a delay, use the
	// refresh interval as the cap so notifications don't sit indefinitely.
	if batchMaxSize > 0 && batchMaxDelay <= 0 {
		batchMaxDelay = interval
	}
	batchEnabled := batchMaxSize > 0 || batchMaxDelay > 0
	var batchKeys map[string]struct{}
	var batchTimer *time.Timer
	var batchTimerC <-chan time.Time
	defer func() {
		if batchTimer != nil {
			batchTimer.Stop()
		}
	}()

	fireBatch := func() {
		if len(batchKeys) == 0 {
			return
		}
		keys := make([]string, 0, len(batchKeys))
		for k := range batchKeys {
			keys = append(keys, k)
		}
		batchKeys = nil
		if batchTimer != nil {
			batchTimer.Stop()
			batchTimerC = nil
		}
		s.runNotifiedRefreshCycle(ctx, node, channel, keys, m.sem)
	}

	for {
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
			cycleStart := time.Now()
			spreadDelay := s.runRefreshCycle(ctx, node, channel, m.sem)
			cycleDuration := time.Since(cycleStart)
			cc := node.metrics.getSharedPollChannelCached(channel)
			cc.cycleDuration.Observe(cycleDuration.Seconds())
			workTime := cycleDuration - spreadDelay
			if workTime < 0 {
				workTime = 0
			}
			cc.cycleWorkDuration.Observe(workTime.Seconds())
			if s.opts.RefreshIntervalFn != nil {
				// Pass work time (excluding intentional spread delays)
				// so backpressure reacts to actual backend load.
				interval = s.opts.RefreshIntervalFn(workTime)
				if interval <= 0 {
					interval = s.opts.RefreshInterval
					if interval <= 0 {
						interval = 10 * time.Second
					}
				}
			}
			// Period-based timer: subtract cycle duration so the
			// period stays close to the configured interval.
			remaining := interval - cycleDuration
			if remaining < 0 {
				remaining = 0
			}
			timer.Reset(remaining)
		case key := <-s.notifCh:
			if !batchEnabled {
				// No batching — fire immediately.
				s.runNotifiedRefreshCycle(ctx, node, channel, []string{key}, m.sem)
				continue
			}
			if batchKeys == nil {
				batchKeys = make(map[string]struct{})
			}
			batchKeys[key] = struct{}{}
			if batchMaxSize > 0 && len(batchKeys) >= batchMaxSize {
				fireBatch()
			} else if batchMaxDelay > 0 && batchTimerC == nil {
				batchTimer = time.NewTimer(batchMaxDelay)
				batchTimerC = batchTimer.C
			}
		case <-batchTimerC:
			fireBatch()
		}
	}
}

// runNotifiedRefreshCycle runs an immediate backend poll for just the notified keys.
// Unlike runRefreshCycle, there's no spread delay or chunking — the batch is already bounded.
func (s *sharedPollChannelState) runNotifiedRefreshCycle(ctx context.Context, node *Node, channel string, keys []string, sem chan struct{}) {
	// Filter to keys still in itemIndex.
	s.mu.Lock()
	filtered := keys[:0]
	for _, k := range keys {
		if _, ok := s.itemIndex[k]; ok {
			filtered = append(filtered, k)
		}
	}
	s.mu.Unlock()
	if len(filtered) == 0 {
		return
	}

	hub := node.keyedManager.getHub(channel)
	if hub == nil {
		return
	}

	handler := node.clientEvents.sharedPollHandler
	if handler == nil {
		return
	}

	callTimeout := s.opts.CallTimeout
	if callTimeout <= 0 {
		callTimeout = 30 * time.Second
	}

	hc := node.metrics.getSharedPollHandlerCached("notification", channel)

	// Acquire semaphore.
	semStart := time.Now()
	select {
	case sem <- struct{}{}:
	case <-ctx.Done():
		return
	}
	defer func() { <-sem }()
	hc.semWait.Observe(time.Since(semStart).Seconds())

	// Build event items.
	items := make([]SharedPollItem, len(filtered))
	if !s.opts.isVersionless() {
		s.mu.Lock()
		for i, key := range filtered {
			items[i] = SharedPollItem{Key: key}
			if entry, ok := s.itemIndex[key]; ok {
				if entry.needsBroadcast {
					items[i].Version = 0
				} else {
					items[i].Version = entry.version
				}
			}
		}
		s.mu.Unlock()
	} else {
		for i, key := range filtered {
			items[i] = SharedPollItem{Key: key}
		}
	}

	hc.itemsPolled.Add(float64(len(items)))

	callCtx, cancel := context.WithTimeout(ctx, callTimeout)
	callStart := time.Now()
	result, err := handler(callCtx, SharedPollEvent{
		Channel: channel,
		Items:   items,
	})
	cancel()
	hc.duration.Observe(time.Since(callStart).Seconds())

	if err != nil {
		hc.errorCount.Inc()
		if node.logEnabled(LogLevelWarn) {
			node.logger.log(newLogEntry(LogLevelWarn, "shared poll notified refresh error", map[string]any{
				"channel": channel,
				"error":   err.Error(),
			}))
		}
		return
	}

	// Notified refresh does not track absences — absence tracking is only
	// meaningful for full-channel timer-based polls.
	s.onNotifiedRefreshResponse(channel, result.Epoch, result.Items, hub, node)
}

func (s *sharedPollChannelState) runRefreshCycle(ctx context.Context, node *Node, channel string, sem chan struct{}) time.Duration {
	// 1. Collect all item keys from itemIndex.
	s.mu.Lock()
	if len(s.itemIndex) == 0 {
		s.mu.Unlock()
		return 0
	}
	keys := make([]string, 0, len(s.itemIndex))
	for k, entry := range s.itemIndex {
		if entry.freshFromPublish {
			entry.freshFromPublish = false // Clear flag.
			continue                       // Skip — data is fresh from publish.
		}
		keys = append(keys, k)
	}
	s.mu.Unlock()

	if len(keys) == 0 {
		return 0
	}

	sort.Strings(keys)

	batchSize := s.opts.RefreshBatchSize
	if batchSize <= 0 {
		batchSize = 1000
	}

	// Split into chunks.
	chunks := make([][]string, 0, (len(keys)+batchSize-1)/batchSize)
	for i := 0; i < len(keys); i += batchSize {
		end := i + batchSize
		if end > len(keys) {
			end = len(keys)
		}
		chunks = append(chunks, keys[i:end])
	}

	hub := node.keyedManager.getHub(channel)
	if hub == nil {
		return 0
	}

	callTimeout := s.opts.CallTimeout
	if callTimeout <= 0 {
		callTimeout = 30 * time.Second
	}

	handler := node.clientEvents.sharedPollHandler
	if handler == nil {
		return 0
	}

	// Spread chunk dispatches evenly over the refresh interval to avoid
	// bursting all backend calls at once. The delay between dispatches is
	// interval / num_chunks. The semaphore still limits actual concurrency.
	var dispatchDelay time.Duration
	if len(chunks) > 1 {
		interval := s.opts.RefreshInterval
		if interval <= 0 {
			interval = 10 * time.Second
		}
		dispatchDelay = interval / time.Duration(len(chunks))
	}

	var wg sync.WaitGroup

	for i, chunk := range chunks {
		if ctx.Err() != nil {
			break
		}
		if i > 0 && dispatchDelay > 0 {
			// time.NewTimer + explicit Stop avoids leaking a timer per iteration
			// when ctx is cancelled (time.After leaves the unselected timer to
			// fire later, garbage-collected only after the duration elapses).
			timer := time.NewTimer(dispatchDelay)
			select {
			case <-ctx.Done():
				if !timer.Stop() {
					<-timer.C
				}
			case <-timer.C:
			}
			if ctx.Err() != nil {
				break
			}
		}

		hc := node.metrics.getSharedPollHandlerCached("timer", channel)
		semStart := time.Now()
		sem <- struct{}{}
		wg.Add(1)
		go func(chunk []string, semStart time.Time) {
			defer func() {
				<-sem
				wg.Done()
			}()
			hc.semWait.Observe(time.Since(semStart).Seconds())

			// Build event items.
			items := make([]SharedPollItem, len(chunk))
			if !s.opts.isVersionless() {
				s.mu.Lock()
				for i, key := range chunk {
					items[i] = SharedPollItem{Key: key}
					if entry, ok := s.itemIndex[key]; ok {
						if entry.needsBroadcast {
							items[i].Version = 0
						} else {
							items[i].Version = entry.version
						}
					}
				}
				s.mu.Unlock()
			} else {
				for i, key := range chunk {
					items[i] = SharedPollItem{Key: key}
				}
			}

			hc.itemsPolled.Add(float64(len(items)))

			callCtx, cancel := context.WithTimeout(ctx, callTimeout)
			callStart := time.Now()
			result, err := handler(callCtx, SharedPollEvent{
				Channel: channel,
				Items:   items,
			})
			cancel()
			hc.duration.Observe(time.Since(callStart).Seconds())

			if err != nil {
				hc.errorCount.Inc()
				if node.logEnabled(LogLevelWarn) {
					node.logger.log(newLogEntry(LogLevelWarn, "shared poll refresh error", map[string]any{
						"channel": channel,
						"error":   err.Error(),
					}))
				}
				return
			}

			s.onRefreshResponse(channel, result.Epoch, result.Items, hub, node)
		}(chunk, semStart)
	}

	wg.Wait()

	// Return total intentional spread delay so callers (e.g. backpressure)
	// can distinguish work time from spread time.
	totalSpreadDelay := time.Duration(len(chunks)-1) * dispatchDelay
	return totalSpreadDelay
}

type pendingBroadcast struct {
	key     string
	version uint64
	pub     *protocol.Publication
	prep    preparedData
	removal bool
}

// buildPreparedPollData computes delta data for a shared poll publication.
// If prevData is available, computes a fossil delta patch; otherwise returns
// an empty preparedData (no delta). prevVersion is the version of the bytes
// in prevData (entry.version BEFORE this publish); keyedWritePublication
// uses it to verify the delta's base matches the client's current data,
// falling back to FULL when concurrent broadcasts mean the client never
// received the version this patch is built against. The actual per-client
// encoding (JSON escaping, protocol framing) is done lazily in
// keyedWritePublication.
func buildPreparedPollData(pub *protocol.Publication, prevData []byte, prevVersion uint64) preparedData {
	if len(prevData) == 0 {
		return preparedData{}
	}
	patch := fdelta.Create(prevData, pub.Data)
	isReal := len(patch) < len(pub.Data)
	deltaData := patch
	if !isReal {
		deltaData = pub.Data
	}
	return preparedData{
		deltaSub:              true,
		keyedDeltaPatch:       deltaData,
		keyedDeltaIsReal:      isReal,
		keyedDeltaPrevVersion: prevVersion,
	}
}

// onNotifiedRefreshResponse processes a backend response triggered by a
// notification (targeted poll). Identical to onRefreshResponse except for the
// metric source label.
func (s *sharedPollChannelState) onNotifiedRefreshResponse(channel string, respEpoch string, items []SharedPollRefreshItem, hub *keyedHub, node *Node) {
	s.applyRefreshResponse(channel, respEpoch, items, hub, node, "notification")
}

// onRefreshResponse processes a backend response from the periodic timer poll.
func (s *sharedPollChannelState) onRefreshResponse(channel string, respEpoch string, items []SharedPollRefreshItem, hub *keyedHub, node *Node) {
	s.applyRefreshResponse(channel, respEpoch, items, hub, node, "timer")
}

// applyRefreshResponse is the shared implementation for both notified and
// timer-driven refresh responses. The two source kinds differ only in the
// metric label used for instrumentation — all per-item handling (versioned vs
// versionless change detection, epoch flip, broadcast, removals, prev_data /
// KeepLatestData delta base capture) is identical.
func (s *sharedPollChannelState) applyRefreshResponse(channel string, respEpoch string, items []SharedPollRefreshItem, hub *keyedHub, node *Node, source string) {
	// Versioned channels: detect publisher epoch change before any per-item
	// processing. A flip resets all per-key state and unsubscribes current
	// subscribers; the items in this response then repopulate state under
	// the new epoch via the standard path below.
	if !s.opts.isVersionless() {
		if clients := s.flipEpochAndCollectClients(hub, respEpoch); len(clients) > 0 {
			for _, c := range clients {
				c.Unsubscribe(channel, unsubscribeInsufficientState)
			}
		}
	}

	type pendingUpdate struct {
		key         string
		version     uint64
		data        []byte
		prevData    []byte
		prevVersion uint64
	}

	updates := make([]pendingUpdate, 0, len(items))
	var removals []string
	var changedCount, unchangedCount int

	s.mu.Lock()

	isVersionless := s.opts.isVersionless()

	for _, e := range items {
		if e.Removed {
			removals = append(removals, e.Key)
			continue
		}

		entry := s.itemIndex[e.Key]
		if entry == nil {
			continue
		}

		if isVersionless && e.Version == 0 {
			// Versionless backend response: detect changes by content hash.
			changed := false
			if s.opts.KeepLatestData && entry.version > 0 {
				changed = !bytes.Equal(entry.data, e.Data)
			} else if entry.version > 0 {
				newHash := xxHash64(e.Data)
				changed = newHash != entry.dataHash
				if changed {
					entry.dataHash = newHash
				}
			} else {
				// First data for this key.
				changed = true
				if !s.opts.KeepLatestData {
					entry.dataHash = xxHash64(e.Data)
				}
			}
			if !changed {
				unchangedCount++
				if entry.needsBroadcast && entry.version > 0 {
					entry.needsBroadcast = false
					updates = append(updates, pendingUpdate{
						key: e.Key, version: entry.version, data: e.Data,
					})
				}
				continue
			}
			entry.needsBroadcast = false
			changedCount++
			s.versionCounter++
			syntheticVersion := s.versionCounter
			prevVersion := entry.version
			var prevData []byte
			if s.opts.KeepLatestData {
				prevData = entry.data
				entry.data = e.Data
			}
			entry.version = syntheticVersion
			updates = append(updates, pendingUpdate{
				key: e.Key, version: syntheticVersion, data: e.Data, prevData: prevData, prevVersion: prevVersion,
			})
			continue
		}

		if e.Version <= entry.version {
			unchangedCount++
			if entry.needsBroadcast && entry.version > 0 {
				// Choose (version, data) so the pair always matches. Without
				// this, a concurrent SharedPollPublish that bumped
				// entry.version while the backend response was in flight
				// would have us emit (entry.version, e.Data) — the new
				// (higher) version paired with the old bytes. The receiving
				// client advances keyState.version to entry.version, then
				// suppresses any later legitimate broadcast at the same
				// version, pinning the client to wrong data until the next
				// entry update.
				//
				// KeepLatestData: entry.data is the live publish payload at
				// entry.version. Safe to broadcast both; clear needsBroadcast.
				//
				// !KeepLatestData: entry.data is empty. The only valid pair
				// we can emit is the backend's (e.Version, e.Data). Emitting
				// that pins the client to the backend's older version, which
				// is fine — a future publish at any higher version will
				// pass the keyState.version filter. But only do so when the
				// versions match (no concurrent publish raced ahead); if
				// they don't, leave needsBroadcast set so the next poll
				// retries once the backend catches up.
				if s.opts.KeepLatestData {
					entry.needsBroadcast = false
					updates = append(updates, pendingUpdate{
						key: e.Key, version: entry.version, data: entry.data,
					})
				} else if e.Version == entry.version {
					entry.needsBroadcast = false
					updates = append(updates, pendingUpdate{
						key: e.Key, version: e.Version, data: e.Data,
					})
				}
				// else: backend behind a concurrent publish; retry next poll.
			}
			continue
		}

		entry.needsBroadcast = false
		changedCount++
		prevVersion := entry.version
		// Capture prevData before updating — this is the delta base.
		var prevData []byte
		if s.opts.KeepLatestData {
			prevData = entry.data
			entry.data = e.Data
		}
		if len(e.PrevData) > 0 && !s.opts.KeepLatestData {
			prevData = e.PrevData
		}
		entry.version = e.Version

		updates = append(updates, pendingUpdate{
			key: e.Key, version: e.Version, data: e.Data, prevData: prevData, prevVersion: prevVersion,
		})
	}

	s.mu.Unlock()

	rc := node.metrics.getSharedPollResultCached(source, channel)
	rc.changed.Add(float64(changedCount))
	rc.unchanged.Add(float64(unchangedCount))
	rc.removed.Add(float64(len(removals)))

	// Build publications outside the lock. Batch-allocate Publication structs
	// in a single slice to avoid one heap allocation per changed key.
	broadcasts := make([]pendingBroadcast, 0, len(updates)+len(removals))
	pubs := make([]protocol.Publication, len(updates))
	for i, u := range updates {
		pubs[i] = protocol.Publication{Key: u.key, Data: u.data, Version: u.version}
		prep := buildPreparedPollData(&pubs[i], u.prevData, u.prevVersion)
		broadcasts = append(broadcasts, pendingBroadcast{key: u.key, version: u.version, pub: &pubs[i], prep: prep})
	}
	for _, key := range removals {
		broadcasts = append(broadcasts, pendingBroadcast{key: key, removal: true})
	}

	// Fan out outside the lock.
	for _, b := range broadcasts {
		if b.removal {
			hub.broadcastRemoval(channel, b.key)
		} else {
			hub.broadcastToKey(channel, b.key, b.version, b.pub, b.prep)
		}
	}

	// Clean up removed items from hub and itemIndex.
	if len(removals) > 0 {
		s.mu.Lock()
		for _, key := range removals {
			hub.removeAllSubscribers(key)
			delete(s.itemIndex, key)
		}
		s.mu.Unlock()
	}
}