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

Code before:
		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()
		readyOnce := sync.Once{}

		watchCtx := &watchCtx{
			cache:           cache,
			backoffStart:    cfg.InitialBackoff,
			backoffMax:      cfg.MaxBackoff,
			onFirstResponse: func() { readyOnce.Do(func() { close(cache.ready) }) },
		}
		serveWatchEvents(internalCtx, watchCtx)
	}()

	return cache, nil
}

// Watch registers a cache-backed watcher for a given key or prefix.
// It returns a WatchChan that streams WatchResponses containing batches.
func (c *Cache) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
	select {
	case <-c.ready:
	case <-ctx.Done():
		emptyWatchChan := make(chan clientv3.WatchResponse)
		close(emptyWatchChan)
		return emptyWatchChan
	}

	op := clientv3.OpGet(key, opts...)
	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) {
			var compactRev int64
			if oldestBatch := c.demux.PeekOldest(); oldestBatch != nil {
				compactRev = oldestBatch.rev
			}
			responseChan <- clientv3.WatchResponse{CompactRevision: compactRev}
		} else {
			responseChan <- clientv3.WatchResponse{Canceled: true}

Review comment (suggestion):
```suggestion
// It returns a WatchChan that streams WatchResponses containing events.
```

Code after:

	cache.waitGroup.Add(1)
	go func() {
		defer cache.waitGroup.Done()
		readyOnce := sync.Once{}

		watchCtx := &watchCtx{
			cache:           cache,
			backoffStart:    cfg.InitialBackoff,
			backoffMax:      cfg.MaxBackoff,
			onFirstResponse: func() { readyOnce.Do(func() { close(cache.ready) }) },
		}
		serveWatchEvents(internalCtx, watchCtx)
	}()

	return cache, nil
}

// Watch registers a cache-backed watcher for a given key or prefix.
// It returns a WatchChan that streams WatchResponses containing events.
func (c *Cache) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
	select {
	case <-c.ready:
	case <-ctx.Done():
		emptyWatchChan := make(chan clientv3.WatchResponse)
		close(emptyWatchChan)
		return emptyWatchChan
	}

	op := clientv3.OpGet(key, opts...)
	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)