ronantakizawa/codereview-bench huggingface dataset:
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
	requestedRev := op.Rev()

	// build the internal Watch event stream (for historic reply + live)
	stream, w, err := c.newWatchEventStream(ctx, key, requestedRev, opts)
	if err != nil {
		// terminal error → one-shot reply, needs a single buffer slot
		responseChan := make(chan clientv3.WatchResponse, 1)
		if errors.Is(err, rpctypes.ErrCompacted) {
			compactRev := c.demux.PeekOldest()
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}
		}
		close(responseChan)
		return responseChan
	}

	ctx, cancel := context.WithCancel(ctx)
	go func() { <-w.Done(); cancel() }()

	responseChan := make(chan clientv3.WatchResponse)
	go func() {
		defer cancel()
		defer close(responseChan)

		for eventBatch := range stream {
			select {
			case <-ctx.Done():
				return
			case responseChan <- clientv3.WatchResponse{Events: eventBatch}:
			}
		}
	}()
	return responseChan
}

// Ready reports whether the cache has finished its initial load.
func (c *Cache) Ready() bool {
	select {
	case <-c.ready:
		return true
	default:
		return false
	}
}

// WaitReady blocks until the cache is ready or the ctx is cancelled.
func (c *Cache) WaitReady(ctx context.Context) error {
	select {
	case <-c.ready:
		return nil

Review comment (suggestion):
```suggestion
		for events := range stream {
```

Code after:
	requestedRev := op.Rev()

	// build the internal Watch event stream (for historic reply + live)
	stream, w, err := c.newWatchEventStream(ctx, key, requestedRev, opts)
	if err != nil {
		// terminal error → one-shot reply, needs a single buffer slot
		responseChan := make(chan clientv3.WatchResponse, 1)
		if errors.Is(err, rpctypes.ErrCompacted) {
			compactRev := c.demux.PeekOldest()
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}
		}
		close(responseChan)
		return responseChan
	}

	ctx, cancel := context.WithCancel(ctx)
	go func() { <-w.Done(); cancel() }()

	responseChan := make(chan clientv3.WatchResponse)
	go func() {
		defer cancel()
		defer close(responseChan)

		for events := range stream {
			select {
			case <-ctx.Done():
				return
			case responseChan <- clientv3.WatchResponse{Events: events}:
			}
		}
	}()
	return responseChan
}

// Ready reports whether the cache has finished its initial load.
func (c *Cache) Ready() bool {
	select {
	case <-c.ready:
		return true
	default:
		return false
	}
}

// WaitReady blocks until the cache is ready or the ctx is cancelled.
func (c *Cache) WaitReady(ctx context.Context) error {
	select {
	case <-c.ready:
		return nil
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
	if err != nil {
		// terminal error → one-shot reply, needs a single buffer slot
		responseChan := make(chan clientv3.WatchResponse, 1)
		if errors.Is(err, rpctypes.ErrCompacted) {
			compactRev := c.demux.PeekOldest()
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}
		}
		close(responseChan)
		return responseChan
	}

	ctx, cancel := context.WithCancel(ctx)
	go func() { <-w.Done(); cancel() }()

	responseChan := make(chan clientv3.WatchResponse)
	go func() {
		defer cancel()
		defer close(responseChan)

		for eventBatch := range stream {
			select {
			case <-ctx.Done():
				return
			case responseChan <- clientv3.WatchResponse{Events: eventBatch}:
			}
		}
	}()
	return responseChan
}

// Ready reports whether the cache has finished its initial load.
func (c *Cache) Ready() bool {
	select {
	case <-c.ready:
		return true
	default:
		return false
	}
}

// WaitReady blocks until the cache is ready or the ctx is cancelled.
func (c *Cache) WaitReady(ctx context.Context) error {
	select {
	case <-c.ready:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

Review comment (suggestion):
```suggestion
			case responseChan <- clientv3.WatchResponse{Events: events}:
```

Code after:
	if err != nil {
		// terminal error → one-shot reply, needs a single buffer slot
		responseChan := make(chan clientv3.WatchResponse, 1)
		if errors.Is(err, rpctypes.ErrCompacted) {
			compactRev := c.demux.PeekOldest()
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}
		}
		close(responseChan)
		return responseChan
	}

	ctx, cancel := context.WithCancel(ctx)
	go func() { <-w.Done(); cancel() }()

	responseChan := make(chan clientv3.WatchResponse)
	go func() {
		defer cancel()
		defer close(responseChan)

		for events := range stream {
			select {
			case <-ctx.Done():
				return
			case responseChan <- clientv3.WatchResponse{Events: events}:
			}
		}
	}()
	return responseChan
}

