Ability to set the minimum value for a seqset if known. (#4075)

Signed-off-by: Derek Collison <derek@nats.io>

server/avl/seqset.go

@@ -62,17 +62,34 @@ func (ss *SequenceSet) Exists(seq uint64) bool {
 	return false
 }
 
+// SetInitialMin should be used to set the initial minimum sequence when known.
+// This will more effectively utilize space versus self selecting.
+// The set should be empty.
+func (ss *SequenceSet) SetInitialMin(min uint64) error {
+	if !ss.IsEmpty() {
+		return ErrSetNotEmpty
+	}
+	ss.root, ss.nodes = &node{base: min, h: 1}, 1
+	return nil
+}
+
 // Delete will remove the sequence from the set.
-// Wil optionally remove nodes and rebalance.
-func (ss *SequenceSet) Delete(seq uint64) {
+// Will optionally remove nodes and rebalance.
+// Returns where the sequence was set.
+func (ss *SequenceSet) Delete(seq uint64) bool {
 	if ss == nil || ss.root == nil {
-		return
+		return false
 	}
 	ss.root = ss.root.delete(seq, &ss.changed, &ss.nodes)
 	if ss.changed {
 		ss.changed = false
 		ss.size--
+		if ss.size == 0 {
+			ss.Empty()
+		}
+		return true
 	}
+	return false
 }
 
 // Size returns the number of items in the set.
@@ -228,7 +245,10 @@ func (ss SequenceSet) Encode(buf []byte) ([]byte, error) {
 }
 
 // ErrBadEncoding is returned when we can not decode properly.
-var ErrBadEncoding = errors.New("ss: bad encoding")
+var (
+	ErrBadEncoding = errors.New("ss: bad encoding")
+	ErrSetNotEmpty = errors.New("ss: set not empty")
+)
 
 func Decode(buf []byte) (*SequenceSet, error) {
 	if len(buf) < minLen || buf[0] != magic || buf[1] != version {

server/avl/seqset_test.go

@@ -218,6 +218,23 @@ func TestSeqSetUnion(t *testing.T) {
 	}
 }
 
+func TestSeqSetFirst(t *testing.T) {
+	var ss SequenceSet
+
+	seqs := []uint64{22, 222, 2222, 222_222}
+	for _, seq := range seqs {
+		// Normal case where we pick first/base.
+		ss.Insert(seq)
+		require_True(t, ss.root.base == (seq/numEntries)*numEntries)
+		ss.Empty()
+		// Where we set the minimum start value.
+		ss.SetInitialMin(seq)
+		ss.Insert(seq)
+		require_True(t, ss.root.base == seq)
+		ss.Empty()
+	}
+}
+
 func require_NoError(t *testing.T, err error) {
 	t.Helper()
 	if err != nil {

server/filestore.go

@@ -2817,6 +2817,10 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 			}
 		}
 	} else if !isEmpty {
+		if mb.dmap.IsEmpty() {
+			// Mark initial base for delete set.
+			mb.dmap.SetInitialMin(mb.first.seq)
+		}
 		// Out of order delete.
 		mb.dmap.Insert(seq)
 		// Check if <25% utilization and minimum size met.