memstore_test.go

// Copyright 2019-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//go:build !skip_store_tests

package server

import (
"bytes"
"errors"
"fmt"
"math/rand"
"reflect"
"slices"
"testing"
"time"

"github.com/nats-io/nuid"
)

funcTestMemStoreBasics(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

subj, msg:="foo", []byte("Hello World")
now:=time.Now().UnixNano()
ifseq, ts, err:=ms.StoreMsg(subj, nil, msg, 0); err!=nil {
t.Fatalf("Error storing msg: %v", err)
 } elseifseq!=1 {
t.Fatalf("Expected sequence to be 1, got %d", seq)
 } elseifts<now||ts>now+int64(time.Millisecond) {
t.Fatalf("Expected timestamp to be current, got %v", ts-now)
 }

state:=ms.State()
ifstate.Msgs!=1 {
t.Fatalf("Expected 1 msg, got %d", state.Msgs)
 }
expectedSize:=memStoreMsgSize(subj, nil, msg)
ifstate.Bytes!=expectedSize {
t.Fatalf("Expected %d bytes, got %d", expectedSize, state.Bytes)
 }
sm, err:=ms.LoadMsg(1, nil)
iferr!=nil {
t.Fatalf("Unexpected error looking up msg: %v", err)
 }
ifsm.subj!=subj {
t.Fatalf("Subjects don't match, original %q vs %q", subj, sm.subj)
 }
if!bytes.Equal(sm.msg, msg) {
t.Fatalf("Msgs don't match, original %q vs %q", msg, sm.msg)
 }
}

funcTestMemStoreMsgLimit(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage, MaxMsgs: 10})
require_NoError(t, err)
deferms.Stop()

subj, msg:="foo", []byte("Hello World")
fori:=0; i<10; i++ {
ms.StoreMsg(subj, nil, msg, 0)
 }
state:=ms.State()
ifstate.Msgs!=10 {
t.Fatalf("Expected %d msgs, got %d", 10, state.Msgs)
 }
if_, _, err:=ms.StoreMsg(subj, nil, msg, 0); err!=nil {
t.Fatalf("Error storing msg: %v", err)
 }
state=ms.State()
ifstate.Msgs!=10 {
t.Fatalf("Expected %d msgs, got %d", 10, state.Msgs)
 }
ifstate.LastSeq!=11 {
t.Fatalf("Expected the last sequence to be 11 now, but got %d", state.LastSeq)
 }
ifstate.FirstSeq!=2 {
t.Fatalf("Expected the first sequence to be 2 now, but got %d", state.FirstSeq)
 }
// Make sure we can not lookup seq 1.
if_, err:=ms.LoadMsg(1, nil); err==nil {
t.Fatalf("Expected error looking up seq 1 but got none")
 }
}

funcTestMemStoreBytesLimit(t*testing.T) {
subj, msg:="foo", make([]byte, 512)
storedMsgSize:=memStoreMsgSize(subj, nil, msg)

toStore:=uint64(1024)
maxBytes:=storedMsgSize*toStore

ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage, MaxBytes: int64(maxBytes)})
require_NoError(t, err)
deferms.Stop()

fori:=uint64(0); i<toStore; i++ {
ms.StoreMsg(subj, nil, msg, 0)
 }
state:=ms.State()
ifstate.Msgs!=toStore {
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 }
ifstate.Bytes!=storedMsgSize*toStore {
t.Fatalf("Expected bytes to be %d, got %d", storedMsgSize*toStore, state.Bytes)
 }

// Now send 10 more and check that bytes limit enforced.
fori:=0; i<10; i++ {
if_, _, err:=ms.StoreMsg(subj, nil, msg, 0); err!=nil {
t.Fatalf("Error storing msg: %v", err)
 }
 }
state=ms.State()
ifstate.Msgs!=toStore {
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 }
ifstate.Bytes!=storedMsgSize*toStore {
t.Fatalf("Expected bytes to be %d, got %d", storedMsgSize*toStore, state.Bytes)
 }
