WIP: fast batch reply

Signed-off-by: Maurice van Veen <[email protected]>

Author: MauriceVanVeen

server/errors.json

@@ -2010,10 +2010,10 @@
     "deprecates": ""
   },
   {
-    "constant": "JSBatchPublishMissingSeqErr",
+    "constant": "JSBatchPublishInvalidPatternErr",
     "code": 400,
     "error_code": 10203,
-    "description": "batch publish sequence is missing",
+    "description": "batch publish pattern is invalid",
     "comment": "",
     "help": "",
     "url": "",

server/jetstream.go

@@ -1458,7 +1458,6 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 				smv           StoreMsg
 				batchId       string
 				batchSeq      uint64
-				atomic        bool
 				commit        bool
 				commitEob     bool
 				batchStoreDir string
@@ -1470,10 +1469,10 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 			if err != nil || sm == nil {
 				goto SKIP
 			}
-			batchId, atomic = getBatchId(sm.hdr)
+			batchId = getBatchId(sm.hdr)
 			batchSeq, ok = getBatchSequence(sm.hdr)
 			commit = len(sliceHeader(JSBatchCommit, sm.hdr)) != 0
-			if batchId == _EMPTY_ || !atomic || !ok || commit {
+			if batchId == _EMPTY_ || !ok || commit {
 				goto SKIP
 			}
 			// We've observed a partial batch write. Write the remainder of the batch.

server/jetstream_batching_test.go

@@ -2902,15 +2902,24 @@ func TestJetStreamFastBatchPublish(t *testing.T) {
 		_, err := jsStreamCreate(t, nc, cfg)
 		require_NoError(t, err)
 
-		reply := nats.NewInbox()
-		sub, err := nc.SubscribeSync(reply)
+		inbox := nats.NewInbox()
+		generateReply := func(batchId string, batchSeq uint64, flow uint64, gap string, v int) string {
+			return fmt.Sprintf("%s.%d.%s.%s.%d.%d.$FI", inbox, flow, gap, batchId, batchSeq, v)
+		}
+		generateNormalReply := func(batchId string, batchSeq uint64, gap string) string {
+			return generateReply(batchId, batchSeq, 0, gap, 0)
+		}
+		generateCommitReply := func(batchId string, batchSeq uint64, gap string) string {
+			return generateReply(batchId, batchSeq, 0, gap, 1)
+		}
+
+		sub, err := nc.SubscribeSync(fmt.Sprintf("%s.>", inbox))
 		require_NoError(t, err)
 		defer sub.Drain()
 
 		m := nats.NewMsg("foo.0")
-		m.Reply = reply
+		m.Reply = generateNormalReply("uuid", 0, JSFastBatchGapFail)
 		m.Data = []byte("foo.0")
-		m.Header.Set("Nats-Fast-Batch-Id", "uuid")
 
 		// Publish with batch publish disabled.
 		require_NoError(t, nc.PublishMsg(m))
@@ -2925,17 +2934,18 @@ func TestJetStreamFastBatchPublish(t *testing.T) {
 		_, err = jsStreamUpdate(t, nc, cfg)
 		require_NoError(t, err)
 
-		// Publish without batch sequence errors.
+		// Publish with incorrect batch sequence errors.
+		m.Reply = generateNormalReply("uuid", 0, JSFastBatchGapFail)
 		require_NoError(t, nc.PublishMsg(m))
 		rmsg, err = sub.NextMsg(time.Second)
 		require_NoError(t, err)
 		pubAck = JSPubAckResponse{}
 		require_NoError(t, json.Unmarshal(rmsg.Data, &pubAck))
-		require_Error(t, pubAck.Error, NewJSBatchPublishMissingSeqError())
+		require_Error(t, pubAck.Error, NewJSBatchPublishInvalidPatternError())
 
 		// A batch ID must not exceed the maximum length.
 		longBatchId := strings.Repeat("A", 65)
-		m.Header.Set("Nats-Fast-Batch-Id", longBatchId)
+		m.Reply = generateNormalReply(longBatchId, 1, JSFastBatchGapFail)
 		require_NoError(t, nc.PublishMsg(m))
 		rmsg, err = sub.NextMsg(time.Second)
 		require_NoError(t, err)
@@ -2945,8 +2955,7 @@ func TestJetStreamFastBatchPublish(t *testing.T) {
 		require_Error(t, pubAck.Error, NewJSBatchPublishInvalidBatchIDError())
 
 		// Publish a batch, misses start.
-		m.Header.Set("Nats-Fast-Batch-Id", "uuid")
-		m.Header.Set("Nats-Batch-Sequence", "2")
+		m.Reply = generateNormalReply("uuid", 2, JSFastBatchGapFail)
 		require_NoError(t, nc.PublishMsg(m))
 		rmsg, err = sub.NextMsg(time.Second)
 		require_NoError(t, err)
@@ -2955,9 +2964,7 @@ func TestJetStreamFastBatchPublish(t *testing.T) {
 		require_Error(t, pubAck.Error, NewJSBatchPublishUnknownBatchIDError())
 
 		// Publish a "batch" which immediately commits.
-		m.Header.Set("Nats-Fast-Batch-Id", "uuid")
-		m.Header.Set("Nats-Batch-Sequence", "1")
-		m.Header.Set("Nats-Batch-Commit", "1")
+		m.Reply = generateCommitReply("uuid", 1, JSFastBatchGapFail)
 		require_NoError(t, nc.PublishMsg(m))
 		rmsg, err = sub.NextMsg(time.Second)
 		require_NoError(t, err)
@@ -2973,13 +2980,13 @@ func TestJetStreamFastBatchPublish(t *testing.T) {
 		require_Equal(t, pubAck.BatchSize, 1)
 
 		// Publish a batch of N messages.
-		m.Header.Del("Nats-Batch-Commit")
 		for seq, batch := uint64(1), uint64(5); seq <= batch; seq++ {
 			m.Subject = fmt.Sprintf("foo.%d", seq)
 			m.Data = []byte(m.Subject)
-			m.Header.Set("Nats-Batch-Sequence", strconv.FormatUint(seq, 10))
 			if seq == batch {
-				m.Header.Set("Nats-Batch-Commit", "1")
+				m.Reply = generateCommitReply("uuid", seq, JSFastBatchGapFail)
+			} else {
+				m.Reply = generateNormalReply("uuid", seq, JSFastBatchGapFail)
 			}
 			require_NoError(t, nc.PublishMsg(m))
 

server/jetstream_errors_generated.go

@@ -44,8 +44,8 @@ const (
 	// JSBatchPublishInvalidGapModeErr batch publish gap mode is invalid
 	JSBatchPublishInvalidGapModeErr ErrorIdentifier = 10206
 
-	// JSBatchPublishMissingSeqErr batch publish sequence is missing
-	JSBatchPublishMissingSeqErr ErrorIdentifier = 10203
+	// JSBatchPublishInvalidPatternErr batch publish pattern is invalid
+	JSBatchPublishInvalidPatternErr ErrorIdentifier = 10203
 
 	// JSBatchPublishUnknownBatchIDErr batch publish ID unknown
 	JSBatchPublishUnknownBatchIDErr ErrorIdentifier = 10205
@@ -639,7 +639,7 @@ var (
 		JSBatchPublishDisabledErr:                    {Code: 400, ErrCode: 10202, Description: "batch publish is disabled"},
 		JSBatchPublishInvalidBatchIDErr:              {Code: 400, ErrCode: 10204, Description: "batch publish ID is invalid"},
 		JSBatchPublishInvalidGapModeErr:              {Code: 400, ErrCode: 10206, Description: "batch publish gap mode is invalid"},
-		JSBatchPublishMissingSeqErr:                  {Code: 400, ErrCode: 10203, Description: "batch publish sequence is missing"},
+		JSBatchPublishInvalidPatternErr:              {Code: 400, ErrCode: 10203, Description: "batch publish pattern is invalid"},
 		JSBatchPublishUnknownBatchIDErr:              {Code: 400, ErrCode: 10205, Description: "batch publish ID unknown"},
 		JSClusterIncompleteErr:                       {Code: 503, ErrCode: 10004, Description: "incomplete results"},
 		JSClusterNoPeersErrF:                         {Code: 400, ErrCode: 10005, Description: "{err}"},
@@ -999,14 +999,14 @@ func NewJSBatchPublishInvalidGapModeError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSBatchPublishInvalidGapModeErr]
 }
 
-// NewJSBatchPublishMissingSeqError creates a new JSBatchPublishMissingSeqErr error: "batch publish sequence is missing"
-func NewJSBatchPublishMissingSeqError(opts ...ErrorOption) *ApiError {
+// NewJSBatchPublishInvalidPatternError creates a new JSBatchPublishInvalidPatternErr error: "batch publish pattern is invalid"
+func NewJSBatchPublishInvalidPatternError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
 	if ae, ok := eopts.err.(*ApiError); ok {
 		return ae
 	}
 
-	return ApiErrors[JSBatchPublishMissingSeqErr]
+	return ApiErrors[JSBatchPublishInvalidPatternErr]
 }
 
 // NewJSBatchPublishUnknownBatchIDError creates a new JSBatchPublishUnknownBatchIDErr error: "batch publish ID unknown"