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

Code before:
	if startingRev <= latestRev {
		d.laggingWatchers[w] = startingRev
	} else {
		d.activeWatchers[w] = startingRev
	}
}

func (d *demux) Unregister(w *watcher) {
	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.AppendMixed(events)
	for i := 0; i < len(events); {
		rev := events[i].Kv.ModRevision
		j := i + 1
		for j < len(events) && events[j].Kv.ModRevision == rev {
			j++
		}
		batch := events[i:j]

		for w, nextRev := range d.activeWatchers {
			if rev < nextRev {
				continue
			}
			if !w.enqueueEvent(batch) { // buffer overflow
				d.laggingWatchers[w] = nextRev
				delete(d.activeWatchers, w)
			} else {
				d.activeWatchers[w] = rev + 1
			}
		}
		i = j
	}
}

// Purge is called when etcd compaction invalidates our cached history, so clients should resubscribe.
func (d *demux) Purge() {

Review comment (suggestion):
No-one asked for rename.
```suggestion
	d.history.Append(events)
```

Code after:

func (d *demux) Unregister(w *watcher) {
	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()
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
		d.laggingWatchers[w] = startingRev
	} else {
		d.activeWatchers[w] = startingRev
	}
}

func (d *demux) Unregister(w *watcher) {
	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.AppendMixed(events)
	for i := 0; i < len(events); {
		rev := events[i].Kv.ModRevision
		j := i + 1
		for j < len(events) && events[j].Kv.ModRevision == rev {
			j++
		}
		batch := events[i:j]

		for w, nextRev := range d.activeWatchers {
			if rev < nextRev {
				continue
			}
			if !w.enqueueEvent(batch) { // buffer overflow
				d.laggingWatchers[w] = nextRev
				delete(d.activeWatchers, w)
			} else {
				d.activeWatchers[w] = rev + 1
			}
		}
		i = j
	}
}

// Purge is called when etcd compaction invalidates our cached history, so clients should resubscribe.
func (d *demux) Purge() {
	d.mu.Lock()

Review comment (suggestion):
I don't understand this code at all.

Please take a look at the code below.
```
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 !w.enqueueEvent(events[start:]) {
			d.laggingWatchers[w] = nextRev
			delete(d.activeWatchers, w)
		} else {
			d.activeWatchers[w] = lastRev + 1
		}
	}
```


Code after:
func (d *demux) Unregister(w *watcher) {
	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()
	}
Repository: etcd-io/etcd
File: cache/ringbuffer.go (Go)

Code before:

	clientv3 "go.etcd.io/etcd/client/v3"
)

type ringBuffer struct {
	buffer []*batch
	// head is the index immediately after the last non-empty entry in the buffer (i.e., the next write position).
	head, tail, size int
}

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

type KeyPredicate = func([]byte) bool

func newRingBuffer(capacity int) *ringBuffer {
	// assume capacity > 0 – validated by Cache
	return &ringBuffer{
		buffer: make([]*batch, capacity),
	}
}

func (r *ringBuffer) Append(eventBatch []*clientv3.Event) {
	if len(eventBatch) == 0 {
		panic("ringBuffer.Append: empty group")
	}
	rev := eventBatch[0].Kv.ModRevision
	for i, ev := range eventBatch[1:] {
		if ev.Kv.ModRevision != rev {
			panic(fmt.Sprintf("ringBuffer.Append: mixed revisions at idx %d (%d vs %d)", i+1, ev.Kv.ModRevision, rev))
		}
	}
	b := &batch{rev: rev, events: eventBatch}

	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)
}

