(2.12) [IMPROVED] Error on missing Nats-Expected-Last-Subject-Sequence (#7196)

The `Nats-Expected-Last-Subject-Sequence-Subject` header was added in
#5281, but we would not
return an error if it was specified without the
`Nats-Expected-Last-Subject-Sequence` header.

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

Author: neilalexander

server/errors.json

@@ -1908,5 +1908,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSStreamExpectedLastSeqPerSubjectInvalid",
+    "code": 400,
+    "error_code": 10193,
+    "description": "missing sequence for expected last sequence per subject",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]

server/jetstream_batching.go

@@ -465,6 +465,9 @@ func checkMsgHeadersPreClusteredProposal(
 					diff.expectedPerSubject[seqSubj] = e
 				}
 			}
+		} else if getExpectedLastSeqPerSubjectForSubject(hdr) != _EMPTY_ {
+			apiErr := NewJSStreamExpectedLastSeqPerSubjectInvalidError()
+			return hdr, msg, 0, apiErr, apiErr
 		}
 
 		// Message scheduling.

server/jetstream_errors_generated.go

@@ -434,6 +434,9 @@ const (
 	// JSStreamDuplicateMessageConflict duplicate message id is in process
 	JSStreamDuplicateMessageConflict ErrorIdentifier = 10158
 
+	// JSStreamExpectedLastSeqPerSubjectInvalid missing sequence for expected last sequence per subject
+	JSStreamExpectedLastSeqPerSubjectInvalid ErrorIdentifier = 10193
+
 	// JSStreamExpectedLastSeqPerSubjectNotReady expected last sequence per subject temporarily unavailable
 	JSStreamExpectedLastSeqPerSubjectNotReady ErrorIdentifier = 10163
 
@@ -724,6 +727,7 @@ var (
 		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"},
+		JSStreamExpectedLastSeqPerSubjectInvalid:   {Code: 400, ErrCode: 10193, Description: "missing sequence for expected last sequence per subject"},
 		JSStreamExpectedLastSeqPerSubjectNotReady:  {Code: 503, ErrCode: 10163, Description: "expected last sequence per subject temporarily unavailable"},
 		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}"},
@@ -2383,6 +2387,16 @@ func NewJSStreamDuplicateMessageConflictError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSStreamDuplicateMessageConflict]
 }
 
+// NewJSStreamExpectedLastSeqPerSubjectInvalidError creates a new JSStreamExpectedLastSeqPerSubjectInvalid error: "missing sequence for expected last sequence per subject"
+func NewJSStreamExpectedLastSeqPerSubjectInvalidError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSStreamExpectedLastSeqPerSubjectInvalid]
+}
+
 // NewJSStreamExpectedLastSeqPerSubjectNotReadyError creates a new JSStreamExpectedLastSeqPerSubjectNotReady error: "expected last sequence per subject temporarily unavailable"
 func NewJSStreamExpectedLastSeqPerSubjectNotReadyError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)

server/jetstream_test.go

@@ -9173,8 +9173,8 @@ func TestJetStreamLastSequenceBySubject(t *testing.T) {
 }
 
 func TestJetStreamLastSequenceBySubjectWithSubject(t *testing.T) {
-	for _, st := range []StorageType{FileStorage, MemoryStorage} {
-		t.Run(st.String(), func(t *testing.T) {
+	test := func(replicas int, st StorageType) {
+		t.Run(fmt.Sprintf("R%d/%s", replicas, st), func(t *testing.T) {
 			c := createJetStreamClusterExplicit(t, "JSC", 3)
 			defer c.shutdown()
 
@@ -9185,7 +9185,7 @@ func TestJetStreamLastSequenceBySubjectWithSubject(t *testing.T) {
 				Name:       "KV",
 				Subjects:   []string{"kv.>"},
 				Storage:    st,
-				Replicas:   3,
+				Replicas:   replicas,
 				MaxMsgsPer: 1,
 			}
 
@@ -9262,8 +9262,21 @@ func TestJetStreamLastSequenceBySubjectWithSubject(t *testing.T) {
 			pubAndCheck("kv.xxx", "kv.*.*", "0", false)   // Last is 11 for kv.xxx; 11 for kv.*.*;
 			pubAndCheck("kv.3.xxx", "kv.3.*", "4", true)  // Last is 12 for kv.3.xxx; 12 for kv.3.*;
 			pubAndCheck("kv.3.xyz", "kv.3.*", "12", true) // Last is 13 for kv.3.xyz; 13 for kv.3.*;
+
+			// When using the last-subj-seq-subj header, but the sequence header is missing.
+			m = nats.NewMsg("kv.invalid")
+			m.Data = []byte("HELLO")
+			m.Header.Set(JSExpectedLastSubjSeqSubj, "kv.invalid")
+			_, err = js.PublishMsg(m)
+			require_Error(t, err, NewJSStreamExpectedLastSeqPerSubjectInvalidError())
 		})
 	}
+
+	for _, replicas := range []int{1, 3} {
+		for _, st := range []StorageType{FileStorage, MemoryStorage} {
+			test(replicas, st)
+		}
+	}
 }
 
 func TestJetStreamFilteredConsumersWithWiderFilter(t *testing.T) {

server/stream.go

@@ -4587,7 +4587,7 @@ func getExpectedLastSeqPerSubject(hdr []byte) (uint64, bool) {
 
 // Fast lookup of expected subject for the expected stream sequence per subject.
 func getExpectedLastSeqPerSubjectForSubject(hdr []byte) string {
-	return string(getHeader(JSExpectedLastSubjSeqSubj, hdr))
+	return bytesToString(sliceHeader(JSExpectedLastSubjSeqSubj, hdr))
 }
 
 // Fast lookup of the message TTL from headers:
@@ -5446,6 +5446,16 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 					}
 					return fmt.Errorf("last sequence by subject mismatch: %d vs %d", seq, fseq)
 				}
+			} else if getExpectedLastSeqPerSubjectForSubject(hdr) != _EMPTY_ {
+				mset.mu.Unlock()
+				apiErr := NewJSStreamExpectedLastSeqPerSubjectInvalidError()
+				if canRespond {
+					resp.PubAck = &PubAck{Stream: name}
+					resp.Error = apiErr
+					b, _ := json.Marshal(resp)
+					outq.sendMsg(reply, b)
+				}
+				return apiErr
 			}
 
 			// Expected last sequence.