perf(replication): optimize gossip tx replication paths and stabilize async lifecycle

Batch non-boundary ops in ReplicateRequest buffers, add sync direct stream send/recv fast path, and make async send non-blocking while fixing shutdown/context races that caused stale peer writes and queue-close replication failures.

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

Author: Adphi

internal/badgerd/replication/async/queue.go

@@ -177,6 +177,8 @@ func (q *Queue[Req, Res]) run() {
 				return
 			}
 			msg.ch <- &Result[Res]{}
+			closeSafe(q.close)
+			return
 		case <-q.close:
 			return
 		}

internal/badgerd/replication/gossip/gossip.go

@@ -231,7 +231,7 @@ func New(ctx context.Context, db replication.DB, opts ...replication.Option) (re
 		return nil, err
 	}
 
-	r.run(ctx)
+	r.run(r.ctx)
 	<-r.ready
 	log.Infof("init replication")
 	if err := r.init(ctx); err != nil {
@@ -277,9 +277,20 @@ func (r *Gossip) run(ctx context.Context) {
 	go r.handleEvents(ctx)
 	go func() {
 		log.Infof("waiting for leader or version to converge")
-		<-r.converged
+		select {
+		case <-r.converged:
+		case <-ctx.Done():
+			return
+		}
 		log.Infof("leader or version converged")
-		time.Sleep(time.Duration(rand.Intn(1000))*time.Millisecond + 100*time.Millisecond)
+		select {
+		case <-time.After(time.Duration(rand.Intn(1000))*time.Millisecond + 100*time.Millisecond):
+		case <-ctx.Done():
+			return
+		}
+		if ctx.Err() != nil {
+			return
+		}
 		r.Elect(ctx)
 	}()
 }

internal/badgerd/replication/gossip/nodes.go

@@ -47,11 +47,17 @@ type node struct {
 
 func (r *Gossip) handleEvents(ctx context.Context) {
 	for e := range r.events {
-		go r.handleEvent(ctx, e)
+		if ctx.Err() != nil {
+			return
+		}
+		r.handleEvent(ctx, e)
 	}
 }
 
 func (r *Gossip) handleEvent(ctx context.Context, e memberlist.NodeEvent) {
+	if ctx.Err() != nil {
+		return
+	}
 	log := logger.C(ctx).WithField("component", "replication")
 	r.mu.Lock()
 	defer r.mu.Unlock()