// Ready reports whether the cache has finished its initial load.
func (c *Cache) Ready() bool {
	select {
	case <-c.ready:
		return true
	default:
		return false
	}
}

// WaitReady blocks until the cache is ready or the ctx is cancelled.
func (c *Cache) WaitReady(ctx context.Context) error {
	select {
	case <-c.ready:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
		if oldestRev != 0 && nextRev < oldestRev {
			w.Stop()
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		missed := d.history.Filter(AfterRev(nextRev))

		for _, eventBatch := range missed {
			if !w.enqueueEvent(eventBatch) { // buffer overflow: watcher still lagging
				break
			}
			nextRev = eventBatch[0].Kv.ModRevision + 1
		}

		if len(missed) > 0 && nextRev > missed[len(missed)-1][0].Kv.ModRevision {
			delete(d.laggingWatchers, w)
			d.activeWatchers[w] = nextRev
		} else {
			d.laggingWatchers[w] = nextRev
		}
	}
}

func (d *demux) PeekOldest() int64 {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return d.history.PeekOldest()
}


Review comment (bug):
Cannot make a suggestion because github sucks.
```
		enqueueFailed := false
		for _, eventBatch := range missed {
			if !w.enqueueEvent(eventBatch) { // buffer overflow: watcher still lagging
				enqueueFailed = true
				break
			}
			nextRev = eventBatch[0].Kv.ModRevision + 1
		}

		if !enqueueFailed {
```

Code after:
	for w := range d.activeWatchers {
		w.Stop()
	}
	for w := range d.laggingWatchers {
		w.Stop()
	}
	d.activeWatchers = make(map[*watcher]int64)
	d.laggingWatchers = make(map[*watcher]int64)
}

func (d *demux) resyncLaggingWatchers() {
	d.mu.Lock()
	defer d.mu.Unlock()

	oldestRev := d.history.PeekOldest()
	for w, nextRev := range d.laggingWatchers {
		if oldestRev != 0 && nextRev < oldestRev {
			w.Stop()
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		missed := d.history.Filter(nextRev)

		enqueueFailed := false
		for _, eventBatch := range missed {
			if !w.enqueueEvent(eventBatch) { // buffer overflow: watcher still lagging
				enqueueFailed = true
				break
			}
			nextRev = eventBatch[0].Kv.ModRevision + 1
		}

		if !enqueueFailed {
			delete(d.laggingWatchers, w)
			d.activeWatchers[w] = nextRev
		} else {
			d.laggingWatchers[w] = nextRev
		}
	}
}

func (d *demux) PeekOldest() int64 {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return d.history.PeekOldest()
}
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
	d.laggingWatchers = make(map[*watcher]int64)
}

func (d *demux) resyncLaggingWatchers() {
	d.mu.Lock()
	defer d.mu.Unlock()

	for w, nextRev := range d.laggingWatchers {
		oldestRev := d.history.PeekOldest()

		if oldestRev != 0 && nextRev < oldestRev {
			w.Stop()
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		missed := d.history.Filter(AfterRev(nextRev))

		for _, eventBatch := range missed {
			if !w.enqueueEvent(eventBatch) { // buffer overflow: watcher still lagging
				break
			}
			nextRev = eventBatch[0].Kv.ModRevision + 1
		}

		if len(missed) > 0 && nextRev > missed[len(missed)-1][0].Kv.ModRevision {
			delete(d.laggingWatchers, w)
			d.activeWatchers[w] = nextRev
		} else {
			d.laggingWatchers[w] = nextRev
		}
	}
}

func (d *demux) PeekOldest() int64 {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return d.history.PeekOldest()
}


Review comment (bug):
Looks like I missed this. The `len(missed) > 0` check is incorrect, as watchers that are don't have any missing event will not be marked as synchronized. 

Code after:
			d.activeWatchers[w] = lastRev + 1
		}
	}
}

