Repository: nats-io/nats-server
Pull request #5827 (2024-08-26 08:43:27)

Don't return `seq=0, duplicate=true` when processing duplicate messages asynchronously
This issue was reported by Marco and Antithesis.

`seq=0, duplicate=true` would be returned if two duplicate messages would be processed at the same time. The first will propagate, which will take some time for a clustered stream, but then the second message gets processed and rejected with `seq=0, duplicate=true` (without `error`) which would be invalid.

One option to fix this is timing out the second publish and wait for the original message to finish processing.
That fixes this issue, and the user would just retry the request and get the valid sequence once the other message has gone through.

However, it could decrease message throughput when this is hit. An alternative could be to throw an error, but if the user would always retry if `err != nil` then they'd be "spamming" the same request until the original message is fully propagated.

Would like to hear your thoughts :) 
Maybe this would be more suitable for 2.11.x as well?

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>


MauriceVanVeen: `TestJetStreamClusterStreamLagWarning` genuinely fails now, since it relies on publishing many messages with the same `msgId` without timeout. Could be solved by not using msgId, or the error approach.
derekcollison: @MauriceVanVeen will you be modifying the test or waiting on others for that?
MauriceVanVeen: > will you be modifying the test or waiting on others for that?

Have pushed a fix for the test, to publish a message into the stream first to ensure we know the stored seq.

Assuming the timeout approach is okay?
MauriceVanVeen: Yeah me too, I'm wondering what the proper way to go about solving this is.

Currently you'd get:
```
msg1: { stream: A, seq: 1 }
msg2: { stream: A, seq: 0, duplicate: true }
```
For two messages sent at the same time with the same `msgId`. Sequence should be 1 instead.


So, the desired state would be:
```
msg1: { stream: A, seq: 1 }
msg2: { stream: A, seq: 1, duplicate: true }
```
So that the second message would be `duplicate` but also the correct seq>0 / `seq=0` would be returned.


Currently the PR does that by introducing a timeout:
```
msg1: { stream: A, seq: 1 }
msg2 (try 1): // timeout
msg2 (try 2): { stream: A, seq: 1, duplicate: true }
```

That could instead be made into an error instead? Noting a message with the same `msgId` is currently already being propagated / it's in progress?
MauriceVanVeen: Another alternative could be to set the seq to the seq that's being proposed to the followers, but that could be a bit dangerous if the proposal fails and is retried. Then the seq would be different than what was returned with `duplicate`.
MauriceVanVeen: Updated to use an error instead of a timeout, which will give more insight into the message being updated at the same time.
derekcollison: The duplicate flag still set though yes?

Patch:
diff --git a/server/errors.json b/server/errors.json
index c7eaef29..12912ae3 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1558,5 +1558,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSStreamDuplicateMessageConflict",
+    "code": 409,
+    "error_code": 10158,
+    "description": "duplicate message id is in process",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 28bfea7d..a59ae021 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -7893,10 +7893,18 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 				pubAck := append(buf[:0], mset.pubAck...)
 				seq := dde.seq
 				mset.mu.Unlock()
+				// Should not return an invalid sequence, in that case timeout.
 				if canRespond {
-					response := append(pubAck, strconv.FormatUint(seq, 10)...)
-					response = append(response, ",\"duplicate\": true}"...)
-					outq.sendMsg(reply, response)
+					if seq > 0 {
+						response := append(pubAck, strconv.FormatUint(seq, 10)...)
+						response = append(response, ",\"duplicate\": true}"...)
+						outq.sendMsg(reply, response)
+					} else {
+						var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
+						resp.Error = ApiErrors[JSStreamDuplicateMessageConflict]
+						b, _ := json.Marshal(resp)
+						outq.sendMsg(reply, b)
+					}
 				}
 				return errMsgIdDuplicate
 			}
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 7115dc4a..6c68094e 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -341,6 +341,9 @@ const (
 	// JSStreamDeleteErrF General stream deletion error string ({err})
 	JSStreamDeleteErrF ErrorIdentifier = 10050
 
+	// JSStreamDuplicateMessageConflict duplicate message id is in process
+	JSStreamDuplicateMessageConflict ErrorIdentifier = 10158
+
 	// JSStreamExternalApiOverlapErrF stream external api prefix {prefix} must not overlap with {subject}
 	JSStreamExternalApiOverlapErrF ErrorIdentifier = 10021
 
@@ -588,6 +591,7 @@ var (
 		JSStreamAssignmentErrF:                     {Code: 500, ErrCode: 10048, Description: "{err}"},
 		JSStreamCreateErrF:                         {Code: 500, ErrCode: 10049, Description: "{err}"},
 		JSStreamDeleteErrF:                         {Code: 500, ErrCode: 10050, Description: "{err}"},
+		JSStreamDuplicateMessageConflict:           {Code: 409, ErrCode: 10158, Description: "duplicate message id is in process"},
 		JSStreamExternalApiOverlapErrF:             {Code: 400, ErrCode: 10021, Description: "stream external api prefix {prefix} must not overlap with {subject}"},
 		JSStreamExternalDelPrefixOverlapsErrF:      {Code: 400, ErrCode: 10022, Description: "stream external delivery prefix {prefix} overlaps with stream subject {subject}"},
 		JSStreamGeneralErrorF:                      {Code: 500, ErrCode: 10051, Description: "{err}"},
@@ -1921,6 +1925,16 @@ func NewJSStreamDeleteError(err error, opts ...ErrorOption) *ApiError {
 	}
 }
 
+// NewJSStreamDuplicateMessageConflictError creates a new JSStreamDuplicateMessageConflict error: "duplicate message id is in process"
+func NewJSStreamDuplicateMessageConflictError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSStreamDuplicateMessageConflict]
+}
+
 // NewJSStreamExternalApiOverlapError creates a new JSStreamExternalApiOverlapErrF error: "stream external api prefix {prefix} must not overlap with {subject}"
 func NewJSStreamExternalApiOverlapError(prefix interface{}, subject interface{}, opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)


