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

Code before:
// 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 cache

import (
	"fmt"
	"sync"

	"github.com/google/btree"

	"go.etcd.io/etcd/api/v3/mvccpb"
	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
	clientv3 "go.etcd.io/etcd/client/v3"
)

var ErrNotReady = fmt.Errorf("cache: store not ready")

// The store keeps a bounded history of snapshots using ringBuffer so that
// reads at historical revisions can be served until they fall out of the window.
type store struct {
	mu      sync.RWMutex
	degree  int
	latest  snapshot
	history ringBuffer[*snapshot]
}

func newStore(degree int, historyCapacity int) *store {
	tree := btree.New(degree)
	return &store{
		degree:  degree,
		latest:  snapshot{rev: 0, tree: tree},
		history: *newRingBuffer(historyCapacity, func(s *snapshot) int64 { return s.rev }),
	}
}

type kvItem struct {
	key string
	kv  *mvccpb.KeyValue
}

func newKVItem(kv *mvccpb.KeyValue) *kvItem {
	return &kvItem{key: string(kv.Key), kv: kv}
}

func (a *kvItem) Less(b btree.Item) bool {
	return a.key < b.(*kvItem).key
}

func (s *store) Get(startKey, endKey []byte, rev int64) ([]*mvccpb.KeyValue, int64, error) {

Review comment (suggestion):
Would be good to clarify relation between `latest` and `history` in comments. Things like:
* Is latest snapshot also present in history?
* If latest is in history, is it same tree or a clone?

My suggestions: 
* History should have latest snapshot for simpler `Get` implementation. Just scan history without checking latest.
* It should be a clone and not direct latest, as history should be immutable. This allows us for lockless read from store.

Code after:
// 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 cache

import (
	"fmt"
	"sync"

	"github.com/google/btree"

	"go.etcd.io/etcd/api/v3/mvccpb"
	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
	clientv3 "go.etcd.io/etcd/client/v3"
)

var ErrNotReady = fmt.Errorf("cache: store not ready")

// The store keeps a bounded history of snapshots using ringBuffer so that
// reads at historical revisions can be served until they fall out of the window.
type store struct {
	mu      sync.RWMutex
	degree  int
	latest  snapshot              // latest is the mutable working snapshot
	history ringBuffer[*snapshot] // history stores immutable cloned snapshots
}

func newStore(degree int, historyCapacity int) *store {
	tree := btree.New(degree)
	return &store{
		degree:  degree,
		latest:  snapshot{rev: 0, tree: tree},
		history: *newRingBuffer(historyCapacity, func(s *snapshot) int64 { return s.rev }),
	}
}

type kvItem struct {
	key string
	kv  *mvccpb.KeyValue
}

func newKVItem(kv *mvccpb.KeyValue) *kvItem {
	return &kvItem{key: string(kv.Key), kv: kv}
}

func (a *kvItem) Less(b btree.Item) bool {
	return a.key < b.(*kvItem).key
}

func (s *store) Get(startKey, endKey []byte, rev int64) ([]*mvccpb.KeyValue, int64, error) {
Repository: etcd-io/etcd
File: cache/store.go (Go)

Code before:
func (s *store) Restore(kvs []*mvccpb.KeyValue, rev int64) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.latest.tree = btree.New(s.degree)
	for _, kv := range kvs {
		s.latest.tree.ReplaceOrInsert(newKVItem(kv))
	}
	s.history.RebaseHistory()
	s.latest.rev = rev
	s.history.Append(newClonedSnapshot(rev, s.latest.tree))
}

func (s *store) Apply(events []*clientv3.Event) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if err := validateRevisions(events, s.latest.rev); err != nil {
		return err
	}

	curRev := s.latest.rev
	workingTree := s.latest.tree

	for i := 0; i < len(events); {
		batchRev := events[i].Kv.ModRevision

		for i < len(events) && events[i].Kv.ModRevision == batchRev {
			ev := events[i]
			switch ev.Type {
			case clientv3.EventTypeDelete:
				if removed := workingTree.Delete(&kvItem{key: string(ev.Kv.Key)}); removed == nil {
					return fmt.Errorf("cache: delete non-existent key %s", string(ev.Kv.Key))
				}
			case clientv3.EventTypePut:
				workingTree.ReplaceOrInsert(newKVItem(ev.Kv))
			}
			i++
		}
		curRev = batchRev
		s.history.Append(newClonedSnapshot(curRev, workingTree))
	}

	s.latest.tree = workingTree
	s.latest.rev = curRev
	return nil
}