// Purge is called when etcd compaction invalidates our cached history, so clients should resubscribe.
func (d *demux) Purge() {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.history.RebaseHistory()
	for w := range d.activeWatchers {
		w.Stop()
	}
	for w := range d.laggingWatchers {
		w.Stop()
	}
	d.activeWatchers = make(map[*watcher]int64)
	d.laggingWatchers = make(map[*watcher]int64)
}

func (d *demux) resyncLaggingWatchers() {
	d.mu.Lock()
	defer d.mu.Unlock()

	oldestRev := d.history.PeekOldest()
	for w, nextRev := range d.laggingWatchers {
		if oldestRev != 0 && nextRev < oldestRev {
			w.Stop()
			delete(d.laggingWatchers, w)
			continue
		}
		// TODO: re-enable key‐predicate in Filter when non‐zero startRev or performance tuning is needed
		missed := d.history.Filter(nextRev)

		enqueueFailed := false
		for _, eventBatch := range missed {
			if !w.enqueueEvent(eventBatch) { // buffer overflow: watcher still lagging
				enqueueFailed = true
				break
			}
			nextRev = eventBatch[0].Kv.ModRevision + 1
		}

		if !enqueueFailed {
			delete(d.laggingWatchers, w)
			d.activeWatchers[w] = nextRev
		} else {
			d.laggingWatchers[w] = nextRev
		}
	}
}
Repository: etcd-io/etcd
File: cache/ringbuffer.go (Go)

Code before:
		r.size++
	}

	r.buffer[r.head] = b
	r.head = (r.head + 1) % len(r.buffer)
}

// Filter returns the batches that satisfy every predicate
// TODO: use binary search on the ring buffer to locate the first entry >= nextRev instead of a full scan
func (r *ringBuffer) Filter(batchPred BatchPredicate) (eventBatches [][]*clientv3.Event) {

	eventBatches = make([][]*clientv3.Event, 0, r.size)

	for n, i := 0, r.tail; n < r.size; n, i = n+1, (i+1)%len(r.buffer) {
		batch := r.buffer[i]
		if batch == nil {
			panic(fmt.Sprintf("ringBuffer.Filter: unexpected nil eventBatch at index %d", i))
		}
		if batchPred == nil || batchPred(batch) {
			eventBatches = append(eventBatches, batch.events)
		}
	}
	return eventBatches
}

// PeekLatest returns the most recently-appended batch (or nil if empty).
func (r *ringBuffer) PeekLatest() int64 {
	if r.size == 0 {
		return 0
	}
	idx := (r.head - 1 + len(r.buffer)) % len(r.buffer)
	return r.buffer[idx].rev
}

// PeekOldest returns the oldest batch currently stored (or nil if empty).
func (r *ringBuffer) PeekOldest() int64 {
	if r.size == 0 {
		return 0
	}
	return r.buffer[r.tail].rev
}

func (r *ringBuffer) RebaseHistory() {
	r.head, r.tail, r.size = 0, 0, 0
	for i := range r.buffer {
		r.buffer[i] = nil
	}
}


Review comment (suggestion):
```suggestion
// PeekLatest returns the most recently-appended event (or nil if empty).
```

Code after:
			rev:    events[start].Kv.ModRevision,
			events: events[start:],
		})
	}
}

func (r *ringBuffer) append(b batch) {
	if len(b.events) == 0 {
		return
	}
	if r.size == len(r.buffer) {
		r.tail = (r.tail + 1) % len(r.buffer)
	} else {
		r.size++
	}
	r.buffer[r.head] = b
	r.head = (r.head + 1) % len(r.buffer)
}

// Filter returns all events in the buffer whose ModRevision is >= minRev.
// TODO: use binary search on the ring buffer to locate the first entry >= nextRev instead of a full scan
func (r *ringBuffer) Filter(minRev int64) (eventBatches [][]*clientv3.Event) {
	eventBatches = make([][]*clientv3.Event, 0, r.size)

	for n, i := 0, r.tail; n < r.size; n, i = n+1, (i+1)%len(r.buffer) {
		eventBatch := r.buffer[i]
		if eventBatch.events == nil {
			panic(fmt.Sprintf("ringBuffer.Filter: unexpected nil eventBatch at index %d", i))
		}
		if eventBatch.rev >= minRev {
			eventBatches = append(eventBatches, eventBatch.events)
		}
	}
	return eventBatches
}

