(2.12) Read-after-write & monotonic reads (#6970)

See [ADR: JetStream
Read-after-Write](nats-io/nats-architecture-and-design#358)
for context, problem statement, and design.

Resolves #6557

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

Author: neilalexander

server/auth_callout.go

@@ -403,7 +403,7 @@ func (s *Server) processClientOrLeafCallout(c *client, opts *Options) (authorize
 		return false, errStr
 	}
 	req := []byte(b)
-	var hdr map[string]string
+	var hdr []byte
 
 	// Check if we have been asked to encrypt.
 	if xkp != nil {
@@ -413,7 +413,7 @@ func (s *Server) processClientOrLeafCallout(c *client, opts *Options) (authorize
 			s.Warnf(errStr)
 			return false, errStr
 		}
-		hdr = map[string]string{AuthRequestXKeyHeader: xkey}
+		hdr = genHeader(hdr, AuthRequestXKeyHeader, xkey)
 	}
 
 	// Send out our request.

server/consumer.go

@@ -114,6 +114,8 @@ type ConsumerConfig struct {
 	Replicas int `json:"num_replicas"`
 	// Force memory storage.
 	MemoryStorage bool `json:"mem_storage,omitempty"`
+	// Force the consumer to only deliver messages if the stream has at minimum this specified last sequence.
+	MinLastSeq uint64 `json:"min_last_seq,omitempty"`
 
 	// Don't add to general clients.
 	Direct bool `json:"direct,omitempty"`
@@ -4647,6 +4649,20 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			}
 		}
 
+		// If a minimum last sequence was specified, we need to check if the
+		// underlying stream has sufficient data. As an optimization, we only
+		// do this if what we want to deliver is below this floor.
+		if o.cfg.MinLastSeq > 0 && pmsg.seq < o.cfg.MinLastSeq {
+			var state StreamState
+			o.mset.store.FastState(&state)
+			if state.LastSeq < o.cfg.MinLastSeq {
+				// We only block deliveries at the start until we reach min last seq,
+				// so simply put back our pointer to account for the o.getNextMsg advancing it.
+				o.sseq--
+				goto waitForMsgs
+			}
+		}
+
 		// Update our cached num pending here first.
 		if dc == 1 {
 			o.npc--

server/errors.json

@@ -1778,5 +1778,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSStreamMinLastSeqErr",
+    "code": 412,
+    "error_code": 10180,
+    "description": "min last sequence",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]

server/events.go

@@ -420,7 +420,7 @@ type pubMsg struct {
 	sub  string
 	rply string
 	si   *ServerInfo
-	hdr  map[string]string
+	hdr  []byte
 	msg  any
 	oct  compressionType
 	echo bool
@@ -429,7 +429,7 @@ type pubMsg struct {
 
 var pubMsgPool sync.Pool
 
-func newPubMsg(c *client, sub, rply string, si *ServerInfo, hdr map[string]string,
+func newPubMsg(c *client, sub, rply string, si *ServerInfo, hdr []byte,
 	msg any, oct compressionType, echo, last bool) *pubMsg {
 
 	var m *pubMsg
@@ -604,17 +604,28 @@ RESET:
 				// Add in NL
 				b = append(b, _CRLF_...)
 
+				// Optional raw header addition.
+				if pm.hdr != nil {
+					b = append(pm.hdr, b...)
+					nhdr := len(pm.hdr)
+					nsize := len(b) - LEN_CR_LF
+					// MQTT producers don't have CRLF, so add it back.
+					if c.isMqtt() {
+						nsize += LEN_CR_LF
+					}
+					// Update pubArgs
+					// If others will use this later we need to save and restore original.
+					c.pa.hdr = nhdr
+					c.pa.size = nsize
+					c.pa.hdb = []byte(strconv.Itoa(nhdr))
+					c.pa.szb = []byte(strconv.Itoa(nsize))
+				}
+
 				// Check if we should set content-encoding
 				if contentHeader != _EMPTY_ {
 					b = c.setHeader(contentEncodingHeader, contentHeader, b)
 				}
 
-				// Optional header processing.
-				if pm.hdr != nil {
-					for k, v := range pm.hdr {
-						b = c.setHeader(k, v, b)
-					}
-				}
 				// Tracing
 				if trace {
 					c.traceInOp(fmt.Sprintf("PUB %s %s %d", c.pa.subject, c.pa.reply, c.pa.size), nil)
@@ -691,7 +702,7 @@ func (s *Server) sendInternalAccountMsg(a *Account, subject string, msg any) err
 }
 
 // Used to send an internal message with an optional reply to an arbitrary account.
-func (s *Server) sendInternalAccountMsgWithReply(a *Account, subject, reply string, hdr map[string]string, msg any, echo bool) error {
+func (s *Server) sendInternalAccountMsgWithReply(a *Account, subject, reply string, hdr []byte, msg any, echo bool) error {
 	s.mu.RLock()
 	if s.sys == nil || s.sys.sendq == nil {
 		s.mu.RUnlock()

server/jetstream_api.go

@@ -129,6 +129,10 @@ const (
 	JSDirectMsgGet  = "$JS.API.DIRECT.GET.*"
 	JSDirectMsgGetT = "$JS.API.DIRECT.GET.%s"
 
+	// JSDirectLeaderMsgGet is JSDirectLeaderMsgGetT but only answered by the current stream leader.
+	JSDirectLeaderMsgGet  = "$JS.API.DIRECT_LEADER.GET.*"
+	JSDirectLeaderMsgGetT = "$JS.API.DIRECT_LEADER.GET.%s"
+
 	// This is a direct version of get last by subject, which will be the dominant pattern for KV access once 2.9 is released.
 	// The stream and the key will be part of the subject to allow for no-marshal payloads and subject based security permissions.
 	JSDirectGetLastBySubject  = "$JS.API.DIRECT.GET.*.>"
@@ -675,7 +679,10 @@ type JSApiMsgGetRequest struct {
 	LastFor string `json:"last_by_subj,omitempty"`
 	NextFor string `json:"next_by_subj,omitempty"`
 
-	// Batch support. Used to request more then one msg at a time.
+	// Force the server to only deliver messages if the stream has at minimum this specified last sequence.
+	MinLastSeq uint64 `json:"min_last_seq,omitempty"`
+
+	// Batch support. Used to request more than one msg at a time.
 	// Can be used with simple starting seq, but also NextFor with wildcards.
 	Batch int `json:"batch,omitempty"`
 	// This will make sure we limit how much data we blast out. If not set we will
@@ -1057,9 +1064,11 @@ type delayedAPIResponse struct {
 	subject  string
 	reply    string
 	request  string
+	hdr      []byte
 	response string
 	rg       *raftGroup
 	deadline time.Time
+	noJs     bool
 	next     *delayedAPIResponse
 }
 
@@ -1162,7 +1171,12 @@ func (s *Server) delayedAPIResponder() {
 			next()
 		case <-tm.C:
 			if r != nil {
-				s.sendAPIErrResponse(r.ci, r.acc, r.subject, r.reply, r.request, r.response)
+				// If it's not a JS API error, send it as a raw response without additional API/audit tracking.
+				if r.noJs {
+					s.sendInternalAccountMsgWithReply(r.acc, r.subject, _EMPTY_, r.hdr, r.response, false)
+				} else {
+					s.sendAPIErrResponse(r.ci, r.acc, r.subject, r.reply, r.request, r.response)
+				}
 				pop()
 			}
 			next()
@@ -1172,7 +1186,13 @@ func (s *Server) delayedAPIResponder() {
 
 func (s *Server) sendDelayedAPIErrResponse(ci *ClientInfo, acc *Account, subject, reply, request, response string, rg *raftGroup, duration time.Duration) {
 	s.delayedAPIResponses.push(&delayedAPIResponse{
-		ci, acc, subject, reply, request, response, rg, time.Now().Add(duration), nil,
+		ci, acc, subject, reply, request, nil, response, rg, time.Now().Add(duration), false, nil,
+	})
+}
+
+func (s *Server) sendDelayedErrResponse(acc *Account, subject string, hdr []byte, response string, duration time.Duration) {
+	s.delayedAPIResponses.push(&delayedAPIResponse{
+		nil, acc, subject, _EMPTY_, _EMPTY_, hdr, response, nil, time.Now().Add(duration), true, nil,
 	})
 }
 
@@ -3467,6 +3487,16 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, _ *Account, subje
 		return
 	}
 
+	// Reject request if we can't guarantee the precondition of min last sequence.
+	if req.MinLastSeq > 0 && mset.lastSeq() < req.MinLastSeq {
+		// Even though only the leader is subscribed and will respond, we must delay the error.
+		// An old leader could think it's still leader, and it must not
+		// error sooner than the real leader can answer.
+		resp.Error = NewJSStreamMinLastSeqError()
+		s.sendDelayedAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp), nil, errRespDelay)
+		return
+	}
+
 	var svp StoreMsg
 	var sm *StoreMsg