func (s *store) LatestRev() int64 {
	s.mu.RLock()
	defer s.mu.RUnlock()

Review comment (performance):
```suggestion
		curRev = events[i].Kv.ModRevision
```

batchRev works same as curRev. No need to have a separate variable.

Code after:
func (s *store) Restore(kvs []*mvccpb.KeyValue, rev int64) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.latest.tree = btree.New(s.degree)
	for _, kv := range kvs {
		s.latest.tree.ReplaceOrInsert(newKVItem(kv))
	}
	s.history.RebaseHistory()
	s.latest.rev = rev
	s.history.Append(newClonedSnapshot(rev, s.latest.tree))
}

func (s *store) Apply(events []*clientv3.Event) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if err := validateRevisions(events, s.latest.rev); err != nil {
		return err
	}

	for i := 0; i < len(events); {
		rev := events[i].Kv.ModRevision

		for i < len(events) && events[i].Kv.ModRevision == rev {
			ev := events[i]
			switch ev.Type {
			case clientv3.EventTypeDelete:
				if removed := s.latest.tree.Delete(&kvItem{key: string(ev.Kv.Key)}); removed == nil {
					return fmt.Errorf("cache: delete non-existent key %s", string(ev.Kv.Key))
				}
			case clientv3.EventTypePut:
				s.latest.tree.ReplaceOrInsert(newKVItem(ev.Kv))
			}
			i++
		}
		s.latest.rev = rev
		s.history.Append(newClonedSnapshot(rev, s.latest.tree))
	}
	return nil
}

func (s *store) LatestRev() int64 {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.latest.rev
}

func validateRevisions(events []*clientv3.Event, latestRev int64) error {
	if len(events) == 0 {
		return nil
Repository: etcd-io/etcd
File: cache/store.go (Go)

Code before:
	return targetSnapshot, s.latest.rev, nil
}

// Restore replaces state with the bootstrap snapshot and resets history.
func (s *store) Restore(kvs []*mvccpb.KeyValue, rev int64) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.latest.tree = btree.New(s.degree)
	for _, kv := range kvs {
		s.latest.tree.ReplaceOrInsert(newKVItem(kv))
	}
	s.history.RebaseHistory()
	s.latest.rev = rev
	s.history.Append(newClonedSnapshot(rev, s.latest.tree))
}

func (s *store) Apply(events []*clientv3.Event) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if err := validateRevisions(events, s.latest.rev); err != nil {
		return err
	}

	curRev := s.latest.rev
	workingTree := s.latest.tree

	for i := 0; i < len(events); {
		batchRev := events[i].Kv.ModRevision

		for i < len(events) && events[i].Kv.ModRevision == batchRev {
			ev := events[i]
			switch ev.Type {
			case clientv3.EventTypeDelete:
				if removed := workingTree.Delete(&kvItem{key: string(ev.Kv.Key)}); removed == nil {
					return fmt.Errorf("cache: delete non-existent key %s", string(ev.Kv.Key))
				}
			case clientv3.EventTypePut:
				workingTree.ReplaceOrInsert(newKVItem(ev.Kv))
			}
			i++
		}
		curRev = batchRev
		s.history.Append(newClonedSnapshot(curRev, workingTree))
	}

	s.latest.tree = workingTree
	s.latest.rev = curRev
	return nil
}

Review comment (suggestion):
Recommend to not have separate variables for `curRev` or `workingTree`. Just use `s.latest.rev` and `s.latest.tree`