// PeekLatest returns the most recently-appended event (or nil if empty).
func (r *ringBuffer) PeekLatest() int64 {
	if r.size == 0 {
		return 0
	}
	idx := (r.head - 1 + len(r.buffer)) % len(r.buffer)
	return r.buffer[idx].rev
}

// PeekOldest returns the oldest event currently stored (or nil if empty).
func (r *ringBuffer) PeekOldest() int64 {
	if r.size == 0 {
		return 0
	}
	return r.buffer[r.tail].rev
Repository: etcd-io/etcd
File: cache/ringbuffer.go (Go)

Code before:
func (r *ringBuffer) Filter(batchPred BatchPredicate) (eventBatches [][]*clientv3.Event) {

	eventBatches = make([][]*clientv3.Event, 0, r.size)

	for n, i := 0, r.tail; n < r.size; n, i = n+1, (i+1)%len(r.buffer) {
		batch := r.buffer[i]
		if batch == nil {
			panic(fmt.Sprintf("ringBuffer.Filter: unexpected nil eventBatch at index %d", i))
		}
		if batchPred == nil || batchPred(batch) {
			eventBatches = append(eventBatches, batch.events)
		}
	}
	return eventBatches
}

// PeekLatest returns the most recently-appended batch (or nil if empty).
func (r *ringBuffer) PeekLatest() int64 {
	if r.size == 0 {
		return 0
	}
	idx := (r.head - 1 + len(r.buffer)) % len(r.buffer)
	return r.buffer[idx].rev
}

// PeekOldest returns the oldest batch currently stored (or nil if empty).
func (r *ringBuffer) PeekOldest() int64 {
	if r.size == 0 {
		return 0
	}
	return r.buffer[r.tail].rev
}

func (r *ringBuffer) RebaseHistory() {
	r.head, r.tail, r.size = 0, 0, 0
	for i := range r.buffer {
		r.buffer[i] = nil
	}
}


Review comment (suggestion):
```suggestion
// PeekOldest returns the oldest event currently stored (or nil if empty).
```

Code after:
	}
	if r.size == len(r.buffer) {
		r.tail = (r.tail + 1) % len(r.buffer)
	} else {
		r.size++
	}
	r.buffer[r.head] = b
	r.head = (r.head + 1) % len(r.buffer)
}

// Filter returns all events in the buffer whose ModRevision is >= minRev.
// TODO: use binary search on the ring buffer to locate the first entry >= nextRev instead of a full scan
func (r *ringBuffer) Filter(minRev int64) (eventBatches [][]*clientv3.Event) {
	eventBatches = make([][]*clientv3.Event, 0, r.size)

	for n, i := 0, r.tail; n < r.size; n, i = n+1, (i+1)%len(r.buffer) {
		eventBatch := r.buffer[i]
		if eventBatch.events == nil {
			panic(fmt.Sprintf("ringBuffer.Filter: unexpected nil eventBatch at index %d", i))
		}
		if eventBatch.rev >= minRev {
			eventBatches = append(eventBatches, eventBatch.events)
		}
	}
	return eventBatches
}

// PeekLatest returns the most recently-appended event (or nil if empty).
func (r *ringBuffer) PeekLatest() int64 {
	if r.size == 0 {
		return 0
	}
	idx := (r.head - 1 + len(r.buffer)) % len(r.buffer)
	return r.buffer[idx].rev
}

// PeekOldest returns the oldest event currently stored (or nil if empty).
func (r *ringBuffer) PeekOldest() int64 {
	if r.size == 0 {
		return 0
	}
	return r.buffer[r.tail].rev
}

func (r *ringBuffer) RebaseHistory() {
	r.head, r.tail, r.size = 0, 0, 0
	for i := range r.buffer {
		r.buffer[i] = batch{}
	}
}
Repository: etcd-io/etcd
File: cache/watcher.go (Go)

Code before:
}

// true  -> events delivered (or filtered/duplicate)
// false -> buffer full (caller should mark watcher “lagging”)
func (w *watcher) enqueueEvent(eventBatch []*clientv3.Event) bool {
	if w.keyPred != nil {
		filtered := make([]*clientv3.Event, 0, len(eventBatch))
		for _, event := range eventBatch {
			if w.keyPred(event.Kv.Key) {
				filtered = append(filtered, event)
			}
		}
		if len(filtered) == 0 {
			return true
		}
		eventBatch = filtered
	}
	select {
	case w.eventQueue <- eventBatch:
		return true
	default:
		return false
	}
}