func (r *ringBuffer) AppendMixed(events []*clientv3.Event) {
	for i := 0; i < len(events); {
		rev := events[i].Kv.ModRevision
		j := i + 1
		for j < len(events) && events[j].Kv.ModRevision == rev {

Review comment (performance):
```suggestion
func (r *ringBuffer) append(eventBatch []*clientv3.Event) {
```

Code after:

	clientv3 "go.etcd.io/etcd/client/v3"
)

type ringBuffer struct {
	buffer []batch
	// head is the index immediately after the last non-empty entry in the buffer (i.e., the next write position).
	head, tail, size int
}

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

type KeyPredicate = func([]byte) bool

func newRingBuffer(capacity int) *ringBuffer {
	// assume capacity > 0 – validated by Cache
	return &ringBuffer{
		buffer: make([]batch, capacity),
	}
}

func (r *ringBuffer) Append(events []*clientv3.Event) {
	start := 0
	for end := 1; end < len(events); end++ {
		if events[end].Kv.ModRevision != events[start].Kv.ModRevision {
			r.append(batch{
				rev:    events[start].Kv.ModRevision,
				events: events[start:end],
			})
			start = end
		}
	}
	if start < len(events) {
		r.append(batch{
			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 {
Repository: etcd-io/etcd
File: cache/ringbuffer.go (Go)

Code before:
		buffer: make([]*batch, capacity),
	}
}

func (r *ringBuffer) Append(eventBatch []*clientv3.Event) {
	if len(eventBatch) == 0 {
		panic("ringBuffer.Append: empty group")
	}
	rev := eventBatch[0].Kv.ModRevision
	for i, ev := range eventBatch[1:] {
		if ev.Kv.ModRevision != rev {
			panic(fmt.Sprintf("ringBuffer.Append: mixed revisions at idx %d (%d vs %d)", i+1, ev.Kv.ModRevision, rev))
		}
	}
	b := &batch{rev: rev, events: eventBatch}

	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)
}

func (r *ringBuffer) AppendMixed(events []*clientv3.Event) {
	for i := 0; i < len(events); {
		rev := events[i].Kv.ModRevision
		j := i + 1
		for j < len(events) && events[j].Kv.ModRevision == rev {
			j++
		}
		r.Append(events[i:j])
		i = j
	}
}

// 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) {
		batch := r.buffer[i]
		if batch == nil {
			panic(fmt.Sprintf("ringBuffer.Filter: unexpected nil eventBatch at index %d", i))
		}
		if batch.rev >= minRev {
			eventBatches = append(eventBatches, batch.events)
		}
	}

Review comment (suggestion):
There is no `AppendMixed` function standard for data structures. Just keep the external function name `Append` and hide internal function by using lower case.
```suggestion
func (r *ringBuffer) Append(events []*clientv3.Event) {
```

Code after:
		buffer: make([]batch, capacity),
	}
}

func (r *ringBuffer) Append(events []*clientv3.Event) {
	start := 0
	for end := 1; end < len(events); end++ {
		if events[end].Kv.ModRevision != events[start].Kv.ModRevision {
			r.append(batch{
				rev:    events[start].Kv.ModRevision,
				events: events[start:end],
			})
			start = end
		}
	}
	if start < len(events) {
		r.append(batch{
			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
Repository: etcd-io/etcd
File: cache/ringbuffer.go (Go)

Code before:
		buffer: make([]*batch, capacity),
	}
}

func (r *ringBuffer) Append(eventBatch []*clientv3.Event) {
	if len(eventBatch) == 0 {
		panic("ringBuffer.Append: empty group")
	}
	rev := eventBatch[0].Kv.ModRevision
	for i, ev := range eventBatch[1:] {
		if ev.Kv.ModRevision != rev {
			panic(fmt.Sprintf("ringBuffer.Append: mixed revisions at idx %d (%d vs %d)", i+1, ev.Kv.ModRevision, rev))
		}
	}
	b := &batch{rev: rev, events: eventBatch}

	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)
}

func (r *ringBuffer) AppendMixed(events []*clientv3.Event) {
	for i := 0; i < len(events); {
		rev := events[i].Kv.ModRevision
		j := i + 1
		for j < len(events) && events[j].Kv.ModRevision == rev {
			j++
		}
		r.Append(events[i:j])
		i = j
	}
}

// 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) {
		batch := r.buffer[i]
		if batch == nil {
			panic(fmt.Sprintf("ringBuffer.Filter: unexpected nil eventBatch at index %d", i))
		}
		if batch.rev >= minRev {
			eventBatches = append(eventBatches, batch.events)
		}
	}

Review comment (suggestion):
Again with the order of functions, the important functions at the top. The low level functions at the bottom. Order of functions should match importance, from top to bottom.

Code after:
		buffer: make([]batch, capacity),
	}
}