Code after:
	return targetSnapshot, s.latest.rev, nil
}

// Restore replaces state with the bootstrap snapshot and resets history.
func (s *store) Restore(kvs []*mvccpb.KeyValue, rev int64) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.latest.tree = btree.New(s.degree)
	for _, kv := range kvs {
		s.latest.tree.ReplaceOrInsert(newKVItem(kv))
	}
	s.history.RebaseHistory()
	s.latest.rev = rev
	s.history.Append(newClonedSnapshot(rev, s.latest.tree))
}

func (s *store) Apply(events []*clientv3.Event) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if err := validateRevisions(events, s.latest.rev); err != nil {
		return err
	}

	for i := 0; i < len(events); {
		rev := events[i].Kv.ModRevision

		for i < len(events) && events[i].Kv.ModRevision == rev {
			ev := events[i]
			switch ev.Type {
			case clientv3.EventTypeDelete:
				if removed := s.latest.tree.Delete(&kvItem{key: string(ev.Kv.Key)}); removed == nil {
					return fmt.Errorf("cache: delete non-existent key %s", string(ev.Kv.Key))
				}
			case clientv3.EventTypePut:
				s.latest.tree.ReplaceOrInsert(newKVItem(ev.Kv))
			}
			i++
		}
		s.latest.rev = rev
		s.history.Append(newClonedSnapshot(rev, s.latest.tree))
	}
	return nil
}

func (s *store) LatestRev() int64 {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.latest.rev
}
Repository: etcd-io/etcd
File: server/embed/config_logging.go (Go)

Code before:

package embed

import (
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/url"
	"os"

	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"go.uber.org/zap/zapgrpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/grpclog"
	"gopkg.in/natefinch/lumberjack.v2"

	"go.etcd.io/etcd/client/pkg/v3/logutil"
)

// tlsConn is the minimal interface for mocking *tls.Conn for the TLS handshake failure handler.
// Only used for testing as we cannot really chang the public interface using *tls.Conn.
type tlsConn interface {
	ConnectionState() tls.ConnectionState
	RemoteAddr() net.Addr
}

// testingHandshakeFailureTLSConn is used to replace the real TLS conn when testing the TLS handshake failure handler.
// Only used for testing.
var testingHandshakeFailureTLSConn tlsConn

// GetLogger returns the logger.
func (cfg *Config) GetLogger() *zap.Logger {
	cfg.loggerMu.RLock()
	l := cfg.logger
	cfg.loggerMu.RUnlock()
	return l
}