// Stop closes the batch channel atomically.
func (w *watcher) Stop() {
	if atomic.CompareAndSwapInt32(&w.stopped, 0, 1) {
		close(w.eventQueue)
		close(w.done)
	}
}

func (w *watcher) Done() <-chan struct{} { return w.done }


Review comment (suggestion):
```suggestion
// Stop closes the event channel atomically.
```

Code after:
}

// true  -> events delivered (or filtered/duplicate)
// false -> buffer full (caller should mark watcher “lagging”)
func (w *watcher) enqueueEvent(eventBatch []*clientv3.Event) bool {
	if w.keyPred != nil {
		filtered := make([]*clientv3.Event, 0, len(eventBatch))
		for _, event := range eventBatch {
			if w.keyPred(event.Kv.Key) {
				filtered = append(filtered, event)
			}
		}
		if len(filtered) == 0 {
			return true
		}
		eventBatch = filtered
	}
	select {
	case w.eventQueue <- eventBatch:
		return true
	default:
		return false
	}
}

// Stop closes the event channel atomically.
func (w *watcher) Stop() {
	if atomic.CompareAndSwapInt32(&w.stopped, 0, 1) {
		close(w.eventQueue)
		close(w.done)
	}
}

func (w *watcher) Done() <-chan struct{} { return w.done }
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:
func readWatchChannel(
	watchChan clientv3.WatchChan,
	cache *Cache,
	demux *demux,
	onFirstResponse func(),
) error {
	for resp := range watchChan {
		onFirstResponse()

		if err := resp.Err(); err != nil {
			if errors.Is(err, rpctypes.ErrCompacted) {
				select {
				case <-cache.ready:
					// TODO: Reinitialize cache.ready safely; current direct channel assignment can race with concurrent watchers
					cache.ready = make(chan struct{})
				default:
				}
				demux.Purge()
			}
			return err
		}

		var (
			eventBatch []*clientv3.Event
			batchRev   int64
		)
		for _, event := range resp.Events {
			rev := event.Kv.ModRevision
			if len(eventBatch) == 0 {
				batchRev = rev
			}
			if rev != batchRev {
				demux.Broadcast(eventBatch)
				eventBatch = eventBatch[:0]
				batchRev = rev
			}
			eventBatch = append(eventBatch, event)
		}

		if len(eventBatch) > 0 {
			demux.Broadcast(eventBatch)
		}
	}
	return nil
}


Review comment (performance):
```suggestion
		if len(resp.Events) == 0 {
			continue
		}
		var (
			eventBatch []*clientv3.Event
			batchRev   int64 = resp.Events[0].Kv.ModRevision
		)
```

Code after:
func readWatchChannel(
	watchChan clientv3.WatchChan,
	cache *Cache,
	demux *demux,
	onFirstResponse func(),
) error {
	for resp := range watchChan {
		onFirstResponse()

		if err := resp.Err(); err != nil {
			if errors.Is(err, rpctypes.ErrCompacted) {
				select {
				case <-cache.ready:
					// TODO: Reinitialize cache.ready safely; current direct channel assignment can race with concurrent watchers
					cache.ready = make(chan struct{})
				default:
				}
				demux.Purge()
			}
			return err
		}
		demux.Broadcast(resp.Events)
	}
	return nil
}
Repository: etcd-io/etcd
File: tests/integration/cache_test.go (Go)

Code before:
package integration

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/google/go-cmp/cmp"
	mvccpb "go.etcd.io/etcd/api/v3/mvccpb"
	. "go.etcd.io/etcd/cache"
	"go.etcd.io/etcd/client/pkg/v3/types"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/server/v3/embed"
)

func TestWatchWithCache(t *testing.T) {
	client := startEtcd(t)
	cache, err := New(client, "/", WithHistoryWindowSize(32))
	if err != nil {
		t.Fatalf("New(...): %v", err)
	}
	t.Cleanup(cache.Close)
	if err := cache.WaitReady(t.Context()); err != nil {
		t.Fatalf("cache not ready: %v", err)
	}
	testWatch(t, client.KV, cache)
}