func (r *ringBuffer) Append(events []*clientv3.Event) {
	start := 0
	for end := 1; end < len(events); end++ {
		if events[end].Kv.ModRevision != events[start].Kv.ModRevision {
			r.append(batch{
				rev:    events[start].Kv.ModRevision,
				events: events[start:end],
			})
			start = end
		}
	}
	if start < len(events) {
		r.append(batch{
			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
Repository: etcd-io/etcd
File: tests/integration/cache_test.go (Go)

Code before:
			resp, ok := <-watchCh

			if !ok || !resp.Canceled {
				t.Errorf("expected canceled response, got %#v (closed=%v)", resp, !ok)
			}
		})
	}
}

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

func generateEvents(t *testing.T, client *clientv3.Client, prefix string, n int) {
	t.Helper()
	for i := 0; i < n; i++ {
		key := fmt.Sprintf("%s%d", prefix, i)
		if _, err := client.Put(t.Context(), key, fmt.Sprintf("%d", i)); err != nil {
			t.Fatalf("Put(%q): %v", key, err)
		}
	}
}

type Watcher interface {
	Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan
}

func readEvents(watch clientv3.WatchChan) (events []*clientv3.Event, ok bool) {
	deadline := time.After(200 * time.Millisecond)
	for {
		select {
		case resp, ok := <-watch:
			if !ok {
				return events, false
			}
			events = append(events, resp.Events...)
		case <-deadline:
			return events, true
		}
	}
}


Review comment (refactor):
Don't over refactor. Having dedicated functions just for 3 tests just makes it harder to follow versus if you just copied those 4 lines.

Code after:
			key:  "",
			opts: []clientv3.OpOption{clientv3.WithPrefix(), clientv3.WithRev(123)},
		},
		{
			name: "subprefix_watch",
			key:  "foo/",
			opts: []clientv3.OpOption{clientv3.WithPrefix()},
		},
		{
			name: "single_key_watch",
			key:  "foo/bar",
			opts: nil, // exact-key watch (no WithPrefix)
		},
	}

	for _, tt := range tests {
		tt := tt
		t.Run(tt.name, func(t *testing.T) {
			watchCh := c.Watch(ctx, tt.key, tt.opts...)
			resp, ok := <-watchCh

			if !ok || !resp.Canceled {
				t.Errorf("expected canceled response, got %#v (closed=%v)", resp, !ok)
			}
		})
	}
}

func generateEvents(t *testing.T, client *clientv3.Client, prefix string, n int) {
	t.Helper()
	for i := 0; i < n; i++ {
		key := fmt.Sprintf("%s%d", prefix, i)
		if _, err := client.Put(t.Context(), key, fmt.Sprintf("%d", i)); err != nil {
			t.Fatalf("Put(%q): %v", key, err)
		}
	}
}

type Watcher interface {
	Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan
}