ifstate.FirstSeq!=11 {
t.Fatalf("Expected first sequence to be 11, got %d", state.FirstSeq)
 }
ifstate.LastSeq!=toStore+10 {
t.Fatalf("Expected last sequence to be %d, got %d", toStore+10, state.LastSeq)
 }
}

// https://github.com/nats-io/nats-server/issues/4771
funcTestMemStoreBytesLimitWithDiscardNew(t*testing.T) {
subj, msg:="tiny", make([]byte, 7)
storedMsgSize:=memStoreMsgSize(subj, nil, msg)

toStore:=uint64(3)
maxBytes:=100

ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage, MaxBytes: int64(maxBytes), Discard: DiscardNew})
require_NoError(t, err)
deferms.Stop()

// Now send 10 messages and check that bytes limit enforced.
fori:=0; i<10; i++ {
_, _, err:=ms.StoreMsg(subj, nil, msg, 0)
ifi<int(toStore) {
iferr!=nil {
t.Fatalf("Error storing msg: %v", err)
 }
 } elseif!errors.Is(err, ErrMaxBytes) {
t.Fatalf("Storing msg should result in: %v", ErrMaxBytes)
 }
 }
state:=ms.State()
ifstate.Msgs!=toStore {
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 }
ifstate.Bytes!=storedMsgSize*toStore {
t.Fatalf("Expected bytes to be %d, got %d", storedMsgSize*toStore, state.Bytes)
 }
}

funcTestMemStoreAgeLimit(t*testing.T) {
maxAge:=10*time.Millisecond
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage, MaxAge: maxAge})
require_NoError(t, err)
deferms.Stop()

// Store some messages. Does not really matter how many.
subj, msg:="foo", []byte("Hello World")
toStore:=100
fori:=0; i<toStore; i++ {
ms.StoreMsg(subj, nil, msg, 0)
 }
state:=ms.State()
ifstate.Msgs!=uint64(toStore) {
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 }
checkExpired:=func(t*testing.T) {
t.Helper()
checkFor(t, time.Second, maxAge, func() error {
state=ms.State()
ifstate.Msgs!=0 {
returnfmt.Errorf("Expected no msgs, got %d", state.Msgs)
 }
ifstate.Bytes!=0 {
returnfmt.Errorf("Expected no bytes, got %d", state.Bytes)
 }
returnnil
 })
 }
// Let them expire
checkExpired(t)
// Now add some more and make sure that timer will fire again.
fori:=0; i<toStore; i++ {
ms.StoreMsg(subj, nil, msg, 0)
 }
state=ms.State()
ifstate.Msgs!=uint64(toStore) {
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 }
checkExpired(t)
}

funcTestMemStoreTimeStamps(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

last:=time.Now().UnixNano()
subj, msg:="foo", []byte("Hello World")
fori:=0; i<10; i++ {
time.Sleep(5*time.Microsecond)
ms.StoreMsg(subj, nil, msg, 0)
 }
varsmvStoreMsg
forseq:=uint64(1); seq<=10; seq++ {
sm, err:=ms.LoadMsg(seq, &smv)
iferr!=nil {
t.Fatalf("Unexpected error looking up msg: %v", err)
 }
// These should be different
ifsm.ts<=last {
t.Fatalf("Expected different timestamps, got %v", sm.ts)
 }
last=sm.ts
 }
}

funcTestMemStorePurge(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

subj, msg:="foo", []byte("Hello World")
fori:=0; i<10; i++ {
ms.StoreMsg(subj, nil, msg, 0)
 }
ifstate:=ms.State(); state.Msgs!=10 {
t.Fatalf("Expected 10 msgs, got %d", state.Msgs)
 }
ms.Purge()
ifstate:=ms.State(); state.Msgs!=0 {
t.Fatalf("Expected no msgs, got %d", state.Msgs)
 }
}

