Skip to content

Commit

Permalink
[FIXED] Expire on recover could not update global per subject map (#4439
Browse files Browse the repository at this point in the history
)

When expiring complete blocks on recover make sure to update global
subject index psim.

Signed-off-by: Derek Collison <derek@nats.io>
  • Loading branch information
derekcollison committed Aug 27, 2023
2 parents 5b18e80 + b66a7f6 commit f9a2efd
Show file tree
Hide file tree
Showing 2 changed files with 39 additions and 0 deletions.
5 changes: 5 additions & 0 deletions server/filestore.go
Expand Up @@ -1315,6 +1315,11 @@ func (fs *fileStore) expireMsgsOnRecover() {
fs.psim = make(map[string]*psi)
return false
}
// Make sure we do subject cleanup as well.
mb.ensurePerSubjectInfoLoaded()
for subj := range mb.fss {
fs.removePerSubject(subj)
}
mb.dirtyCloseWithRemove(true)
deleted++
return true
Expand Down
34 changes: 34 additions & 0 deletions server/filestore_test.go
Expand Up @@ -4662,6 +4662,40 @@ func TestFileStoreFSSCloseAndKeepOnExpireOnRecoverBug(t *testing.T) {
})
}

func TestFileStoreExpireOnRecoverSubjectAccounting(t *testing.T) {
const msgLen = 19
msg := bytes.Repeat([]byte("A"), msgLen)

testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
fcfg.BlockSize = 100
ttl := 200 * time.Millisecond
scfg := StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage, MaxAge: ttl}

fs, err := newFileStore(fcfg, scfg)
require_NoError(t, err)
defer fs.Stop()

// These are in first block.
fs.StoreMsg("A", nil, msg)
fs.StoreMsg("B", nil, msg)
time.Sleep(ttl / 2)
// This one in 2nd block.
fs.StoreMsg("C", nil, msg)
fs.Stop()

time.Sleep(ttl/2 + 10*time.Millisecond)

fs, err = newFileStore(fcfg, scfg)
require_NoError(t, err)
defer fs.Stop()

// Make sure we take into account PSIM when throwing a whole block away.
if state := fs.State(); state.NumSubjects != 1 {
t.Fatalf("Expected 1 subject, got %d", state.NumSubjects)
}
})
}

func TestFileStoreFSSBadStateBug(t *testing.T) {
testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
fs, err := newFileStore(
Expand Down

0 comments on commit f9a2efd

Please sign in to comment.