fix(badgerd/gossip): wait for leader version catchup during init

Avoid transient init failures during sync replication rolling restarts.
A follower can observe a replicated commit before the leader's MaxVersion
has advanced, which caused "invalid replication version" errors in nightly
stress runs.

Add a bounded wait in Init and cover it with a regression test.

Signed-off-by: Adphi <philippe.adrien.nousse@gmail.com>

Author: Adphi

Makefile

@@ -162,10 +162,10 @@ ci-replication-smoke:
 
 .PHONY: ci-replication-stress
 ci-replication-stress:
-	@go test -v -count 2 -shuffle=on -p 1 -timeout 20m -run '^TestReplicationModes/(async|sync)/leader_churn_under_writes$$' ./tests
-	@go test -v -count 2 -shuffle=on -p 1 -timeout 20m -run '^TestReplicationModes/(async|sync)/follower_offline_catchup_under_load$$' ./tests
-	@go test -v -count 1 -shuffle=on -p 1 -timeout 20m -run '^TestReplicationModes/(async|sync)/rolling_restart_no_data_loss$$' ./tests
-	@go test -v -count 1 -shuffle=on -p 1 -timeout 20m -run '^TestReplicationModes/(async|sync)/delete_propagation_no_resurrection$$' ./tests
+	@go test -v -count 10 -shuffle=on -p 1 -timeout 20m -run '^TestReplicationModes/(async|sync)/leader_churn_under_writes$$' ./tests
+	@go test -v -count 10 -shuffle=on -p 1 -timeout 20m -run '^TestReplicationModes/(async|sync)/follower_offline_catchup_under_load$$' ./tests
+	@go test -v -count 5 -shuffle=on -p 1 -timeout 20m -run '^TestReplicationModes/(async|sync)/rolling_restart_no_data_loss$$' ./tests
+	@go test -v -count 5 -shuffle=on -p 1 -timeout 20m -run '^TestReplicationModes/(async|sync)/delete_propagation_no_resurrection$$' ./tests
 
 .PHONY: ci-integration
 ci-integration:

internal/badgerd/replication/gossip/service.go

@@ -61,6 +61,20 @@ func (r *Gossip) Init(req *pb2.InitRequest, ss pb2.ReplicationService_InitServer
 		return gerrs.Abortedf("node %s is not in the cluster", addr)
 	}
 	m := r.db.MaxVersion()
+	if req.Since > m {
+		// In sync mode a follower can observe an incoming replicated commit just before
+		// the leader's local MaxVersion advances. Give the leader a short window to catch up
+		// before treating the follower as genuinely ahead.
+		deadline := time.Now().Add(250 * time.Millisecond)
+		for req.Since > m && time.Now().Before(deadline) {
+			select {
+			case <-ss.Context().Done():
+				return ss.Context().Err()
+			case <-time.After(10 * time.Millisecond):
+			}
+			m = r.db.MaxVersion()
+		}
+	}
 	if req.Since > m {
 		return gerrs.FailedPreconditionf("invalid replication version %d, max is %d", req.Since, m)
 	}

internal/badgerd/replication/gossip/service_test.go

@@ -99,6 +99,27 @@ func TestInitWaitsForNodeToAppear(t *testing.T) {
 	assert.True(t, called)
 }
 
+func TestInitWaitsForLeaderVersionToCatchUp(t *testing.T) {
+	called := false
+	db := &fakeDB{maxVersion: 0, streamFn: func(_ context.Context, at, since uint64, _ io.Writer) error {
+		called = true
+		assert.EqualValues(t, 2, at)
+		assert.EqualValues(t, 1, since)
+		return nil
+	}}
+	r := &Gossip{leading: NewAtomic(true), db: db, nodes: Map[*node]{}}
+	r.nodes.Store("n1", &node{name: "n1", addr: net.ParseIP("127.0.0.1")})
+
+	go func() {
+		time.Sleep(15 * time.Millisecond)
+		db.SetMaxVersion(2)
+	}()
+
+	err := r.Init(&pb.InitRequest{Since: 1}, &fakeInitSrv{ctx: peerCtx("127.0.0.1", 7000)})
+	require.NoError(t, err)
+	assert.True(t, called)
+}
+
 func TestAliveTable(t *testing.T) {
 	r := &Gossip{}
 	err := r.Alive(&fakeAliveSrv{ctx: context.Background(), recvErrs: []error{io.EOF}})

internal/badgerd/replication/gossip/test_helpers_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"io"
 	"net"
+	"sync"
 
 	"github.com/dgraph-io/badger/v3"
 	"google.golang.org/grpc"
@@ -16,6 +17,7 @@ import (
 )
 
 type fakeDB struct {
+	mu            sync.RWMutex
 	path          string
 	inMemory      bool
 	maxVersion    uint64
@@ -27,10 +29,23 @@ type fakeDB struct {
 	batch         *fakeWriteBatch
 }
 
-func (f *fakeDB) Path() string                                    { return f.path }
-func (f *fakeDB) InMemory() bool                                  { return f.inMemory }
-func (f *fakeDB) MaxVersion() uint64                              { return f.maxVersion }
-func (f *fakeDB) SetVersion(v uint64)                             { f.version = v }
+func (f *fakeDB) Path() string   { return f.path }
+func (f *fakeDB) InMemory() bool { return f.inMemory }
+func (f *fakeDB) MaxVersion() uint64 {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+	return f.maxVersion
+}
+func (f *fakeDB) SetVersion(v uint64) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.version = v
+}
+func (f *fakeDB) SetMaxVersion(v uint64) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.maxVersion = v
+}
 func (f *fakeDB) Drop() error                                     { return nil }
 func (f *fakeDB) Load(context.Context, io.Reader) (uint64, error) { return 0, nil }
 func (f *fakeDB) Stream(ctx context.Context, at, since uint64, w io.Writer) error {