(2.14) Reset consumer to new starting sequence

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

Author: MauriceVanVeen

server/consumer.go

@@ -436,11 +436,13 @@ type consumer struct {
 	lss               *lastSeqSkipList
 	rlimit            *rate.Limiter
 	reqSub            *subscription
+	resetSub          *subscription
 	ackSub            *subscription
 	ackReplyT         string
 	ackSubj           string
 	nextMsgSubj       string
 	nextMsgReqs       *ipQueue[*nextMsgReq]
+	resetSubj         string
 	maxp              int
 	pblimit           int
 	maxpb             int
@@ -1276,6 +1278,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	o.ackReplyT = fmt.Sprintf("%s.%%d.%%d.%%d.%%d.%%d", pre)
 	o.ackSubj = fmt.Sprintf("%s.*.*.*.*.*", pre)
 	o.nextMsgSubj = fmt.Sprintf(JSApiRequestNextT, mn, o.name)
+	o.resetSubj = fmt.Sprintf(JSApiConsumerResetT, mn, o.name)
 
 	// Check/update the inactive threshold
 	o.updateInactiveThreshold(&o.cfg)
@@ -1558,6 +1561,10 @@ func (o *consumer) setLeader(isLeader bool) {
 			o.mu.Unlock()
 			return
 		}
+		if o.resetSub, err = o.subscribeInternal(o.resetSubj, o.processResetReq); err != nil {
+			o.mu.Unlock()
+			return
+		}
 
 		// Check on flow control settings.
 		if o.cfg.FlowControl {
@@ -1677,8 +1684,9 @@ func (o *consumer) setLeader(isLeader bool) {
 		// ok if they are nil, we protect inside unsubscribe()
 		o.unsubscribe(o.ackSub)
 		o.unsubscribe(o.reqSub)
+		o.unsubscribe(o.resetSub)
 		o.unsubscribe(o.fcSub)
-		o.ackSub, o.reqSub, o.fcSub = nil, nil, nil
+		o.ackSub, o.reqSub, o.resetSub, o.fcSub = nil, nil, nil, nil
 		if o.infoSub != nil {
 			o.srv.sysUnsubscribe(o.infoSub)
 			o.infoSub = nil
@@ -2613,6 +2621,78 @@ func (o *consumer) updateSkipped(seq uint64) {
 	o.propose(b[:])
 }
 
+func (o *consumer) resetStartingSeq(seq uint64, reply string) (uint64, bool, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	// Reset to a specific sequence, or back to the ack floor.
+	if seq == 0 {
+		seq = o.asflr + 1
+	} else if o.cfg.DeliverPolicy == DeliverAll {
+		// Always allowed.
+		goto VALID
+	} else if o.cfg.DeliverPolicy == DeliverByStartSequence {
+		// Only allowed if not going below what's configured.
+		if seq < o.cfg.OptStartSeq {
+			return 0, false, errors.New("below start seq")
+		}
+		goto VALID
+	} else if o.cfg.DeliverPolicy == DeliverByStartTime && o.mset != nil {
+		// Only allowed if not going below what's configured.
+		nseq := o.mset.store.GetSeqFromTime(*o.cfg.OptStartTime)
+		if seq < nseq {
+			return 0, false, errors.New("below start time")
+		}
+		goto VALID
+	} else {
+		return 0, false, errors.New("not allowed")
+	}
+
+VALID:
+	// Must be a minimum of 1.
+	if seq <= 0 {
+		seq = 1
+	}
+	o.resetLocalStartingSeq(seq)
+	// Clustered mode and R>1.
+	if o.node != nil {
+		b := make([]byte, 1+8+len(reply))
+		b[0] = byte(resetSeqOp)
+		var le = binary.LittleEndian
+		le.PutUint64(b[1:], seq)
+		copy(b[1+8:], reply)
+		o.propose(b[:])
+		return seq, false, nil
+	} else if o.store != nil {
+		o.store.Reset(seq - 1)
+		// Cleanup messages that lost interest.
+		if o.retention == InterestPolicy {
+			if mset := o.mset; mset != nil {
+				o.mu.Unlock()
+				ss := mset.state()
+				o.checkStateForInterestStream(&ss)
+				o.mu.Lock()
+			}
+		}
+
+		// Recalculate pending, and re-trigger message delivery.
+		o.streamNumPending()
+		o.signalNewMessages()
+		return seq, true, nil
+	}
+	return seq, false, nil
+}
+
+// Lock should be held.
+func (o *consumer) resetLocalStartingSeq(seq uint64) {
+	o.pending, o.rdc = nil, nil
+	o.rdq = nil
+	o.rdqi.Empty()
+	o.sseq, o.dseq = seq, 1
+	o.adflr, o.asflr = o.dseq-1, o.sseq-1
+	o.ldt, o.lat = time.Time{}, time.Time{}
+}
+
 func (o *consumer) loopAndForwardProposals(qch chan struct{}) {
 	// On exit make sure we nil out pch.
 	defer func() {
@@ -4140,6 +4220,48 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 	o.nextMsgReqs.push(newNextMsgReq(reply, copyBytes(msg)))
 }
 
+// processResetReq will reset a consumer to a new starting sequence.
+func (o *consumer) processResetReq(_ *subscription, c *client, a *Account, _, reply string, rmsg []byte) {
+	if reply == _EMPTY_ {
+		return
+	}
+
+	s := o.srv
+	var resp = JSApiConsumerResetResponse{ApiResponse: ApiResponse{Type: JSApiConsumerResetResponseType}}
+
+	hdr, msg := c.msgParts(rmsg)
+	if errorOnRequiredApiLevel(hdr) {
+		resp.Error = NewJSRequiredApiLevelError()
+		s.sendInternalAccountMsg(a, reply, s.jsonResponse(&resp))
+		return
+	}
+
+	// An empty message resets back to the ack floor, otherwise a custom sequence is used.
+	var req JSApiConsumerResetRequest
+	if len(msg) > 0 {
+		if err := json.Unmarshal(msg, &req); err != nil {
+			resp.Error = NewJSInvalidJSONError(err)
+			s.sendInternalAccountMsg(a, reply, s.jsonResponse(&resp))
+			return
+		}
+		// Resetting to 0 is invalid.
+		if req.Seq == 0 {
+			resp.Error = NewJSInvalidJSONError(errors.New("reset to zero seq"))
+			s.sendInternalAccountMsg(a, reply, s.jsonResponse(&resp))
+			return
+		}
+	}
+	resetSeq, canRespond, err := o.resetStartingSeq(req.Seq, reply)
+	if err != nil {
+		resp.Error = NewJSConsumerInvalidResetError(err)
+		s.sendInternalAccountMsg(a, reply, s.jsonResponse(&resp))
+	} else if canRespond {
+		resp.ConsumerInfo = setDynamicConsumerInfoMetadata(o.info())
+		resp.ResetSeq = resetSeq
+		s.sendInternalAccountMsg(a, reply, s.jsonResponse(&resp))
+	}
+}
+
 func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
 	o.mu.Lock()
 	defer o.mu.Unlock()
@@ -6096,9 +6218,11 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 	o.active = false
 	o.unsubscribe(o.ackSub)
 	o.unsubscribe(o.reqSub)
+	o.unsubscribe(o.resetSub)
 	o.unsubscribe(o.fcSub)
 	o.ackSub = nil
 	o.reqSub = nil
+	o.resetSub = nil
 	o.fcSub = nil
 	if o.infoSub != nil {
 		o.srv.sysUnsubscribe(o.infoSub)

server/errors.json

@@ -2018,5 +2018,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSConsumerInvalidResetErr",
+    "code": 400,
+    "error_code": 10204,
+    "description": "invalid reset: {err}",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]

server/filestore.go

@@ -11658,6 +11658,7 @@ func (o *consumerFileStore) flushLoop(fch, qch chan struct{}) {
 func (o *consumerFileStore) SetStarting(sseq uint64) error {
 	o.mu.Lock()
 	o.state.Delivered.Stream = sseq
+	o.state.AckFloor.Stream = sseq
 	buf, err := o.encodeState()
 	o.mu.Unlock()
 	if err != nil {
@@ -11682,6 +11683,14 @@ func (o *consumerFileStore) UpdateStarting(sseq uint64) {
 	o.kickFlusher()
 }
 
+// Reset all values in the store, and reset the starting sequence.
+func (o *consumerFileStore) Reset(sseq uint64) error {
+	o.mu.Lock()
+	o.state = ConsumerState{}
+	o.mu.Unlock()
+	return o.SetStarting(sseq)
+}
+
 // HasState returns if this store has a recorded state.
 func (o *consumerFileStore) HasState() bool {
 	o.mu.Lock()

server/jetstream_api.go

@@ -155,6 +155,9 @@ const (
 	// JSApiRequestNextT is the prefix for the request next message(s) for a consumer in worker/pull mode.
 	JSApiRequestNextT = "$JS.API.CONSUMER.MSG.NEXT.%s.%s"
 
+	// JSApiConsumerResetT is the prefix for resetting a given consumer to a new starting sequence.
+	JSApiConsumerResetT = "$JS.API.CONSUMER.RESET.%s.%s"
+
 	// JSApiConsumerUnpinT is the prefix for unpinning subscription for a given consumer.
 	JSApiConsumerUnpin  = "$JS.API.CONSUMER.UNPIN.*.*"
 	JSApiConsumerUnpinT = "$JS.API.CONSUMER.UNPIN.%s.%s"
@@ -757,6 +760,19 @@ type JSApiConsumerGetNextRequest struct {
 	PriorityGroup
 }
 
+// JSApiConsumerResetRequest is for resetting a consumer to a specific sequence.
+type JSApiConsumerResetRequest struct {
+	Seq uint64 `json:"seq"`
+}
+
+type JSApiConsumerResetResponse struct {
+	ApiResponse
+	*ConsumerInfo
+	ResetSeq uint64 `json:"reset_seq"`
+}
+
+const JSApiConsumerResetResponseType = "io.nats.jetstream.api.v1.consumer_reset_response"
+
 // Structure that holds state for a JetStream API request that is processed
 // in a separate long-lived go routine. This is to avoid blocking connections.
 type jsAPIRoutedReq struct {

server/jetstream_cluster.go

@@ -144,6 +144,8 @@ const (
 	// Batch stream ops.
 	batchMsgOp
 	batchCommitMsgOp
+	// Consumer rest to specific starting sequence.
+	resetSeqOp
 )
 
 // raftGroups are controlled by the metagroup controller.
@@ -6127,6 +6129,39 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 					o.store.UpdateStarting(sseq - 1)
 				}
 				o.mu.Unlock()
+			case resetSeqOp:
+				o.mu.Lock()
+				var le = binary.LittleEndian
+				sseq := le.Uint64(buf[1:9])
+				reply := string(buf[9:])
+				o.resetLocalStartingSeq(sseq)
+				if o.store != nil {
+					o.store.Reset(sseq - 1)
+				}
+				// Cleanup messages that lost interest.
+				if o.retention == InterestPolicy {
+					if mset := o.mset; mset != nil {
+						o.mu.Unlock()
+						ss := mset.state()
+						o.checkStateForInterestStream(&ss)
+						o.mu.Lock()
+					}
+				}
+				// Recalculate pending, and re-trigger message delivery.
+				if !o.isLeader() {
+					o.mu.Unlock()
+				} else {
+					o.streamNumPending()
+					o.signalNewMessages()
+					s, a := o.srv, o.acc
+					o.mu.Unlock()
+					if reply != _EMPTY_ {
+						var resp = JSApiConsumerResetResponse{ApiResponse: ApiResponse{Type: JSApiConsumerResetResponseType}}
+						resp.ConsumerInfo = setDynamicConsumerInfoMetadata(o.info())
+						resp.ResetSeq = sseq
+						s.sendInternalAccountMsg(a, reply, s.jsonResponse(&resp))
+					}
+				}
 			case addPendingRequest:
 				o.mu.Lock()
 				if !o.isLeader() {