func readEvents(watch clientv3.WatchChan) (events []*clientv3.Event, ok bool) {
	deadline := time.After(time.Second)
	for {
		select {
		case resp, ok := <-watch:
			if !ok {
				return events, false
			}
			events = append(events, resp.Events...)
Repository: etcd-io/etcd
File: cache/ringbuffer.go (Go)

Code before:

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

type KeyPredicate = func([]byte) bool

func newRingBuffer(capacity int) *ringBuffer {
	// assume capacity > 0 – validated by Cache
	return &ringBuffer{
		buffer: make([]*batch, capacity),
	}
}

func (r *ringBuffer) Append(events []*clientv3.Event) {
	start := 0
	for end := 1; end < len(events); end++ {
		if events[end].Kv.ModRevision != events[start].Kv.ModRevision {
			r.append(events[start:end])
			start = end
		}
	}
	if start < len(events) {
		r.append(events[start:])
	}
}

func (r *ringBuffer) append(eventBatch []*clientv3.Event) {
	if len(eventBatch) == 0 {
		return
	}
	rev := eventBatch[0].Kv.ModRevision
	for _, ev := range eventBatch[1:] {
		if ev.Kv.ModRevision != rev {
			return
		}
	}
	b := &batch{rev: rev, events: eventBatch}

	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.

Review comment (performance):
As `append` is an internal function we can use batch now.
```suggestion
func (r *ringBuffer) Append(events []*clientv3.Event) {
	start := 0
	for end := 1; end < len(events); end++ {
		if events[end].Kv.ModRevision != events[start].Kv.ModRevision {
			r.append(batch{rev: events[start].Kv.ModRevision, events: events[start:end]})
			start = end
		}
	}
	if start < len(events) {
		r.append(batch{rev: events[start].Kv.ModRevision, events: events[start:]})
```

Code after:

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

type KeyPredicate = func([]byte) bool

func newRingBuffer(capacity int) *ringBuffer {
	// assume capacity > 0 – validated by Cache
	return &ringBuffer{
		buffer: make([]batch, capacity),
	}
}

func (r *ringBuffer) Append(events []*clientv3.Event) {
	start := 0
	for end := 1; end < len(events); end++ {
		if events[end].Kv.ModRevision != events[start].Kv.ModRevision {
			r.append(batch{
				rev:    events[start].Kv.ModRevision,
				events: events[start:end],
			})
			start = end
		}
	}
	if start < len(events) {
		r.append(batch{
			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) {
Repository: etcd-io/etcd
File: cache/cache.go (Go)

Code before:

// readWatchChannel reads an etcd Watch stream into History and enqueueCh, returning nil on cancel or first error.
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
		}
		if len(resp.Events) == 0 {
			continue
		}
		demux.Broadcast(resp.Events)
	}
	return nil
}


Review comment (suggestion):
```suggestion
```

Not needed, Broadcast already handles it. 

Code after:

// readWatchChannel reads an etcd Watch stream into History and enqueueCh, returning nil on cancel or first error.
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:
		{
			name: "subprefix_watch",
			key:  "foo/",
			opts: []clientv3.OpOption{clientv3.WithPrefix()},
		},
		{
			name: "single_key_watch",
			key:  "foo/bar",
			opts: nil, // exact-key watch (no WithPrefix)
		},
	}

	for _, tt := range tests {
		tt := tt
		t.Run(tt.name, func(t *testing.T) {
			watchCh := c.Watch(ctx, tt.key, tt.opts...)
			resp, ok := <-watchCh

			if !ok || !resp.Canceled {
				t.Errorf("expected canceled response, got %#v (closed=%v)", resp, !ok)
			}
		})
	}
}

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

func generateEvents(t *testing.T, client *clientv3.Client, prefix string, n int) {
	t.Helper()
	for i := 0; i < n; i++ {
		key := fmt.Sprintf("%s%d", prefix, i)
		if _, err := client.Put(t.Context(), key, fmt.Sprintf("%d", i)); err != nil {
			t.Fatalf("Put(%q): %v", key, err)
		}
	}
}

type Watcher interface {
	Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan
}

func readEvents(watch clientv3.WatchChan) (events []*clientv3.Event, ok bool) {
	deadline := time.After(200 * time.Millisecond)
	for {
		select {
		case resp, ok := <-watch:

Review comment (refactor):
Don't over refactor. Having dedicated functions just for 3 tests just makes it harder to follow versus if you just copied those 4 lines.

Code after:
	c, err := cache.New(client, "")
	if err != nil {
		t.Fatalf("New(...): %v", err)
	}
	t.Cleanup(c.Close)
	if err := c.WaitReady(ctx); err != nil {
		t.Fatal(err)
	}

	tests := []struct {
		name string
		key  string
		opts []clientv3.OpOption
	}{
		{
			name: "non_zero_start_revision",
			key:  "",
			opts: []clientv3.OpOption{clientv3.WithPrefix(), clientv3.WithRev(123)},
		},
		{
			name: "subprefix_watch",
			key:  "foo/",
			opts: []clientv3.OpOption{clientv3.WithPrefix()},
		},
		{
			name: "single_key_watch",
			key:  "foo/bar",
			opts: nil, // exact-key watch (no WithPrefix)
		},
	}

	for _, tt := range tests {
		tt := tt
		t.Run(tt.name, func(t *testing.T) {
			watchCh := c.Watch(ctx, tt.key, tt.opts...)
			resp, ok := <-watchCh

			if !ok || !resp.Canceled {
				t.Errorf("expected canceled response, got %#v (closed=%v)", resp, !ok)
			}
		})
	}
}

func generateEvents(t *testing.T, client *clientv3.Client, prefix string, n int) {
	t.Helper()
	for i := 0; i < n; i++ {
		key := fmt.Sprintf("%s%d", prefix, i)
		if _, err := client.Put(t.Context(), key, fmt.Sprintf("%d", i)); err != nil {
			t.Fatalf("Put(%q): %v", key, err)
		}