(2.14) [IMPROVED] Filestore error handling

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

Author: MauriceVanVeen

server/avl/norace_test.go

@@ -98,8 +98,7 @@ func TestNoRaceSeqSetEncodeLarge(t *testing.T) {
 	expected := time.Millisecond
 
 	start := time.Now()
-	b, err := ss.Encode(nil)
-	require_NoError(t, err)
+	b := ss.Encode(nil)
 
 	if elapsed := time.Since(start); elapsed > expected {
 		t.Fatalf("Expected encode of %d items with encoded size %v to take less than %v, got %v",

server/avl/seqset.go

@@ -239,7 +239,7 @@ func (ss SequenceSet) EncodeLen() int {
 	return minLen + (ss.Nodes() * ((numBuckets+1)*8 + 2))
 }
 
-func (ss SequenceSet) Encode(buf []byte) ([]byte, error) {
+func (ss SequenceSet) Encode(buf []byte) []byte {
 	nn, encLen := ss.Nodes(), ss.EncodeLen()
 
 	if cap(buf) < encLen {
@@ -268,7 +268,7 @@ func (ss SequenceSet) Encode(buf []byte) ([]byte, error) {
 		le.PutUint16(buf[i:], uint16(n.h))
 		i += 2
 	})
-	return buf[:i], nil
+	return buf[:i]
 }
 
 // ErrBadEncoding is returned when we can not decode properly.

server/consumer.go

@@ -1237,7 +1237,9 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		o.mu.Unlock()
 	} else {
 		// Select starting sequence number
-		o.selectStartingSeqNo()
+		if err := o.selectStartingSeqNo(); err != nil {
+			return nil, err
+		}
 	}
 
 	// Now register with mset and create the ack subscription.
@@ -5858,7 +5860,7 @@ func (o *consumer) hasSkipListPending() bool {
 }
 
 // Will select the starting sequence.
-func (o *consumer) selectStartingSeqNo() {
+func (o *consumer) selectStartingSeqNo() error {
 	if o.mset == nil || o.mset.store == nil {
 		o.sseq = 1
 	} else {
@@ -5873,7 +5875,10 @@ func (o *consumer) selectStartingSeqNo() {
 				} else {
 					// If we are partitioned here this will be properly set when we become leader.
 					for _, filter := range o.subjf {
-						ss := o.mset.store.FilteredState(1, filter.subject)
+						ss, err := o.mset.store.FilteredState(1, filter.subject)
+						if err != nil {
+							return err
+						}
 						if ss.Last > o.sseq {
 							o.sseq = ss.Last
 						}
@@ -5919,7 +5924,10 @@ func (o *consumer) selectStartingSeqNo() {
 					nseq := state.LastSeq
 					for _, filter := range o.subjf {
 						// Use first sequence since this is more optimized atm.
-						ss := o.mset.store.FilteredState(state.FirstSeq, filter.subject)
+						ss, err := o.mset.store.FilteredState(state.FirstSeq, filter.subject)
+						if err != nil {
+							return err
+						}
 						if ss.First >= o.sseq && ss.First < nseq {
 							nseq = ss.First
 						}
@@ -5961,8 +5969,11 @@ func (o *consumer) selectStartingSeqNo() {
 	// Set our starting sequence state.
 	// But only if we're not clustered, if clustered we propose upon becoming leader.
 	if o.store != nil && o.sseq > 0 && o.cfg.replicas(&o.mset.cfg) == 1 {
-		o.store.SetStarting(o.sseq - 1)
+		if err := o.store.SetStarting(o.sseq - 1); err != nil {
+			return err
+		}
 	}
+	return nil
 }
 
 // Test whether a config represents a durable subscriber.