Repository: etcd-io/etcd
File: tests/robustness/client/watch.go (Go)

Code before:
	"go.etcd.io/etcd/tests/v3/robustness/identity"
	"go.etcd.io/etcd/tests/v3/robustness/report"
)

func CollectClusterWatchEvents(ctx context.Context, lg *zap.Logger, etcdEndpoints [][]string, maxRevisionChan <-chan int64, cfg WatchConfig, baseTime time.Time, ids identity.Provider) ([]report.ClientReport, bool, error) {
	mux := sync.Mutex{}
	var wg sync.WaitGroup
	neverFailed := true
	reports := make([]report.ClientReport, len(etcdEndpoints))
	memberMaxRevisionChans := make([]chan int64, len(etcdEndpoints))
	for i, endpoint := range etcdEndpoints {
		c, err := NewRecordingClient(endpoint, ids, baseTime)
		if err != nil {
			return nil, true, err
		}
		memberMaxRevisionChan := make(chan int64, 1)
		memberMaxRevisionChans[i] = memberMaxRevisionChan
		wg.Add(1)
		go func(i int, c *RecordingClient) {
			defer wg.Done()
			defer c.Close()
			watchSuccessful := watchUntilRevision(ctx, lg, c, memberMaxRevisionChan, cfg)
			mux.Lock()
			if !watchSuccessful && neverFailed {
				neverFailed = false
			}
			reports[i] = c.Report()
			mux.Unlock()
		}(i, c)
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		maxRevision := <-maxRevisionChan
		for _, memberChan := range memberMaxRevisionChans {
			memberChan <- maxRevision
		}
	}()
	wg.Wait()
	return reports, neverFailed, nil
}

type WatchConfig struct {
	RequestProgress bool
}

// watchUntilRevision watches all changes until context is cancelled, it has observed revision provided via maxRevisionChan or maxRevisionChan was closed.
func watchUntilRevision(ctx context.Context, lg *zap.Logger, c *RecordingClient, maxRevisionChan <-chan int64, cfg WatchConfig) bool {
	var maxRevision int64
	var lastRevision int64 = 1
	var success = true

Review comment (bug):
```suggestion
			neverFailed &= watchSuccessful
```

Code after:
	"go.etcd.io/etcd/tests/v3/robustness/identity"
	"go.etcd.io/etcd/tests/v3/robustness/report"
)

func CollectClusterWatchEvents(ctx context.Context, lg *zap.Logger, endpoints []string, maxRevisionChan <-chan int64, cfg WatchConfig, baseTime time.Time, ids identity.Provider) ([]report.ClientReport, bool, error) {
	mux := sync.Mutex{}
	var wg sync.WaitGroup
	neverFailed := true
	reports := make([]report.ClientReport, len(endpoints))
	memberMaxRevisionChans := make([]chan int64, len(endpoints))
	for i, endpoint := range endpoints {
		c, err := NewRecordingClient([]string{endpoint}, ids, baseTime)
		if err != nil {
			return nil, false, err
		}
		memberMaxRevisionChan := make(chan int64, 1)
		memberMaxRevisionChans[i] = memberMaxRevisionChan
		wg.Add(1)
		go func(i int, c *RecordingClient) {
			defer wg.Done()
			defer c.Close()
			watchSuccessful := watchUntilRevision(ctx, lg, c, memberMaxRevisionChan, cfg)
			mux.Lock()
			neverFailed = neverFailed && watchSuccessful
			reports[i] = c.Report()
			mux.Unlock()
		}(i, c)
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		maxRevision := <-maxRevisionChan
		for _, memberChan := range memberMaxRevisionChans {
			memberChan <- maxRevision
		}
	}()
	wg.Wait()
	return reports, neverFailed, nil
}

type WatchConfig struct {
	RequestProgress bool
}

// watchUntilRevision watches all changes until context is cancelled, it has observed revision provided via maxRevisionChan or maxRevisionChan was closed.
func watchUntilRevision(ctx context.Context, lg *zap.Logger, c *RecordingClient, maxRevisionChan <-chan int64, cfg WatchConfig) bool {
	var maxRevision int64
	var lastRevision int64 = 1
	var success = true
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()