funcTestMemStoreCompact(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

subj, msg:="foo", []byte("Hello World")
fori:=0; i<10; i++ {
ms.StoreMsg(subj, nil, msg, 0)
 }
ifstate:=ms.State(); state.Msgs!=10 {
t.Fatalf("Expected 10 msgs, got %d", state.Msgs)
 }
n, err:=ms.Compact(6)
iferr!=nil {
t.Fatalf("Unexpected error: %v", err)
 }
ifn!=5 {
t.Fatalf("Expected to have purged 5 msgs, got %d", n)
 }
state:=ms.State()
ifstate.Msgs!=5 {
t.Fatalf("Expected 5 msgs, got %d", state.Msgs)
 }
ifstate.FirstSeq!=6 {
t.Fatalf("Expected first seq of 6, got %d", state.FirstSeq)
 }
// Now test that compact will also reset first if seq > last
n, err=ms.Compact(100)
iferr!=nil {
t.Fatalf("Unexpected error: %v", err)
 }
ifn!=5 {
t.Fatalf("Expected to have purged 5 msgs, got %d", n)
 }
ifstate=ms.State(); state.FirstSeq!=100 {
t.Fatalf("Expected first seq of 100, got %d", state.FirstSeq)
 }
}

funcTestMemStoreEraseMsg(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

subj, msg:="foo", []byte("Hello World")
ms.StoreMsg(subj, nil, msg, 0)
sm, err:=ms.LoadMsg(1, nil)
iferr!=nil {
t.Fatalf("Unexpected error looking up msg: %v", err)
 }
if!bytes.Equal(msg, sm.msg) {
t.Fatalf("Expected same msg, got %q vs %q", sm.msg, msg)
 }
ifremoved, _:=ms.EraseMsg(1); !removed {
t.Fatalf("Expected erase msg to return success")
 }
}

funcTestMemStoreMsgHeaders(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

subj, hdr, msg:="foo", []byte("name:derek"), []byte("Hello World")
ifsz:=int(memStoreMsgSize(subj, hdr, msg)); sz!= (len(subj) +len(hdr) +len(msg) +16) {
t.Fatalf("Wrong size for stored msg with header")
 }
ms.StoreMsg(subj, hdr, msg, 0)
sm, err:=ms.LoadMsg(1, nil)
iferr!=nil {
t.Fatalf("Unexpected error looking up msg: %v", err)
 }
if!bytes.Equal(msg, sm.msg) {
t.Fatalf("Expected same msg, got %q vs %q", sm.msg, msg)
 }
if!bytes.Equal(hdr, sm.hdr) {
t.Fatalf("Expected same hdr, got %q vs %q", sm.hdr, hdr)
 }
ifremoved, _:=ms.EraseMsg(1); !removed {
t.Fatalf("Expected erase msg to return success")
 }
}

funcTestMemStoreStreamStateDeleted(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

subj, toStore:="foo", uint64(10)
fori:=uint64(1); i<=toStore; i++ {
msg:= []byte(fmt.Sprintf("[%08d] Hello World!", i))
if_, _, err:=ms.StoreMsg(subj, nil, msg, 0); err!=nil {
t.Fatalf("Error storing msg: %v", err)
 }
 }
state:=ms.State()
iflen(state.Deleted) !=0 {
t.Fatalf("Expected deleted to be empty")
 }
// Now remove some interior messages.
varexpected []uint64
forseq:=uint64(2); seq<toStore; seq+=2 {
ms.RemoveMsg(seq)
expected=append(expected, seq)
 }
state=ms.State()
if!reflect.DeepEqual(state.Deleted, expected) {
t.Fatalf("Expected deleted to be %+v, got %+v\n", expected, state.Deleted)
 }
// Now fill the gap by deleting 1 and 3
ms.RemoveMsg(1)
ms.RemoveMsg(3)
expected=expected[2:]
state=ms.State()
if!reflect.DeepEqual(state.Deleted, expected) {
t.Fatalf("Expected deleted to be %+v, got %+v\n", expected, state.Deleted)
 }
ifstate.FirstSeq!=5 {
t.Fatalf("Expected first seq to be 5, got %d", state.FirstSeq)
 }
ms.Purge()
ifstate=ms.State(); len(state.Deleted) !=0 {
t.Fatalf("Expected no deleted after purge, got %+v\n", state.Deleted)
 }
}