Test patch:
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 718f2314..c3ff6e31 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -3070,3 +3070,111 @@ func TestJetStreamClusterKeyValueLastSeqMismatch(t *testing.T) {
 	require_Error(t, err)
 	require_Equal(t, err.Error(), `nats: wrong last sequence: 0`)
 }
+
+func TestJetStreamClusterPubAckSequenceDupe(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "TEST_CLUSTER", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	type client struct {
+		nc *nats.Conn
+		js nats.JetStreamContext
+	}
+
+	clients := make([]client, len(c.servers))
+	for i, server := range c.servers {
+		clients[i].nc, clients[i].js = jsClientConnect(t, server)
+		defer clients[i].nc.Close()
+	}
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:       "TEST_STREAM",
+		Subjects:   []string{"TEST_SUBJECT.*"},
+		Replicas:   3,
+		Duplicates: 1 * time.Minute,
+	})
+	require_NoError(t, err)
+
+	msgData := []byte("...")
+
+	for seq := uint64(1); seq < 10; seq++ {
+
+		if seq%3 == 0 {
+			c.restartAll()
+		}
+
+		msgSubject := "TEST_SUBJECT." + strconv.FormatUint(seq, 10)
+		msgIdOpt := nats.MsgId(nuid.Next())
+
+		firstPublisherClient := &clients[rand.Intn(len(clients))]
+		secondPublisherClient := &clients[rand.Intn(len(clients))]
+
+		pubAck1, err := firstPublisherClient.js.Publish(msgSubject, msgData, msgIdOpt)
+		require_NoError(t, err)
+		require_Equal(t, seq, pubAck1.Sequence)
+		require_False(t, pubAck1.Duplicate)
+
+		pubAck2, err := secondPublisherClient.js.Publish(msgSubject, msgData, msgIdOpt)
+		require_NoError(t, err)
+		require_Equal(t, seq, pubAck2.Sequence)
+		require_True(t, pubAck2.Duplicate)
+
+	}
+
+}
+
+func TestJetStreamClusterPubAckSequenceDupeAsync(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "TEST_CLUSTER", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:       "TEST_STREAM",
+		Subjects:   []string{"TEST_SUBJECT"},
+		Replicas:   3,
+		Duplicates: 1 * time.Minute,
+	})
+	require_NoError(t, err)
+
+	msgData := []byte("...")
+
+	for seq := uint64(1); seq < 10; seq++ {
+
+		msgSubject := "TEST_SUBJECT"
+		msgIdOpt := nats.MsgId(nuid.Next())
+		conflictErr := &nats.APIError{ErrorCode: nats.ErrorCode(JSStreamDuplicateMessageConflict)}
+
+		wg := sync.WaitGroup{}
+		wg.Add(2)
+
+		// Fire off 2 publish requests in parallel
+		// The first one "stages" a duplicate entry before even proposing the message
+		// The second one gets a pubAck with sequence zero by hitting the staged duplicated entry
+
+		pubAcks := [2]*nats.PubAck{}
+		for i := 0; i < 2; i++ {
+			go func(i int) {
+				defer wg.Done()
+				var err error
+				pubAcks[i], err = js.Publish(msgSubject, msgData, msgIdOpt)
+				// Conflict on duplicate message, wait a bit before retrying to get the proper pubAck.
+				if errors.Is(err, conflictErr) {
+					time.Sleep(time.Millisecond * 500)
+					pubAcks[i], err = js.Publish(msgSubject, msgData, msgIdOpt)
+				}
+				require_NoError(t, err)
+			}(i)
+		}
+
+		wg.Wait()
+		require_Equal(t, pubAcks[0].Sequence, seq)
+		require_Equal(t, pubAcks[1].Sequence, seq)
+
+		// Exactly one of the pubAck should be marked dupe
+		require_True(t, (pubAcks[0].Duplicate || pubAcks[1].Duplicate) && (pubAcks[0].Duplicate != pubAcks[1].Duplicate))
+	}
+}