// setupLogging initializes etcd logging.
// Must be called after flag parsing or finishing configuring embed.Config.
func (cfg *Config) setupLogging() error {
	switch cfg.Logger {
	case "capnslog": // removed in v3.5
		return fmt.Errorf("--logger=capnslog is removed in v3.5")

	case "zap":
		if len(cfg.LogOutputs) == 0 {

Review comment (suggestion):
While I usually applaud testing, it might be overkill in this case.

Testing should cover behaviors users depend on. I don't think anyone depends on fact whether log is present. I would remove the test and just leave lines 188-192.

Code after:

package embed

import (
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/url"
	"os"

	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"go.uber.org/zap/zapgrpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/grpclog"
	"gopkg.in/natefinch/lumberjack.v2"

	"go.etcd.io/etcd/client/pkg/v3/logutil"
)

// GetLogger returns the logger.
func (cfg *Config) GetLogger() *zap.Logger {
	cfg.loggerMu.RLock()
	l := cfg.logger
	cfg.loggerMu.RUnlock()
	return l
}

// setupLogging initializes etcd logging.
// Must be called after flag parsing or finishing configuring embed.Config.
func (cfg *Config) setupLogging() error {
	switch cfg.Logger {
	case "capnslog": // removed in v3.5
		return fmt.Errorf("--logger=capnslog is removed in v3.5")

	case "zap":
		if len(cfg.LogOutputs) == 0 {
			cfg.LogOutputs = []string{DefaultLogOutput}
		}
		if len(cfg.LogOutputs) > 1 {
			for _, v := range cfg.LogOutputs {
				if v == DefaultLogOutput {
					return fmt.Errorf("multi logoutput for %q is not supported yet", DefaultLogOutput)
				}
			}
		}
		if cfg.EnableLogRotation {
			if err := setupLogRotation(cfg.LogOutputs, cfg.LogRotationConfigJSON); err != nil {
				return err
Repository: etcd-io/etcd
File: server/embed/config_logging.go (Go)

Code before:
			}

			// WARN: do not change field names in encoder config
			// journald logging writer assumes field names of "level" and "caller"
			cr := zapcore.NewCore(
				encoder,
				syncer,
				lvl,
			)
			if cfg.ZapLoggerBuilder == nil {
				cfg.ZapLoggerBuilder = NewZapLoggerBuilder(zap.New(cr, zap.AddCaller(), zap.ErrorOutput(syncer)))
			}
		}

		err := cfg.ZapLoggerBuilder(cfg)
		if err != nil {
			return err
		}

		logTLSHandshakeFailureFunc := func(msg string) func(conn *tls.Conn, err error) {
			return func(conn *tls.Conn, err error) {
				// Log EOF errors on DEBUG not to spam logs too much.
				log := cfg.logger.Warn
				if errors.Is(err, io.EOF) {
					log = cfg.logger.Debug
				}

				state := conn.ConnectionState()
				remoteAddr := conn.RemoteAddr().String()
				serverName := state.ServerName
				if len(state.PeerCertificates) > 0 {
					cert := state.PeerCertificates[0]
					ips := make([]string, len(cert.IPAddresses))
					for i := range cert.IPAddresses {
						ips[i] = cert.IPAddresses[i].String()
					}
					log(
						msg,
						zap.String("remote-addr", remoteAddr),
						zap.String("server-name", serverName),
						zap.Strings("ip-addresses", ips),
						zap.Strings("dns-names", cert.DNSNames),
						zap.Error(err),
					)
				} else {
					log(
						msg,
						zap.String("remote-addr", remoteAddr),
						zap.String("server-name", serverName),
						zap.Error(err),
					)

Review comment (bug):
Minor comment: to avoid potential confusion with the std package `log`
```suggestion
				logFunc := cfg.logger.Warn
				if errors.Is(err, io.EOF) {
					logFunc = cfg.logger.Debug
				}
```

Code after:
			}

			// WARN: do not change field names in encoder config
			// journald logging writer assumes field names of "level" and "caller"
			cr := zapcore.NewCore(
				encoder,
				syncer,
				lvl,
			)
			if cfg.ZapLoggerBuilder == nil {
				cfg.ZapLoggerBuilder = NewZapLoggerBuilder(zap.New(cr, zap.AddCaller(), zap.ErrorOutput(syncer)))
			}
		}

		err := cfg.ZapLoggerBuilder(cfg)
		if err != nil {
			return err
		}

		logTLSHandshakeFailureFunc := func(msg string) func(conn *tls.Conn, err error) {
			return func(conn *tls.Conn, err error) {
				// Log EOF errors on DEBUG not to spam logs too much.
				logFunc := cfg.logger.Warn
				if errors.Is(err, io.EOF) {
					logFunc = cfg.logger.Debug
				}

				state := conn.ConnectionState()
				remoteAddr := conn.RemoteAddr().String()
				serverName := state.ServerName
				if len(state.PeerCertificates) > 0 {
					cert := state.PeerCertificates[0]
					ips := make([]string, len(cert.IPAddresses))
					for i := range cert.IPAddresses {
						ips[i] = cert.IPAddresses[i].String()
					}
					logFunc(
						msg,
						zap.String("remote-addr", remoteAddr),
						zap.String("server-name", serverName),
						zap.Strings("ip-addresses", ips),
						zap.Strings("dns-names", cert.DNSNames),
						zap.Error(err),
					)
				} else {
					logFunc(
						msg,
						zap.String("remote-addr", remoteAddr),
						zap.String("server-name", serverName),
						zap.Error(err),
					)
Repository: etcd-io/etcd
File: server/embed/config_test.go (Go)

Code before:
	}
}

func TestFastLeaseKeepAliveValidate(t *testing.T) {
	tcs := []struct {
		name               string
		serverFeatureGates string
		expectError        bool
	}{
		{
			name: "Default config should pass",
		},
		{
			name:               "Enabling FastLeaseKeepAlive should pass",
			serverFeatureGates: "FastLeaseKeepAlive=true",
		},
		{
			name:               "Disabling FastLeaseKeepAlive should pass",
			serverFeatureGates: "FastLeaseKeepAlive=false",
		},
	}
	for _, tc := range tcs {
		t.Run(tc.name, func(t *testing.T) {
			cfg := *NewConfig()
			cfg.ServerFeatureGate.(featuregate.MutableFeatureGate).Set(tc.serverFeatureGates)
			err := cfg.Validate()
			if (err != nil) != tc.expectError {
				t.Errorf("config.Validate() = %q, expected error: %v", err, tc.expectError)
			}
		})
	}
}


Review comment (suggestion):
```suggestion
			require.NoError(t, cfg.Validate())
```

Code after:
			cfg.InitialCluster = ""
			cfg.DiscoveryCfg = tc.discoveryCfg
			cfg.DiscoveryCfg.Token = "foo"
			err := cfg.Validate()

			require.Equal(t, tc.wantErr, err != nil)
		})
	}
}

func TestFastLeaseKeepAliveValidate(t *testing.T) {
	tcs := []struct {
		name               string
		serverFeatureGates string
		expectEnabled      bool
	}{
		{
			name:          "Default config should pass",
			expectEnabled: true,
		},
		{
			name:               "Enabling FastLeaseKeepAlive should pass",
			serverFeatureGates: "FastLeaseKeepAlive=true",
			expectEnabled:      true,
		},
		{
			name:               "Disabling FastLeaseKeepAlive should pass",
			serverFeatureGates: "FastLeaseKeepAlive=false",
			expectEnabled:      false,
		},
	}
	for _, tc := range tcs {
		t.Run(tc.name, func(t *testing.T) {
			cfg := *NewConfig()
			cfg.ServerFeatureGate.(featuregate.MutableFeatureGate).Set(tc.serverFeatureGates)
			require.NoError(t, cfg.Validate())
			require.Equal(t, tc.expectEnabled, cfg.ServerFeatureGate.Enabled(features.FastLeaseKeepAlive))
		})
	}
}
Repository: etcd-io/etcd
File: tests/robustness/traffic/traffic.go (Go)

Code before:
		for _, w := range r.Watch {
			var (
				firstInWindow time.Duration
				lastInWindow  time.Duration
				haveInWindow  bool

				noEventsYetInWindow = true
				closedCounted       = false
			)

			for _, resp := range w.Responses {
				if resp.Time < start || resp.Time > end {
					continue
				}
				if !haveInWindow {
					firstInWindow = resp.Time
					haveInWindow = true
				}
				lastInWindow = resp.Time
				if resp.IsProgressNotify {
					ws.ProgressNotifies++
				}
				if len(resp.Events) > 0 {
					ws.Events += len(resp.Events)
					rev := resp.Revision
					if ws.MinRev == 0 || rev < ws.MinRev {
						ws.MinRev = rev
					}
					if rev > ws.MaxRev {
						ws.MaxRev = rev
					}
					seenRevs[rev] = struct{}{}
					noEventsYetInWindow = false
				}
				if resp.Error != "" && noEventsYetInWindow && !closedCounted {
					ws.ImmediateClosures++
					closedCounted = true
				}
			}

			if haveInWindow {
				ws.Requests++
				if lastInWindow > firstInWindow {
					ws.SumDuration += lastInWindow - firstInWindow
					ws.DurationsCount++
				}
			}
		}
	}

	ws.CoveredRevisions = len(seenRevs)

Review comment (bug):
I think you are mixing properties of "watch stream" and singular "events". While it is useful to know what was the duration of watches during the whole test, I don't think it's expose this metric based on whether watch emitted any event within a window of failure injection.

Code after:
			for _, resp := range w.Responses {
				if resp.Time < start || resp.Time > end {
					continue
				}
				if !haveInWindow {
					firstInWindow = resp.Time
					haveInWindow = true
				}
				lastInWindow = resp.Time
				if resp.IsProgressNotify {
					ws.ProgressNotifies++
				}
				if len(resp.Events) > 0 {
					ws.Events += len(resp.Events)
					noEventsYetInWindow = false
				}
				if resp.Error != "" && noEventsYetInWindow && !closedCounted {
					ws.ImmediateClosures++
					closedCounted = true
				}
			}

			if haveInWindow {
				ws.Requests++
				if lastInWindow > firstInWindow {
					ws.SumDuration += lastInWindow - firstInWindow
					ws.DurationsCount++
				}
			}
		}
	}

	return ws
}