funcTestMemStoreStreamTruncate(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

tseq:=uint64(50)

subj, toStore:="foo", uint64(100)
fori:=uint64(1); i<tseq; i++ {
_, _, err:=ms.StoreMsg(subj, nil, []byte("ok"), 0)
require_NoError(t, err)
 }
subj="bar"
fori:=tseq; i<=toStore; i++ {
_, _, err:=ms.StoreMsg(subj, nil, []byte("ok"), 0)
require_NoError(t, err)
 }

ifstate:=ms.State(); state.Msgs!=toStore {
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 }

// Check that sequence has to be interior.
iferr:=ms.Truncate(toStore+1); err!=ErrInvalidSequence {
t.Fatalf("Expected err of '%v', got '%v'", ErrInvalidSequence, err)
 }

iferr:=ms.Truncate(tseq); err!=nil {
t.Fatalf("Unexpected error: %v", err)
 }
ifstate:=ms.State(); state.Msgs!=tseq {
t.Fatalf("Expected %d msgs, got %d", tseq, state.Msgs)
 }

// Now make sure we report properly if we have some deleted interior messages.
ms.RemoveMsg(10)
ms.RemoveMsg(20)
ms.RemoveMsg(30)
ms.RemoveMsg(40)

tseq=uint64(25)
iferr:=ms.Truncate(tseq); err!=nil {
t.Fatalf("Unexpected error: %v", err)
 }
state:=ms.State()
ifstate.Msgs!=tseq-2 {
t.Fatalf("Expected %d msgs, got %d", tseq-2, state.Msgs)
 }
ifstate.NumSubjects!=1 {
t.Fatalf("Expected only 1 subject, got %d", state.NumSubjects)
 }
expected:= []uint64{10, 20}
if!reflect.DeepEqual(state.Deleted, expected) {
t.Fatalf("Expected deleted to be %+v, got %+v\n", expected, state.Deleted)
 }
}

funcTestMemStorePurgeExWithSubject(t*testing.T) {
ms, err:=newMemStore(&StreamConfig{Storage: MemoryStorage})
require_NoError(t, err)
deferms.Stop()

fori:=0; i<100; i++ {
_, _, err=ms.StoreMsg("foo", nil, nil, 0)
require_NoError(t, err)
 }

// This should purge all.
ms.PurgeEx("foo", 1, 0)
require_True(t, ms.State().Msgs==0)
}

