Repository: nats-io/nats-server
Pull request #3601 (2022-11-02 20:53:36)

[FIXED] On stream proposal failures we could accidentally warn about high stream lag.
We were not taking the failures (mset.clfs) into account.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/.github/workflows/cov.yaml b/.github/workflows/cov.yaml
index d878f62f..8e6a1fa3 100644
--- a/.github/workflows/cov.yaml
+++ b/.github/workflows/cov.yaml
@@ -18,10 +18,9 @@ jobs:
         uses: actions/checkout@v1
         with:
           path: src/github.com/nats-io/nats-server
-          ref: dev
 
       - name: Set up Go
-        uses: actions/setup-go@v3
+        uses: actions/setup-go@v2
         with:
           go-version: 1.19.x
 
@@ -33,7 +32,7 @@ jobs:
           set +e
 
       - name: Convert coverage.out to coverage.lcov
-        uses: jandelgado/gcov2lcov-action@v1.0.9
+        uses: jandelgado/gcov2lcov-action@v1.0.8
         with:
           infile: acc.out
 
diff --git a/.github/workflows/nightly.yaml b/.github/workflows/nightly.yaml
index 8944880c..631255aa 100644
--- a/.github/workflows/nightly.yaml
+++ b/.github/workflows/nightly.yaml
@@ -19,15 +19,14 @@ jobs:
         uses: actions/checkout@v1
         with:
           path: src/github.com/nats-io/nats-server
-          ref: dev
 
       - name: Set up Go
-        uses: actions/setup-go@v3
+        uses: actions/setup-go@v2
         with:
           go-version: 1.19
 
       - name: goreleaser
-        uses: goreleaser/goreleaser-action@v3
+        uses: goreleaser/goreleaser-action@v2
         with:
           version: latest
           args: release --snapshot --config .goreleaser-nightly.yml
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 5e9328b9..63b722c5 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -6468,6 +6468,9 @@ func (mset *stream) checkAllowMsgCompress(peers []string) {
 	mset.mu.Unlock()
 }
 
+// To warn when we are getting too far behind from what has been proposed vs what has been committed.
+const streamLagWarnThreshold = 10_000
+
 // processClusteredMsg will propose the inbound message to the underlying raft group.
 func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg []byte) error {
 	// For possible error response.
@@ -6477,7 +6480,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	canRespond := !mset.cfg.NoAck && len(reply) > 0
 	name, stype, store := mset.cfg.Name, mset.cfg.Storage, mset.store
 	s, js, jsa, st, rf, tierName, outq, node := mset.srv, mset.js, mset.jsa, mset.cfg.Storage, mset.cfg.Replicas, mset.tier, mset.outq, mset.node
-	maxMsgSize, lseq := int(mset.cfg.MaxMsgSize), mset.lseq
+	maxMsgSize, lseq, clfs := int(mset.cfg.MaxMsgSize), mset.lseq, mset.clfs
 	isLeader := mset.isLeader()
 	mset.mu.RUnlock()
 
@@ -6620,7 +6623,9 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	// Check if we need to set initial value here
 	mset.clMu.Lock()
 	if mset.clseq == 0 || mset.clseq < lseq {
-		mset.clseq = mset.lastSeq() + mset.clfs
+		// Re-capture
+		lseq, clfs = mset.lastSeqAndCLFS()
+		mset.clseq = lseq + clfs
 	}
 
 	esm := encodeStreamMsgAllowCompress(subject, reply, hdr, msg, mset.clseq, time.Now().UnixNano(), mset.compressOK)
@@ -6634,8 +6639,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 
 	// Check to see if we are being overrun.
 	// TODO(dlc) - Make this a limit where we drop messages to protect ourselves, but allow to be configured.
-	const warnThreshold = 10_000
-	if mset.clseq-lseq > warnThreshold {
+	if mset.clseq-(lseq+clfs) > streamLagWarnThreshold {
 		lerr := fmt.Errorf("JetStream stream '%s > %s' has high message lag", jsa.acc().Name, mset.cfg.Name)
 		s.RateLimitWarnf(lerr.Error())
 	}


Test patch:
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index d7412151..f6500f04 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -943,3 +943,63 @@ func TestJetStreamClusterConsumerInactiveThreshold(t *testing.T) {
 	t.Run("super-cluster-r1", func(t *testing.T) { test(t, sc.randomCluster(), nil, 1) })
 	t.Run("super-cluster-r3", func(t *testing.T) { test(t, sc.randomCluster(), nil, 3) })
 }
+
+// To capture our false warnings for clustered stream lag.
+type testStreamLagWarnLogger struct {
+	DummyLogger
+	ch chan string
+}
+
+func (l *testStreamLagWarnLogger) Warnf(format string, v ...interface{}) {
+	msg := fmt.Sprintf(format, v...)
+	if strings.Contains(msg, "has high message lag") {
+		select {
+		case l.ch <- msg:
+		default:
+		}
+	}
+}
+
+// False triggering warnings on stream lag because not offsetting by failures.
+func TestJetStreamClusterStreamLagWarning(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	sl := c.streamLeader("$G", "TEST")
+
+	l := &testStreamLagWarnLogger{ch: make(chan string, 10)}
+	sl.SetLogger(l, false, false)
+
+	// We only need to trigger post RAFT propose failures that increment mset.clfs.
+	// Dedupe with msgIDs is one, so we will use that.
+	m := nats.NewMsg("foo")
+	m.Data = []byte("OK")
+	m.Header.Set(JSMsgId, "zz")
+
+	// Make sure we know we will trip the warning threshold.
+	for i := 0; i < 2*streamLagWarnThreshold; i++ {
+		js.PublishMsgAsync(m)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	select {
+	case msg := <-l.ch:
+		t.Fatalf("Unexpected msg lag warning seen: %s", msg)
+	case <-time.After(100 * time.Millisecond):
+		// OK
+	}
+}