type watchStats struct {
	Period            time.Duration
	Requests          int
	Events            int
	ProgressNotifies  int
	ImmediateClosures int
	SumDuration       time.Duration
	DurationsCount    int
}

func (ws *watchStats) AvgDuration() time.Duration {
	if ws.DurationsCount == 0 {
		return 0
	}
	return ws.SumDuration / time.Duration(ws.DurationsCount)
}
Repository: etcd-io/etcd
File: cache/demux.go (Go)

Code before:
		if nextRev != 0 && firstRev > nextRev {
			d.laggingWatchers[w] = nextRev
			delete(d.activeWatchers, w)
			continue
		}
		sendStart := len(events)
		for i, ev := range events {
			if ev.Kv.ModRevision >= nextRev {
				sendStart = i
				break
			}
		}
		if sendStart == len(events) {
			continue
		}

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

func (d *demux) SetProgressRev(rev int64) {
	d.mu.Lock()
	defer d.mu.Unlock()
	if rev > d.progressRev {
		d.progressRev = rev
	}
}

func (d *demux) ProgressRev() int64 {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return d.progressRev
}

// Purge stops all watchers and rebase history on watch errors
func (d *demux) Purge() {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.history.RebaseHistory()
	d.progressRev = 0
	for w := range d.activeWatchers {
		w.Stop()
	}
	for w := range d.laggingWatchers {
		w.Stop()
	}

Review comment (suggestion):
Instead of dedicated function for setting rev, please modify `Broadcast` to accept `watch response.

Code after:
	}
	// Watch opened on revision mismatching dmux last observed revision.
	d.purge()
	d.minRev = minRev
}

func (d *demux) Broadcast(resp clientv3.WatchResponse) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	if d.minRev == 0 {
		return errors.New("demux: not initialized")
	}
	err := validateRevisions(resp, d.maxRev)
	if err != nil {
		return err
	}
	d.updateStoreLocked(resp)
	d.broadcastLocked(resp)
	return nil
}

func (d *demux) LatestRev() int64 {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return d.maxRev
}

func (d *demux) updateStoreLocked(resp clientv3.WatchResponse) {
	if resp.IsProgressNotify() {
		d.maxRev = resp.Header.Revision
		return
	}
	if len(resp.Events) == 0 {
		return
	}
	events := resp.Events
	batchStart := 0
	for end := 1; end < len(events); end++ {
		if events[end].Kv.ModRevision != events[batchStart].Kv.ModRevision {
			if end > batchStart {
				if end+1 == len(events) && d.history.full() {
					d.minRev = d.history.PeekOldest() + 1
				}
				d.history.Append(events[batchStart:end])
			}
			batchStart = end
		}
	}
	if batchStart < len(events) {
		if d.history.full() {
			d.minRev = d.history.PeekOldest() + 1