Skip to content

Commit

Permalink
Common store test for LoadNextMsgMulti
Browse files Browse the repository at this point in the history
It concerns me a little that by duplicating the same test for both
file and memory stores, that differences could slip in later that
mean we are no longer testing for consistency between the two stores.

This approach allows us to write tests that only use the `StreamStore`
API and have them tested against all store permutations.

Signed-off-by: Neil Twigg <[email protected]>
  • Loading branch information
neilalexander committed Apr 4, 2024
1 parent 3170cc9 commit 1433ee3
Show file tree
Hide file tree
Showing 3 changed files with 111 additions and 125 deletions.
61 changes: 0 additions & 61 deletions server/filestore_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -6755,67 +6755,6 @@ func TestFileStoreMsgBlockFirstAndLastSeqCorrupt(t *testing.T) {
require_Equal(t, lseq, 10)
}

func TestFileStoreMsgLoadNextMsgMulti(t *testing.T) {
fs, err := newFileStore(
FileStoreConfig{StoreDir: t.TempDir()},
StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
require_NoError(t, err)
defer fs.Stop()

// Put 1k msgs in
for i := 0; i < 1000; i++ {
subj := fmt.Sprintf("foo.%d", i)
fs.StoreMsg(subj, nil, []byte("ZZZ"))
}

var smv StoreMsg
// Do multi load next with 1 wc entry.
sl := NewSublistWithCache()
sl.Insert(&subscription{subject: []byte("foo.>")})
for i, seq := 0, uint64(1); i < 1000; i++ {
sm, nseq, err := fs.LoadNextMsgMulti(sl, seq, &smv)
require_NoError(t, err)
require_True(t, sm.subj == fmt.Sprintf("foo.%d", i))
require_Equal(t, nseq, seq)
seq++
}

// Now do multi load next with 1000 literal subjects.
sl = NewSublistWithCache()
for i := 0; i < 1000; i++ {
subj := fmt.Sprintf("foo.%d", i)
sl.Insert(&subscription{subject: []byte(subj)})
}
for i, seq := 0, uint64(1); i < 1000; i++ {
sm, nseq, err := fs.LoadNextMsgMulti(sl, seq, &smv)
require_NoError(t, err)
require_True(t, sm.subj == fmt.Sprintf("foo.%d", i))
require_Equal(t, nseq, seq)
seq++
}

// Check that we can pull out 3 individuals.
sl = NewSublistWithCache()
sl.Insert(&subscription{subject: []byte("foo.2")})
sl.Insert(&subscription{subject: []byte("foo.222")})
sl.Insert(&subscription{subject: []byte("foo.999")})
sm, seq, err := fs.LoadNextMsgMulti(sl, 1, &smv)
require_NoError(t, err)
require_True(t, sm.subj == "foo.2")
require_Equal(t, seq, 3)
sm, seq, err = fs.LoadNextMsgMulti(sl, seq+1, &smv)
require_NoError(t, err)
require_True(t, sm.subj == "foo.222")
require_Equal(t, seq, 223)
sm, seq, err = fs.LoadNextMsgMulti(sl, seq+1, &smv)
require_NoError(t, err)
require_True(t, sm.subj == "foo.999")
require_Equal(t, seq, 1000)
_, seq, err = fs.LoadNextMsgMulti(sl, seq+1, &smv)
require_Error(t, err)
require_Equal(t, seq, 1000)
}

///////////////////////////////////////////////////////////////////////////
// Benchmarks
///////////////////////////////////////////////////////////////////////////
Expand Down
64 changes: 0 additions & 64 deletions server/memstore_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1051,67 +1051,3 @@ func TestMemStorePurgeExWithDeletedMsgs(t *testing.T) {
require_Equal(t, state.LastSeq, 10)
require_Equal(t, state.Msgs, 1)
}

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

// Put 1k msgs in
for i := 0; i < 1000; i++ {
subj := fmt.Sprintf("foo.%d", i)
ms.StoreMsg(subj, nil, []byte("ZZZ"))
}

var smv StoreMsg
// Do multi load next with 1 wc entry.
sl := NewSublistWithCache()
sl.Insert(&subscription{subject: []byte("foo.>")})
for i, seq := 0, uint64(1); i < 1000; i++ {
sm, nseq, err := ms.LoadNextMsgMulti(sl, seq, &smv)
require_NoError(t, err)
require_True(t, sm.subj == fmt.Sprintf("foo.%d", i))
require_Equal(t, nseq, seq)
seq++
}

// Now do multi load next with 1000 literal subjects.
sl = NewSublistWithCache()
for i := 0; i < 1000; i++ {
subj := fmt.Sprintf("foo.%d", i)
sl.Insert(&subscription{subject: []byte(subj)})
}
for i, seq := 0, uint64(1); i < 1000; i++ {
sm, nseq, err := ms.LoadNextMsgMulti(sl, seq, &smv)
require_NoError(t, err)
require_True(t, sm.subj == fmt.Sprintf("foo.%d", i))
require_Equal(t, nseq, seq)
seq++
}

// Check that we can pull out 3 individuals.
sl = NewSublistWithCache()
sl.Insert(&subscription{subject: []byte("foo.2")})
sl.Insert(&subscription{subject: []byte("foo.222")})
sl.Insert(&subscription{subject: []byte("foo.999")})
sm, seq, err := ms.LoadNextMsgMulti(sl, 1, &smv)
require_NoError(t, err)
require_True(t, sm.subj == "foo.2")
require_Equal(t, seq, 3)
sm, seq, err = ms.LoadNextMsgMulti(sl, seq+1, &smv)
require_NoError(t, err)
require_True(t, sm.subj == "foo.222")
require_Equal(t, seq, 223)
sm, seq, err = ms.LoadNextMsgMulti(sl, seq+1, &smv)
require_NoError(t, err)
require_True(t, sm.subj == "foo.999")
require_Equal(t, seq, 1000)
_, seq, err = ms.LoadNextMsgMulti(sl, seq+1, &smv)
require_Error(t, err)
require_Equal(t, seq, 1000)
}
111 changes: 111 additions & 0 deletions server/store_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
// Copyright 2012-2024 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
// +build !skip_store_tests

