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

Code before:
	return reports, everFailed, 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 failed = false
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
resetWatch:
	for {
		watch := c.Watch(ctx, "", lastRevision+1, true, true, false)
		for {
			select {
			case <-ctx.Done():
				if maxRevision == 0 {
					lg.Error("Client didn't collect all events, max revision not set")
					failed = true
				}
				if lastRevision < maxRevision {
					lg.Sugar().Errorf("Client didn't collect all events, revision got %d, expected: %d", lastRevision, maxRevision)
					failed = true
				}
				return failed
			case revision, ok := <-maxRevisionChan:
				if ok {
					maxRevision = revision
					if lastRevision >= maxRevision {
						cancel()
					}
				} else {
					// Only cancel if maxRevision was never set.
					if maxRevision == 0 {
						cancel()
					}
				}
			case resp, ok := <-watch:
				if !ok {
					lg.Info("Watch channel closed")
					continue resetWatch
				}
				if cfg.RequestProgress {
					c.RequestProgress(ctx)
				}

				if resp.Err() != nil {

Review comment (suggestion):
```suggestion
					lg.Error("Client didn't collect all events", zap.Int64("revision-got", lastRevision), zap.Int64("revision-expected", maxRevision))
```

Code after:
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()
resetWatch:
	for {
		watch := c.Watch(ctx, "", lastRevision+1, true, true, false)
		for {
			select {
			case <-ctx.Done():
				if maxRevision == 0 {
					lg.Error("Client didn't collect all events, max revision not set")
					success = false
				}
				if lastRevision < maxRevision {
					lg.Error("Client didn't collect all events", zap.Int64("revision-got", lastRevision), zap.Int64("revision-expected", maxRevision))
					success = false
				}
				return success
			case revision, ok := <-maxRevisionChan:
				if ok {
					maxRevision = revision
					if lastRevision >= maxRevision {
						cancel()
					}
				} else {
					// Only cancel if maxRevision was never set.
					if maxRevision == 0 {
						cancel()
					}
				}
			case resp, ok := <-watch:
				if !ok {
					lg.Info("Watch channel closed")
					continue resetWatch
				}
				if cfg.RequestProgress {
					c.RequestProgress(ctx)
				}

				if resp.Err() != nil {
					if resp.Canceled {
						if resp.CompactRevision > lastRevision {
							lastRevision = resp.CompactRevision