Test MaxMsgs and MaxMsgsPer in combination (#4253)

This unit test is modelled around issue #4247 and proves that the
`MaxMsgs` and `MaxMsgsPer` limits are correctly enforced together with
`DiscardNew` and `DiscardNewPer`.

Signed-off-by: Neil Twigg <neil@nats.io>

server/filestore_test.go

@@ -4321,6 +4321,55 @@ func TestFileStoreMaxMsgsPerSubject(t *testing.T) {
  })
 }
 
+// Testing the case in https://github.com/nats-io/nats-server/issues/4247
+func TestFileStoreMaxMsgsAndMaxMsgsPerSubject(t *testing.T) {
+ testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
+ fcfg.BlockSize = 128
+ fcfg.CacheExpire = time.Second
+
+ fs, err := newFileStore(
+ fcfg,
+ StreamConfig{
+ Name: "zzz",
+ Subjects: []string{"kv.>"},
+ Storage: FileStorage,
+ Discard: DiscardNew, MaxMsgs: 100, // Total stream policy
+ DiscardNewPer: true, MaxMsgsPer: 1, // Per-subject policy
+ },
+ )
+ require_NoError(t, err)
+ defer fs.Stop()
+
+ for i := 1; i <= 101; i++ {
+ subj := fmt.Sprintf("kv.%d", i)
+ _, _, err := fs.StoreMsg(subj, nil, []byte("value"))
+ if i == 101 {
+ // The 101th iteration should fail because MaxMsgs is set to
+ // 100 and the policy is DiscardNew.
+ require_Error(t, err)
+ } else {
+ require_NoError(t, err)
+ }
+ }
+
+ for i := 1; i <= 100; i++ {
+ subj := fmt.Sprintf("kv.%d", i)
+ _, _, err := fs.StoreMsg(subj, nil, []byte("value"))
+ // All of these iterations should fail because MaxMsgsPer is set
+ // to 1 and DiscardNewPer is set to true, forcing us to reject
+ // cases where there is already a message on this subject.
+ require_Error(t, err)
+ }
+
+ if state := fs.State(); state.Msgs != 100 || state.FirstSeq != 1 || state.LastSeq != 100 || len(state.Deleted) != 0 {
+ // There should be 100 messages exactly, as the 101st subject
+ // should have been rejected in the first loop, and any duplicates
+ // on the other subjects should have been rejected in the second loop.
+ t.Fatalf("Bad state: %+v", state)
+ }
+ })
+}
+
 func TestFileStoreSubjectStateCacheExpiration(t *testing.T) {
  testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
  fcfg.BlockSize = 32