Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
)

// TODO: add gap-free replay for arbitrary startRevs and drop this guard.
var ErrUnsupportedWatch = errors.New("cache: unsupported watch parameters")

// Cache buffers a single etcd Watch for a given key‐prefix and fan‑outs local watchers.
type Cache struct {
	prefix    string // prefix is the key-prefix this shard is responsible for ("" = root).
	cfg       Config // immutable runtime configuration
	watcher   clientv3.Watcher
	demux     *demux // demux fans incoming batches out to active watchers and manages resync.
	ready     chan struct{}
	stop      context.CancelFunc
	waitGroup sync.WaitGroup
}

// watchCtx collects all the knobs that both serveWatchEvents and watchRetryLoop need.
type watchCtx struct {
	cache           *Cache
	backoffStart    time.Duration
	backoffMax      time.Duration
	onFirstResponse func() // callback to fire once on first upstream response
}

// batch groups all events that share one ModRevision.
type batch struct {
	rev    int64
	events []*clientv3.Event
}

// New builds a cache shard that watches only the requested prefix.
// For the root cache pass "".
func New(watcher clientv3.Watcher, prefix string, opts ...Option) (*Cache, error) {
	cfg := defaultConfig()
	for _, opt := range opts {
		opt(&cfg)
	}

	if cfg.HistoryWindowSize <= 0 {
		return nil, fmt.Errorf("invalid HistoryWindowSize %d (must be > 0)", cfg.HistoryWindowSize)
	}

	internalCtx, cancel := context.WithCancel(context.Background())

	cache := &Cache{
		prefix:  prefix,
		cfg:     cfg,
		watcher: watcher,
		ready:   make(chan struct{}),
		stop:    cancel,
	}

Review comment (performance):
`batch`, the concept you introduced, is not a first-class. It's not standard to etcd API nor any etcd user. It's a structure you introduced for implementation of cache. An implementation detail. This means it should not be exposed externally and should be used in limited scope. Recommending to just limit to ringBuffer.

Basically reverse of replacing all mentions of `events` with `batch`.

Code after:
)

// TODO: add gap-free replay for arbitrary startRevs and drop this guard.
var ErrUnsupportedWatch = errors.New("cache: unsupported watch parameters")

// Cache buffers a single etcd Watch for a given key‐prefix and fan‑outs local watchers.
type Cache struct {
	prefix    string // prefix is the key-prefix this shard is responsible for ("" = root).
	cfg       Config // immutable runtime configuration
	watcher   clientv3.Watcher
	demux     *demux // demux fans incoming events out to active watchers and manages resync.
	ready     chan struct{}
	stop      context.CancelFunc
	waitGroup sync.WaitGroup
}

// watchCtx collects all the knobs that both serveWatchEvents and watchRetryLoop need.
type watchCtx struct {
	cache           *Cache
	backoffStart    time.Duration
	backoffMax      time.Duration
	onFirstResponse func() // callback to fire once on first upstream response
}

// New builds a cache shard that watches only the requested prefix.
// For the root cache pass "".
func New(watcher clientv3.Watcher, prefix string, opts ...Option) (*Cache, error) {
	cfg := defaultConfig()
	for _, opt := range opts {
		opt(&cfg)
	}

	if cfg.HistoryWindowSize <= 0 {
		return nil, fmt.Errorf("invalid HistoryWindowSize %d (must be > 0)", cfg.HistoryWindowSize)
	}

	internalCtx, cancel := context.WithCancel(context.Background())

	cache := &Cache{
		prefix:  prefix,
		cfg:     cfg,
		watcher: watcher,
		ready:   make(chan struct{}),
		stop:    cancel,
	}

	cache.demux = NewDemux(internalCtx, &cache.waitGroup, cfg.HistoryWindowSize, cfg.ResyncInterval)

	cache.waitGroup.Add(1)
	go func() {
		defer cache.waitGroup.Done()