funcTestMemStoreUpdateMaxMsgsPerSubject(t*testing.T) {
cfg:=&StreamConfig{
Name: "TEST",
Storage: MemoryStorage,
Subjects: []string{"foo"},
MaxMsgsPer: 10,
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

// Make sure this is honored on an update.
cfg.MaxMsgsPer=50
err=ms.UpdateConfig(cfg)
require_NoError(t, err)

numStored:=22
fori:=0; i<numStored; i++ {
_, _, err=ms.StoreMsg("foo", nil, nil, 0)
require_NoError(t, err)
 }

ss:=ms.SubjectsState("foo")["foo"]
ifss.Msgs!=uint64(numStored) {
t.Fatalf("Expected to have %d stored, got %d", numStored, ss.Msgs)
 }

// Now make sure we trunk if setting to lower value.
cfg.MaxMsgsPer=10
err=ms.UpdateConfig(cfg)
require_NoError(t, err)

ss=ms.SubjectsState("foo")["foo"]
ifss.Msgs!=10 {
t.Fatalf("Expected to have %d stored, got %d", 10, ss.Msgs)
 }
}

funcTestMemStoreStreamTruncateReset(t*testing.T) {
cfg:=&StreamConfig{
Name: "TEST",
Storage: MemoryStorage,
Subjects: []string{"foo"},
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

subj, msg:="foo", []byte("Hello World")
fori:=0; i<1000; i++ {
_, _, err:=ms.StoreMsg(subj, nil, msg, 0)
require_NoError(t, err)
 }

// Reset everything
require_NoError(t, ms.Truncate(0))

state:=ms.State()
require_True(t, state.Msgs==0)
require_True(t, state.Bytes==0)
require_True(t, state.FirstSeq==0)
require_True(t, state.LastSeq==0)
require_True(t, state.NumSubjects==0)
require_True(t, state.NumDeleted==0)

fori:=0; i<1000; i++ {
_, _, err:=ms.StoreMsg(subj, nil, msg, 0)
require_NoError(t, err)
 }

state=ms.State()
require_True(t, state.Msgs==1000)
require_True(t, state.Bytes==30000)
require_True(t, state.FirstSeq==1)
require_True(t, state.LastSeq==1000)
require_True(t, state.NumSubjects==1)
require_True(t, state.NumDeleted==0)
}

funcTestMemStoreStreamCompactMultiBlockSubjectInfo(t*testing.T) {
cfg:=&StreamConfig{
Name: "TEST",
Storage: MemoryStorage,
Subjects: []string{"foo.*"},
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

fori:=0; i<1000; i++ {
subj:=fmt.Sprintf("foo.%d", i)
_, _, err:=ms.StoreMsg(subj, nil, []byte("Hello World"), 0)
require_NoError(t, err)
 }

// Compact such that we know we throw blocks away from the beginning.
deleted, err:=ms.Compact(501)
require_NoError(t, err)
require_True(t, deleted==500)

// Make sure we adjusted for subjects etc.
state:=ms.State()
require_True(t, state.NumSubjects==500)
}

funcTestMemStoreSubjectsTotals(t*testing.T) {
cfg:=&StreamConfig{
Name: "TEST",
Storage: MemoryStorage,
Subjects: []string{"*.*"},
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

fmap:=make(map[int]int)
bmap:=make(map[int]int)

varmmap[int]int
varftstring

fori:=0; i<10_000; i++ {
// Flip coin for prefix
ifrand.Intn(2) ==0 {
ft, m="foo", fmap
 } else {
ft, m="bar", bmap
 }
dt:=rand.Intn(100)
subj:=fmt.Sprintf("%s.%d", ft, dt)
m[dt]++

_, _, err:=ms.StoreMsg(subj, nil, []byte("Hello World"), 0)
require_NoError(t, err)
 }

// Now test SubjectsTotal
fordt, total:=rangefmap {
subj:=fmt.Sprintf("foo.%d", dt)
m:=ms.SubjectsTotals(subj)
ifm[subj] !=uint64(total) {
t.Fatalf("Expected %q to have %d total, got %d", subj, total, m[subj])
 }
 }

// Check fmap.
ifst:=ms.SubjectsTotals("foo.*"); len(st) !=len(fmap) {
t.Fatalf("Expected %d subjects for %q, got %d", len(fmap), "foo.*", len(st))
 } else {
expected:=0
for_, n:=rangefmap {
expected+=n
 }
received:=uint64(0)
for_, n:=rangest {
received+=n
 }
ifreceived!=uint64(expected) {
t.Fatalf("Expected %d total but got %d", expected, received)
 }
 }

// Check bmap.
ifst:=ms.SubjectsTotals("bar.*"); len(st) !=len(bmap) {
t.Fatalf("Expected %d subjects for %q, got %d", len(bmap), "bar.*", len(st))
 } else {
expected:=0
for_, n:=rangebmap {
expected+=n
 }
received:=uint64(0)
for_, n:=rangest {
received+=n
 }
ifreceived!=uint64(expected) {
t.Fatalf("Expected %d total but got %d", expected, received)
 }
 }

// All with pwc match.
ifst, expected:=ms.SubjectsTotals("*.*"), len(bmap)+len(fmap); len(st) !=expected {
t.Fatalf("Expected %d subjects for %q, got %d", expected, "*.*", len(st))
 }
}

funcTestMemStoreNumPending(t*testing.T) {
cfg:=&StreamConfig{
Name: "TEST",
Storage: MemoryStorage,
Subjects: []string{"*.*.*.*"},
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

tokens:= []string{"foo", "bar", "baz"}
genSubj:=func() string {
returnfmt.Sprintf("%s.%s.%s.%s",
tokens[rand.Intn(len(tokens))],
tokens[rand.Intn(len(tokens))],
tokens[rand.Intn(len(tokens))],
tokens[rand.Intn(len(tokens))],
 )
 }

fori:=0; i<50_000; i++ {
subj:=genSubj()
_, _, err:=ms.StoreMsg(subj, nil, []byte("Hello World"), 0)
require_NoError(t, err)
 }

state:=ms.State()

// Scan one by one for sanity check against other calculations.
sanityCheck:=func(ssequint64, filterstring) SimpleState {
t.Helper()
varssSimpleState
varsmvStoreMsg
// For here we know 0 is invalid, set to 1.
ifsseq==0 {
sseq=1
 }
forseq:=sseq; seq<=state.LastSeq; seq++ {
sm, err:=ms.LoadMsg(seq, &smv)
iferr!=nil {
t.Logf("Encountered error %v loading sequence: %d", err, seq)
continue
 }
ifsubjectIsSubsetMatch(sm.subj, filter) {
ss.Msgs++
ss.Last=seq
ifss.First==0||seq<ss.First {
ss.First=seq
 }
 }
 }
returnss
 }

check:=func(ssequint64, filterstring) {
t.Helper()
np, lvs:=ms.NumPending(sseq, filter, false)
ss:=ms.FilteredState(sseq, filter)
sss:=sanityCheck(sseq, filter)
iflvs!=state.LastSeq {
t.Fatalf("Expected NumPending to return valid through last of %d but got %d", state.LastSeq, lvs)
 }
ifss.Msgs!=np {
t.Fatalf("NumPending of %d did not match ss.Msgs of %d", np, ss.Msgs)
 }
ifss!=sss {
t.Fatalf("Failed sanity check, expected %+v got %+v", sss, ss)
 }
 }

sanityCheckLastOnly:=func(ssequint64, filterstring) SimpleState {
t.Helper()
varssSimpleState
varsmvStoreMsg
// For here we know 0 is invalid, set to 1.
ifsseq==0 {
sseq=1
 }
seen:=make(map[string]bool)
forseq:=state.LastSeq; seq>=sseq; seq-- {
sm, err:=ms.LoadMsg(seq, &smv)
iferr!=nil {
t.Logf("Encountered error %v loading sequence: %d", err, seq)
continue
 }
if!seen[sm.subj] &&subjectIsSubsetMatch(sm.subj, filter) {
ss.Msgs++
ifss.Last==0 {
ss.Last=seq
 }
ifss.First==0||seq<ss.First {
ss.First=seq
 }
seen[sm.subj] =true
 }
 }
returnss
 }

checkLastOnly:=func(ssequint64, filterstring) {
t.Helper()
np, lvs:=ms.NumPending(sseq, filter, true)
ss:=sanityCheckLastOnly(sseq, filter)
iflvs!=state.LastSeq {
t.Fatalf("Expected NumPending to return valid through last of %d but got %d", state.LastSeq, lvs)
 }
ifss.Msgs!=np {
t.Fatalf("NumPending of %d did not match ss.Msgs of %d", np, ss.Msgs)
 }
 }

startSeqs:= []uint64{0, 1, 2, 200, 444, 555, 2222, 8888, 12_345, 28_222, 33_456, 44_400, 49_999}
checkSubs:= []string{"foo.>", "*.bar.>", "foo.bar.*.baz", "*.bar.>", "*.foo.bar.*", "foo.foo.bar.baz"}

for_, filter:=rangecheckSubs {
for_, start:=rangestartSeqs {
check(start, filter)
checkLastOnly(start, filter)
 }
 }
}

funcTestMemStoreInitialFirstSeq(t*testing.T) {
cfg:=&StreamConfig{
Name: "zzz",
Storage: MemoryStorage,
FirstSeq: 1000,
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

seq, _, err:=ms.StoreMsg("A", nil, []byte("OK"), 0)
require_NoError(t, err)
ifseq!=1000 {
t.Fatalf("Message should have been sequence 1000 but was %d", seq)
 }

seq, _, err=ms.StoreMsg("B", nil, []byte("OK"), 0)
require_NoError(t, err)
ifseq!=1001 {
t.Fatalf("Message should have been sequence 1001 but was %d", seq)
 }

varstateStreamState
ms.FastState(&state)
switch {
casestate.Msgs!=2:
t.Fatalf("Expected 2 messages, got %d", state.Msgs)
casestate.FirstSeq!=1000:
t.Fatalf("Expected first seq 1000, got %d", state.FirstSeq)
casestate.LastSeq!=1001:
t.Fatalf("Expected last seq 1001, got %d", state.LastSeq)
 }
}

funcTestMemStoreDeleteBlocks(t*testing.T) {
cfg:=&StreamConfig{
Name: "zzz",
Subjects: []string{"*"},
Storage: MemoryStorage,
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

// Put in 10_000 msgs.
total:=10_000
fori:=0; i<total; i++ {
_, _, err:=ms.StoreMsg("A", nil, []byte("OK"), 0)
require_NoError(t, err)
 }

// Now pick 5k random sequences.
delete:=5000
deleteMap:=make(map[int]struct{}, delete)
forlen(deleteMap) <delete {
deleteMap[rand.Intn(total)+1] =struct{}{}
 }
// Now remove?
forseq:=rangedeleteMap {
ms.RemoveMsg(uint64(seq))
 }

varstateStreamState
ms.FastState(&state)

// For now we just track via one dmap.
ms.mu.RLock()
dmap:=ms.dmap.Clone()
ms.mu.RUnlock()

require_True(t, dmap.Size() ==state.NumDeleted)
}

// https://github.com/nats-io/nats-server/issues/4850
funcTestMemStoreGetSeqFromTimeWithLastDeleted(t*testing.T) {
cfg:=&StreamConfig{
Name: "zzz",
Subjects: []string{"*"},
Storage: MemoryStorage,
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

// Put in 1000 msgs.
total:=1000
varst time.Time
fori:=1; i<=total; i++ {
_, _, err:=ms.StoreMsg("A", nil, []byte("OK"), 0)
require_NoError(t, err)
ifi==total/2 {
time.Sleep(100*time.Millisecond)
st=time.Now()
 }
 }
// Delete last 100
forseq:=total-100; seq<=total; seq++ {
ms.RemoveMsg(uint64(seq))
 }

// Make sure this does not panic with last sequence no longer accessible.
seq:=ms.GetSeqFromTime(st)
// Make sure we get the right value.
require_Equal(t, seq, 501)
}

funcTestMemStoreSkipMsgs(t*testing.T) {
cfg:=&StreamConfig{
Name: "zzz",
Subjects: []string{"*"},
Storage: MemoryStorage,
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

// Test on empty FS first.
// Make sure wrong starting sequence fails.
err=ms.SkipMsgs(10, 100)
require_Error(t, err, ErrSequenceMismatch)

err=ms.SkipMsgs(1, 100)
require_NoError(t, err)

state:=ms.State()
require_Equal(t, state.FirstSeq, 101)
require_Equal(t, state.LastSeq, 100)

// Now add alot.
err=ms.SkipMsgs(101, 100_000)
require_NoError(t, err)
state=ms.State()
require_Equal(t, state.FirstSeq, 100_101)
require_Equal(t, state.LastSeq, 100_100)

// Now add in a message, and then skip to check dmap.
ms, err=newMemStore(cfg)
require_NoError(t, err)
ms.StoreMsg("foo", nil, nil, 0)

err=ms.SkipMsgs(2, 10)
require_NoError(t, err)
state=ms.State()
require_Equal(t, state.FirstSeq, 1)
require_Equal(t, state.LastSeq, 11)
require_Equal(t, state.Msgs, 1)
require_Equal(t, state.NumDeleted, 10)
require_Equal(t, len(state.Deleted), 10)

// Check Fast State too.
state.Deleted=nil
ms.FastState(&state)
require_Equal(t, state.FirstSeq, 1)
require_Equal(t, state.LastSeq, 11)
require_Equal(t, state.Msgs, 1)
require_Equal(t, state.NumDeleted, 10)
}

funcTestMemStoreMultiLastSeqs(t*testing.T) {
cfg:=&StreamConfig{
Name: "zzz",
Subjects: []string{"foo.*", "bar.*"},
Storage: MemoryStorage,
 }
ms, err:=newMemStore(cfg)
require_NoError(t, err)
deferms.Stop()

msg:= []byte("abc")
fori:=0; i<33; i++ {
ms.StoreMsg("foo.foo", nil, msg, 0)
ms.StoreMsg("foo.bar", nil, msg, 0)
ms.StoreMsg("foo.baz", nil, msg, 0)
 }
fori:=0; i<33; i++ {
ms.StoreMsg("bar.foo", nil, msg, 0)
ms.StoreMsg("bar.bar", nil, msg, 0)
ms.StoreMsg("bar.baz", nil, msg, 0)
 }

checkResults:=func(seqs, expected []uint64) {
t.Helper()
iflen(seqs) !=len(expected) {
t.Fatalf("Expected %+v got %+v", expected, seqs)
 }
fori:=rangeseqs {
ifseqs[i] !=expected[i] {
t.Fatalf("Expected %+v got %+v", expected, seqs)
 }
 }
 }

// UpTo sequence 3. Tests block split.
seqs, err:=ms.MultiLastSeqs([]string{"foo.*"}, 3, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{1, 2, 3})
// Up to last sequence of the stream.
seqs, err=ms.MultiLastSeqs([]string{"foo.*"}, 0, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{97, 98, 99})
// Check for bar.* at the end.
seqs, err=ms.MultiLastSeqs([]string{"bar.*"}, 0, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{196, 197, 198})
// This should find nothing.
seqs, err=ms.MultiLastSeqs([]string{"bar.*"}, 99, -1)
require_NoError(t, err)
checkResults(seqs, nil)

// Do multiple subjects explicitly.
seqs, err=ms.MultiLastSeqs([]string{"foo.foo", "foo.bar", "foo.baz"}, 3, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{1, 2, 3})
seqs, err=ms.MultiLastSeqs([]string{"foo.foo", "foo.bar", "foo.baz"}, 0, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{97, 98, 99})
seqs, err=ms.MultiLastSeqs([]string{"bar.foo", "bar.bar", "bar.baz"}, 0, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{196, 197, 198})
seqs, err=ms.MultiLastSeqs([]string{"bar.foo", "bar.bar", "bar.baz"}, 99, -1)
require_NoError(t, err)
checkResults(seqs, nil)

// Check single works
seqs, err=ms.MultiLastSeqs([]string{"foo.foo"}, 3, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{1})

// Now test that we properly de-duplicate between filters.
seqs, err=ms.MultiLastSeqs([]string{"foo.*", "foo.bar"}, 3, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{1, 2, 3})
seqs, err=ms.MultiLastSeqs([]string{"bar.>", "bar.bar", "bar.baz"}, 0, -1)
require_NoError(t, err)
checkResults(seqs, []uint64{196, 197, 198})

// All
seqs, err=ms.MultiLastSeqs([]string{">"}, 0, -1)