func TestWatchWithoutCache(t *testing.T) {
	client := startEtcd(t)
	testWatch(t, client.KV, client.Watcher)
}

func testWatch(t *testing.T, kv clientv3.KV, watcher Watcher) {
	ctx := t.Context()
	event1Put := &clientv3.Event{
		Type: clientv3.EventTypePut,
		Kv: &mvccpb.KeyValue{
			Key:            []byte("/a"),

Review comment (suggestion):
In integration tests we use different package to setup etcd. 

Please remove functions for setting up etcd and for each tests use the following code:
```
import "go.etcd.io/etcd/tests/v3/framework/integration"

func TestX(*testing.T) {
	integration.BeforeTest(t)

	clus := integration.NewCluster(t, &integration2.ClusterConfig{Size: 1})
	defer clus.Terminate(t)

	cc, err := clus.ClusterClient(t)
	require.NoError(t, err)
	...
}
```

Code after:
// Copyright 2025 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package integration

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/google/go-cmp/cmp"

	"go.etcd.io/etcd/api/v3/mvccpb"
	cache "go.etcd.io/etcd/cache/v3"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/framework/integration"
)

func TestCacheWatch(t *testing.T) {
	integration.BeforeTest(t)
	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
	t.Cleanup(func() { clus.Terminate(t) })
	client := clus.Client(0)

	c, err := cache.New(client, "/", cache.WithHistoryWindowSize(32))
	if err != nil {
		t.Fatalf("New(...): %v", err)
	}
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
		delete(d.activeWatchers, w)
		delete(d.laggingWatchers, w)
	}()
	w.Stop()
}

func (d *demux) broadcast(eventBatch []*clientv3.Event) {
	d.mu.Lock()
	defer d.mu.Unlock()

	rev := eventBatch[0].Kv.ModRevision
	d.history.Append(eventBatch)
	for w, nextRev := range d.activeWatchers {
		if rev < nextRev {
			continue
		}
		if !w.enqueueEvent(eventBatch) { // buffer overflow
			d.laggingWatchers[w] = nextRev
			delete(d.activeWatchers, w)
		} else {
			d.activeWatchers[w] = rev + 1
		}
	}
}

func (d *demux) Broadcast(events []*clientv3.Event) {
	batchRev := events[0].Kv.ModRevision
	eventBatch := make([]*clientv3.Event, 0, len(events))

	for _, event := range events {
		if event.Kv.ModRevision != batchRev {
			d.broadcast(eventBatch)
			eventBatch = nil
			batchRev = event.Kv.ModRevision
		}
		eventBatch = append(eventBatch, event)
	}
	d.broadcast(eventBatch)
}

// Purge is called when etcd compaction invalidates our cached history, so clients should resubscribe.
func (d *demux) Purge() {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.history.RebaseHistory()
	for w := range d.activeWatchers {
		w.Stop()
	}
	for w := range d.laggingWatchers {
		w.Stop()
	}

Review comment (performance):
There is nothing in the broadcast code itself that prevents sending whole batch of events at once.
The only part that requires splitting by revision is the `history.Append`.

Can you move the revision batching logic to the `history.Append` instead?

Code after:
	func() {
		d.mu.Lock()
		defer d.mu.Unlock()
		delete(d.activeWatchers, w)
		delete(d.laggingWatchers, w)
	}()
	w.Stop()
}

func (d *demux) Broadcast(events []*clientv3.Event) {
	if len(events) == 0 {
		return
	}

	d.mu.Lock()
	defer d.mu.Unlock()

	d.history.Append(events)

	lastRev := events[len(events)-1].Kv.ModRevision
	for w, nextRev := range d.activeWatchers {
		start := len(events)
		for i, ev := range events {
			if ev.Kv.ModRevision >= nextRev {
				start = i
				break
			}
		}

		if start == len(events) {
			continue
		}

		if !w.enqueueEvent(events[start:]) { // overflow → lagging
			d.laggingWatchers[w] = nextRev
			delete(d.activeWatchers, w)
		} else {
			d.activeWatchers[w] = lastRev + 1
		}
	}
}

// Purge is called when etcd compaction invalidates our cached history, so clients should resubscribe.
func (d *demux) Purge() {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.history.RebaseHistory()
	for w := range d.activeWatchers {
		w.Stop()
	}
	for w := range d.laggingWatchers {