package server

import (
"fmt"
"testing"
)

func testAllStoreAllPermutations(t *testing.T, compressionAndEncryption bool, cfg StreamConfig, fn func(t *testing.T, fs StreamStore)) {
t.Run("Memory", func(t *testing.T) {
cfg.Storage = MemoryStorage
fs, err := newMemStore(&cfg)
require_NoError(t, err)
defer fs.Stop()
fn(t, fs)
})
t.Run("File", func(t *testing.T) {
cfg.Storage = FileStorage
if compressionAndEncryption {
testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
fs, err := newFileStore(fcfg, cfg)
require_NoError(t, err)
defer fs.Stop()
fn(t, fs)
})
} else {
fs, err := newFileStore(FileStoreConfig{
StoreDir: t.TempDir(),
}, cfg)
require_NoError(t, err)
defer fs.Stop()
fn(t, fs)
}
})
}

func TestStoreMsgLoadNextMsgMulti(t *testing.T) {
testAllStoreAllPermutations(
t, false,
StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}},
func(t *testing.T, fs StreamStore) {
// Put 1k msgs in
for i := 0; i < 1000; i++ {
subj := fmt.Sprintf("foo.%d", i)
fs.StoreMsg(subj, nil, []byte("ZZZ"))
}

var smv StoreMsg
// Do multi load next with 1 wc entry.
sl := NewSublistWithCache()
sl.Insert(&subscription{subject: []byte("foo.>")})
for i, seq := 0, uint64(1); i < 1000; i++ {
sm, nseq, err := fs.LoadNextMsgMulti(sl, seq, &smv)
require_NoError(t, err)
require_Equal(t, sm.subj, fmt.Sprintf("foo.%d", i))
require_Equal(t, nseq, seq)
seq++
}

// Now do multi load next with 1000 literal subjects.
sl = NewSublistWithCache()
for i := 0; i < 1000; i++ {
subj := fmt.Sprintf("foo.%d", i)
sl.Insert(&subscription{subject: []byte(subj)})
}
for i, seq := 0, uint64(1); i < 1000; i++ {
sm, nseq, err := fs.LoadNextMsgMulti(sl, seq, &smv)
require_NoError(t, err)
require_Equal(t, sm.subj, fmt.Sprintf("foo.%d", i))
require_Equal(t, nseq, seq)
seq++
}

// Check that we can pull out 3 individuals.
sl = NewSublistWithCache()
sl.Insert(&subscription{subject: []byte("foo.2")})
sl.Insert(&subscription{subject: []byte("foo.222")})
sl.Insert(&subscription{subject: []byte("foo.999")})
sm, seq, err := fs.LoadNextMsgMulti(sl, 1, &smv)
require_NoError(t, err)
require_Equal(t, sm.subj, "foo.2")
require_Equal(t, seq, 3)
sm, seq, err = fs.LoadNextMsgMulti(sl, seq+1, &smv)
require_NoError(t, err)
require_Equal(t, sm.subj, "foo.222")
require_Equal(t, seq, 223)
sm, seq, err = fs.LoadNextMsgMulti(sl, seq+1, &smv)
require_NoError(t, err)
require_Equal(t, sm.subj, "foo.999")
require_Equal(t, seq, 1000)
_, seq, err = fs.LoadNextMsgMulti(sl, seq+1, &smv)
require_Error(t, err)
require_Equal(t, seq, 1000)
},
)
}

0 comments on commit 1433ee3

Please sign in to comment.