hanzo/pubsub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[FIXED] Filestore only stores last block when MaxMsgsPerSubject 1 (#8254)

Browse Source 
Commit
https://github.com/nats-io/nats-server/commit/f8656fe8ea5472e7be24435017e5c7b4a27a174a
removed the `info.fblk = info.lblk` which was done by default when
hitting `info.total == 1`. This is correct, since if you'd remove the
last message in the last block, you wouldn't be able to find the first
message in the first block.

In 2.14 this fix was included with
https://github.com/nats-io/nats-server/commit/08275d90cfb844a148571e913ff7d3fc62e16a9e,
which ensures both `fblk` and `lblk` are stored in the `index.db` file.
However, since this fix was not included since 2.12.7, this had a
side-effect that the stale `fblk` was always included in `index.db`,
without it being changed to reflect latest writes (which the previous
`info.fblk = info.lblk` was in essence hiding). That would result in not
being able to find certain messages if not being written to frequently,
after a server restart that recovers from this stale index. This PR
returns that check, but only does so if `MaxMsgsPerSubject: 1`, since
only then can we be sure to do that optimization.

Also fix a bug where an update from `MaxMsgsPerSubject: -1` to `1` was
not enforced during the update.
This commit is contained in:
Neil
2026-06-02 09:46:44 +01:00
committed by GitHub
parent 6c6a79b9a4 c9c7f4dfc4
commit ce1ec22319
4 changed files with 149 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/filestore.go
+11 -7
View File
@@ -743,7 +743,7 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
fs.ageChkTime = 0
}
if fs.cfg.MaxMsgsPer > 0 && (old_cfg.MaxMsgsPer == 0 || fs.cfg.MaxMsgsPer < old_cfg.MaxMsgsPer) {
if fs.cfg.MaxMsgsPer > 0 && (old_cfg.MaxMsgsPer <= 0 || fs.cfg.MaxMsgsPer < old_cfg.MaxMsgsPer) {
if err := fs.enforceMsgPerSubjectLimit(true); err != nil {
fs.mu.Unlock()
return err
@@ -4930,15 +4930,17 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts, t
}
// Adjust top level tracking of per subject msg counts.
var info *psi
var ok bool
if len(subj) > 0 && fs.psim != nil {
index := fs.lmb.index
if info, ok := fs.psim.Find(stringToBytes(subj)); ok {
if info, ok = fs.psim.Find(stringToBytes(subj)); ok {
info.total++
if index > info.lblk {
info.lblk = index
}
} else {
fs.psim.Insert(stringToBytes(subj), psi{total: 1, fblk: index, lblk: index})
info, _ = fs.psim.Insert(stringToBytes(subj), psi{total: 1, fblk: index, lblk: index})
fs.tsl += len(subj)
}
}
@@ -4993,6 +4995,10 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts, t
}
}
}
// If we only ever store one/last message for a subject, can correct the first block to where we've just written.
if info != nil && info.total == 1 && mmp == 1 {
info.fblk = info.lblk
}
// Limits checks and enforcement.
// If they do any deletions they will update the
@@ -9005,10 +9011,8 @@ func (fs *fileStore) loadLastLocked(subj string, sm *StoreMsg) (lsm *StoreMsg, e
if ss.lastNeedsUpdate {
// mb is already loaded into the cache so should be fast-ish.
if err = mb.recalculateForSubj(subj, ss); err != nil {
if err != nil {
mb.mu.Unlock()
return nil, err
}
mb.mu.Unlock()
return nil, err
}
}
l = ss.Last
server/filestore_test.go
+59
View File
@@ -8583,6 +8583,65 @@ func Benchmark_FileStoreCreateConsumerStores(b *testing.B) {
}
}
func TestFileStoreMaxMsgsPerSubjectOneStaleFblkAfterRestart(t *testing.T) {
sd := t.TempDir()
fcfg := FileStoreConfig{StoreDir: sd, BlockSize: 256}
scfg := StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage, MaxMsgsPer: 1}
fs, err := newFileStore(fcfg, scfg)
require_NoError(t, err)
defer fs.Stop()
msg := []byte("hello")
// Store "foo.0" into the first block.
_, _, err = fs.StoreMsg("foo.0", nil, msg, 0)
require_NoError(t, err)
// Fill the first block with other subjects until a second block is created.
for i := 1; fs.numMsgBlocks() < 2; i++ {
_, _, err = fs.StoreMsg(fmt.Sprintf("foo.%d", i), nil, msg, 0)
require_NoError(t, err)
}
// Store "foo.0" again. The message lands in the second block, and MaxMsgsPer=1 removes
// the copy in the first block. lblk should now point to the last block.
seq, _, err := fs.StoreMsg("foo.0", nil, msg, 0)
require_NoError(t, err)
// Sanity check, while the in-memory lblk is correct, the last message is found.
var smv StoreMsg
sm, err := fs.LoadLastMsg("foo.0", &smv)
require_NoError(t, err)
require_Equal(t, sm.seq, seq)
// Stop writes the stream state file.
require_NoError(t, fs.Stop())
_, err = os.Stat(filepath.Join(sd, msgDir, streamStreamStateFile))
require_NoError(t, err)
// Restart, and recover from the stream state file. On 2.12.x recovery would set lblk to the
// stale fblk, both pointing at the first block.
fs, err = newFileStore(fcfg, scfg)
require_NoError(t, err)
defer fs.Stop()
// The last message for "foo.0" must still be found after a restart.
sm, err = fs.LoadLastMsg("foo.0", &smv)
require_NoError(t, err)
require_Equal(t, sm.subj, "foo.0")
require_Equal(t, sm.seq, seq)
// Validate internal subject state.
fs.mu.RLock()
defer fs.mu.RUnlock()
info, ok := fs.psim.Find(stringToBytes("foo.0"))
require_True(t, ok)
require_Equal(t, info.total, 1)
require_Equal(t, info.lblk, 2)
require_Equal(t, info.fblk, 2) // Since it's MaxMsgsPer:1, this should be optimized to last block.
}
func Benchmark_FileStoreSubjectStateConsistencyOptimizationPerf(b *testing.B) {
fs, err := newFileStore(
FileStoreConfig{StoreDir: b.TempDir()},
server/memstore.go
+1 -1
View File
@@ -123,7 +123,7 @@ func (ms *memStore) UpdateConfig(cfg *StreamConfig) error {
maxp := ms.maxp
ms.maxp = cfg.MaxMsgsPer
// If the value is smaller, or was unset before, we need to enforce that.
if ms.maxp > 0 && (maxp == 0 || ms.maxp < maxp) {
if ms.maxp > 0 && (maxp <= 0 || ms.maxp < maxp) {
lm := uint64(ms.maxp)
ms.fss.IterFast(func(subj []byte, ss *SimpleState) bool {
if ss.Msgs > lm {
server/store_test.go
+78
View File
@@ -447,6 +447,84 @@ func TestStoreMaxMsgsPerUpdateBug(t *testing.T) {
)
}
func TestStoreMaxMsgsPerUpdateToOneRemoveNewest(t *testing.T) {
config := func() StreamConfig {
return StreamConfig{Name: "TEST", Subjects: []string{"foo.*"}, MaxMsgsPer: -1}
}
testAllStoreAllPermutations(
t, false, config(),
func(t *testing.T, fs StreamStore) {
// Store the first copy of "foo.0".
_, _, err := fs.StoreMsg("foo.0", nil, nil, 0)
require_NoError(t, err)
// Store filler subjects with enough data that the filestore rolls over into a
// second message block, so both copies of "foo.0" live in different blocks.
msg := make([]byte, 1024*1024)
fillBlock := func(expectedBlocks int) {
for i := 1; i <= 9; i++ {
_, _, err := fs.StoreMsg(fmt.Sprintf("foo.%d", i), nil, msg, 0)
require_NoError(t, err)
}
if f, ok := fs.(*fileStore); ok {
require_True(t, f.numMsgBlocks() >= expectedBlocks)
}
}
fillBlock(2)
// Store a second copy of "foo.0".
nseq, _, err := fs.StoreMsg("foo.0", nil, nil, 0)
require_NoError(t, err)
// And a third, in its own block, which we'll remove as well.
fillBlock(3)
lseq, _, err := fs.StoreMsg("foo.0", nil, nil, 0)
require_NoError(t, err)
removed, err := fs.RemoveMsg(lseq)
require_NoError(t, err)
require_True(t, removed)
// Update max messages per-subject from -1 (unlimited) to 1.
// This transition does not run per-subject limit enforcement, so both copies remain.
cfg := config()
if _, ok := fs.(*fileStore); ok {
cfg.Storage = FileStorage
} else {
cfg.Storage = MemoryStorage
}
cfg.MaxMsgsPer = 1
require_NoError(t, fs.UpdateConfig(&cfg))
ss := fs.SubjectsState("foo.0")["foo.0"]
require_Equal(t, ss.Msgs, 1)
require_Equal(t, ss.First, nseq)
require_Equal(t, ss.Last, nseq)
var smv StoreMsg
sm, err := fs.LoadLastMsg("foo.0", &smv)
require_NoError(t, err)
require_Equal(t, sm.seq, nseq)
sm, _, err = fs.LoadNextMsg("foo.0", false, 0, &smv)
require_NoError(t, err)
require_Equal(t, sm.seq, nseq)
// The older copy must also still be found after a restart.
if f, ok := fs.(*fileStore); ok {
fcfg := f.fcfg
require_NoError(t, f.Stop())
f, err = newFileStore(fcfg, cfg)
require_NoError(t, err)
defer f.Stop()
sm, err = f.LoadLastMsg("foo.0", &smv)
require_NoError(t, err)
require_Equal(t, sm.seq, nseq)
}
},
)
}
func TestStoreCompactCleansUpDmap(t *testing.T) {
config := func() StreamConfig {
return StreamConfig{Name: "TEST", Subjects: []string{"foo"}, MaxMsgsPer: 0}