Repository: nats-io/nats-server
Pull request #6470 (2025-02-07 14:35:04)

Purge requests with no subject delete markers
As per [the ADR](https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-43.md#purge-api-call-marker), this adds the `no_markers` flag to the JS API stream purge request, such that a stream can be purged without leaving behind any subject delete markers, if they are enabled.

This helps to avoid the "poison pill" problem if there are subjects for example that are causing problems with a consuming application.

Individual message deletes are less of a concern here because a single message delete will leave behind at most one marker, which can be deleted too (and would not be replaced in that case), but a purge can potentially leave behind lots of markers for lots of subjects, which could be difficult to sort through by hand otherwise.

Signed-off-by: Neil Twigg <neil@nats.io>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 3bb463c4..0cfb8a47 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -540,7 +540,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	// sequence. Need to do this locked as by now the age check timer
 	// has started.
 	if cfg.FirstSeq > 0 && firstSeq <= cfg.FirstSeq {
-		if _, err := fs.purge(cfg.FirstSeq); err != nil {
+		if _, err := fs.purge(cfg.FirstSeq, true); err != nil {
 			return nil, err
 		}
 	}
@@ -7671,13 +7671,13 @@ func compareFn(subject string) func(string, string) bool {
 
 // PurgeEx will remove messages based on subject filters, sequence and number of messages to keep.
 // Will return the number of purged messages.
-func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
+func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64, noMarkers bool) (purged uint64, err error) {
 	if subject == _EMPTY_ || subject == fwcs {
 		if keep == 0 && sequence == 0 {
-			return fs.Purge()
+			return fs.purge(0, noMarkers)
 		}
 		if sequence > 1 {
-			return fs.Compact(sequence)
+			return fs.compact(sequence, noMarkers)
 		}
 	}
 
@@ -7752,7 +7752,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 				}
 				// PSIM and FSS updates.
 				mb.removeSeqPerSubject(sm.subj, seq)
-				fs.removePerSubject(sm.subj, fs.cfg.SubjectDeleteMarkerTTL > 0)
+				fs.removePerSubject(sm.subj, !noMarkers && fs.cfg.SubjectDeleteMarkerTTL > 0)
 				// Track tombstones we need to write.
 				tombs = append(tombs, msgId{sm.seq, sm.ts})
 
@@ -7821,10 +7821,10 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 // Purge will remove all messages from this store.
 // Will return the number of purged messages.
 func (fs *fileStore) Purge() (uint64, error) {
-	return fs.purge(0)
+	return fs.purge(0, false)
 }
 
-func (fs *fileStore) purge(fseq uint64) (uint64, error) {
+func (fs *fileStore) purge(fseq uint64, noMarkers bool) (uint64, error) {
 	fs.mu.Lock()
 	if fs.closed {
 		fs.mu.Unlock()
@@ -7848,7 +7848,7 @@ func (fs *fileStore) purge(fseq uint64) (uint64, error) {
 	fs.lmb = nil
 	fs.bim = make(map[uint32]*msgBlock)
 	// Subject delete markers if needed.
-	if fs.cfg.SubjectDeleteMarkerTTL > 0 {
+	if !noMarkers && fs.cfg.SubjectDeleteMarkerTTL > 0 {
 		fs.psim.IterOrdered(func(subject []byte, _ *psi) bool {
 			fs.markers = append(fs.markers, string(subject))
 			return true
@@ -7930,10 +7930,16 @@ func (fs *fileStore) purge(fseq uint64) (uint64, error) {
 
 // Compact will remove all messages from this store up to
 // but not including the seq parameter.
+// No subject delete markers will be left if they are enabled. If they are disabled,
+// then this is functionally equivalent to a normal Compact() call.
 // Will return the number of purged messages.
 func (fs *fileStore) Compact(seq uint64) (uint64, error) {
+	return fs.compact(seq, false)
+}
+
+func (fs *fileStore) compact(seq uint64, noMarkers bool) (uint64, error) {
 	if seq == 0 {
-		return fs.purge(seq)
+		return fs.purge(seq, noMarkers)
 	}
 
 	var purged, bytes uint64
@@ -7942,7 +7948,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 	// Same as purge all.
 	if lseq := fs.state.LastSeq; seq > lseq {
 		fs.mu.Unlock()
-		return fs.purge(seq)
+		return fs.purge(seq, noMarkers)
 	}
 	// We have to delete interior messages.
 	smb := fs.selectMsgBlock(seq)
@@ -7965,7 +7971,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		mb.fss.IterOrdered(func(bsubj []byte, ss *SimpleState) bool {
 			subj := bytesToString(bsubj)
 			for i := uint64(0); i < ss.Msgs; i++ {
-				fs.removePerSubject(subj, fs.cfg.SubjectDeleteMarkerTTL > 0)
+				fs.removePerSubject(subj, !noMarkers && fs.cfg.SubjectDeleteMarkerTTL > 0)
 			}
 			return true
 		})
@@ -8011,7 +8017,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 			}
 			// Update fss
 			smb.removeSeqPerSubject(sm.subj, mseq)
-			fs.removePerSubject(sm.subj, fs.cfg.SubjectDeleteMarkerTTL > 0)
+			fs.removePerSubject(sm.subj, !noMarkers && fs.cfg.SubjectDeleteMarkerTTL > 0)
 		}
 	}
 
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index c05f085f..5a79759f 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -503,6 +503,9 @@ type JSApiStreamPurgeRequest struct {
 	Subject string `json:"filter,omitempty"`
 	// Number of messages to keep.
 	Keep uint64 `json:"keep,omitempty"`
+	// NoMarkers ensures that subject delete markers will not be left. If subject delete markers
+	// are not enabled on the stream, then this flag is no-op.
+	NoMarkers bool `json:"no_markers,omitempty"`
 }
 
 type JSApiStreamPurgeResponse struct {
diff --git a/server/memstore.go b/server/memstore.go
index 2a3e8fb3..9e10ee49 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -64,7 +64,7 @@ func newMemStore(cfg *StreamConfig) (*memStore, error) {
 		ms.ttls = thw.NewHashWheel()
 	}
 	if cfg.FirstSeq > 0 {
-		if _, err := ms.purge(cfg.FirstSeq); err != nil {
+		if _, err := ms.purge(cfg.FirstSeq, true); err != nil {
 			return nil, err
 		}
 	}
@@ -1088,13 +1088,13 @@ func (ms *memStore) expireMsgs() {
 
 // PurgeEx will remove messages based on subject filters, sequence and number of messages to keep.
 // Will return the number of purged messages.
-func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
+func (ms *memStore) PurgeEx(subject string, sequence, keep uint64, noMarkers bool) (purged uint64, err error) {
 	if subject == _EMPTY_ || subject == fwcs {
 		if keep == 0 && sequence == 0 {
-			return ms.Purge()
+			return ms.purge(0, noMarkers)
 		}
 		if sequence > 1 {
-			return ms.Compact(sequence)
+			return ms.compact(sequence, noMarkers)
 		} else if keep > 0 {
 			ms.mu.RLock()
 			msgs, lseq := ms.state.Msgs, ms.state.LastSeq
@@ -1102,7 +1102,7 @@ func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint6
 			if keep >= msgs {
 				return 0, nil
 			}
-			return ms.Compact(lseq - keep + 1)
+			return ms.compact(lseq-keep+1, noMarkers)
 		}
 		return 0, nil
 
@@ -1120,9 +1120,13 @@ func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint6
 			last = sequence - 1
 		}
 		ms.mu.Lock()
+		var removeReason string
+		if !noMarkers {
+			removeReason = JSMarkerReasonPurge
+		}
 		for seq := ss.First; seq <= last; seq++ {
 			if sm, ok := ms.msgs[seq]; ok && eq(sm.subj, subject) {
-				if ok := ms.removeMsg(sm.seq, false, JSMarkerReasonPurge); ok {
+				if ok := ms.removeMsg(sm.seq, false, removeReason); ok {
 					purged++
 					if purged >= ss.Msgs {
 						break
@@ -1141,10 +1145,10 @@ func (ms *memStore) Purge() (uint64, error) {
 	ms.mu.RLock()
 	first := ms.state.LastSeq + 1
 	ms.mu.RUnlock()
-	return ms.purge(first)
+	return ms.purge(first, false)
 }
 
-func (ms *memStore) purge(fseq uint64) (uint64, error) {
+func (ms *memStore) purge(fseq uint64, noMarkers bool) (uint64, error) {
 	ms.mu.Lock()
 	purged := uint64(len(ms.msgs))
 	cb := ms.scb
@@ -1160,7 +1164,7 @@ func (ms *memStore) purge(fseq uint64) (uint64, error) {
 	ms.state.Msgs = 0
 	ms.msgs = make(map[uint64]*StoreMsg)
 	// Subject delete markers if needed.
-	if ms.cfg.SubjectDeleteMarkerTTL > 0 {
+	if !noMarkers && ms.cfg.SubjectDeleteMarkerTTL > 0 {
 		ms.fss.IterOrdered(func(bsubj []byte, ss *SimpleState) bool {
 			ms.markers = append(ms.markers, string(bsubj))
 			return true
@@ -1184,6 +1188,10 @@ func (ms *memStore) purge(fseq uint64) (uint64, error) {
 // but not including the seq parameter.
 // Will return the number of purged messages.
 func (ms *memStore) Compact(seq uint64) (uint64, error) {
+	return ms.compact(seq, false)
+}
+
+func (ms *memStore) compact(seq uint64, noMarkers bool) (uint64, error) {
 	if seq == 0 {
 		return ms.Purge()
 	}
@@ -1206,7 +1214,7 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
 			if sm := ms.msgs[seq]; sm != nil {
 				bytes += memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 				purged++
-				ms.removeSeqPerSubject(sm.subj, seq, ms.cfg.SubjectDeleteMarkerTTL > 0)
+				ms.removeSeqPerSubject(sm.subj, seq, !noMarkers && ms.cfg.SubjectDeleteMarkerTTL > 0)
 				// Must delete message after updating per-subject info, to be consistent with file store.
 				delete(ms.msgs, seq)
 			}
@@ -1230,7 +1238,7 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
 		ms.state.FirstTime = time.Time{}
 		ms.state.LastSeq = seq - 1
 		// Subject delete markers if needed.
-		if ms.cfg.SubjectDeleteMarkerTTL > 0 {
+		if !noMarkers && ms.cfg.SubjectDeleteMarkerTTL > 0 {
 			ms.fss.IterOrdered(func(bsubj []byte, ss *SimpleState) bool {
 				ms.markers = append(ms.markers, string(bsubj))
 				return true
diff --git a/server/raft.go b/server/raft.go
index 2d4354c0..9df22616 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -89,7 +89,7 @@ type WAL interface {
 	RemoveMsg(index uint64) (bool, error)
 	Compact(index uint64) (uint64, error)
 	Purge() (uint64, error)
-	PurgeEx(subject string, seq, keep uint64) (uint64, error)
+	PurgeEx(subject string, seq, keep uint64, noMarkers bool) (uint64, error)
 	Truncate(seq uint64) error
 	State() StreamState
 	FastState(*StreamState)
@@ -3253,7 +3253,7 @@ func (n *raft) truncateWAL(term, index uint64) {
 			n.wal.Truncate(0)
 			// If our index is non-zero use PurgeEx to set us to the correct next index.
 			if index > 0 {
-				n.wal.PurgeEx(fwcs, index+1, 0)
+				n.wal.PurgeEx(fwcs, index+1, 0, true)
 			}
 		} else {
 			n.warn("Error truncating WAL: %v", err)
diff --git a/server/store.go b/server/store.go
index 2e1585e1..53dab938 100644
--- a/server/store.go
+++ b/server/store.go
@@ -100,7 +100,7 @@ type StreamStore interface {
 	RemoveMsg(seq uint64) (bool, error)
 	EraseMsg(seq uint64) (bool, error)
 	Purge() (uint64, error)
-	PurgeEx(subject string, seq, keep uint64) (uint64, error)
+	PurgeEx(subject string, seq, keep uint64, noMarkers bool) (uint64, error)
 	Compact(seq uint64) (uint64, error)
 	Truncate(seq uint64) error
 	GetSeqFromTime(t time.Time) uint64
diff --git a/server/stream.go b/server/stream.go
index 5b9d5c64..7a848b1f 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -2212,7 +2212,7 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
 	mset.mu.RUnlock()
 
 	if preq != nil {
-		purged, err = mset.store.PurgeEx(preq.Subject, preq.Sequence, preq.Keep)
+		purged, err = mset.store.PurgeEx(preq.Subject, preq.Sequence, preq.Keep, preq.NoMarkers)
 	} else {
 		purged, err = mset.store.Purge()
 	}
@@ -2944,7 +2944,7 @@ func (mset *stream) setupMirrorConsumer() error {
 					// Check to see if delivered is past our last and we have no msgs. This will help the
 					// case when mirroring a stream that has a very high starting sequence number.
 					if state.Msgs == 0 && ccr.ConsumerInfo.Delivered.Stream > state.LastSeq {
-						mset.store.PurgeEx(_EMPTY_, ccr.ConsumerInfo.Delivered.Stream+1, 0)
+						mset.store.PurgeEx(_EMPTY_, ccr.ConsumerInfo.Delivered.Stream+1, 0, true)
 						mset.lseq = ccr.ConsumerInfo.Delivered.Stream
 					} else {
 						mset.skipMsgs(state.LastSeq+1, ccr.ConsumerInfo.Delivered.Stream)


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 83c4a405..cc17ddff 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3302,7 +3302,7 @@ func TestFileStorePurgeExKeepOneBug(t *testing.T) {
 			t.Fatalf("Expected to find 2 `A` msgs, got %d", fss.Msgs)
 		}
 
-		n, err := fs.PurgeEx("A", 0, 1)
+		n, err := fs.PurgeEx("A", 0, 1, false)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -3671,7 +3671,7 @@ func TestFileStorePurgeExWithSubject(t *testing.T) {
 		require_NoError(t, err)
 
 		// This should purge all "foo.1"
-		p, err := fs.PurgeEx("foo.1", 1, 0)
+		p, err := fs.PurgeEx("foo.1", 1, 0, false)
 		require_NoError(t, err)
 		require_Equal(t, p, uint64(total))
 
@@ -5048,12 +5048,12 @@ func TestFileStoreKeepWithDeletedMsgsBug(t *testing.T) {
 		fs.StoreMsg("B", nil, msg, 0)
 	}
 
-	n, err := fs.PurgeEx("A", 0, 0)
+	n, err := fs.PurgeEx("A", 0, 0, false)
 	require_NoError(t, err)
 	require_True(t, n == 5)
 
 	// Purge with keep.
-	n, err = fs.PurgeEx(_EMPTY_, 0, 2)
+	n, err = fs.PurgeEx(_EMPTY_, 0, 2, false)
 	require_NoError(t, err)
 	require_True(t, n == 3)
 }
@@ -5445,7 +5445,7 @@ func TestFileStoreFullStatePurge(t *testing.T) {
 			fs.StoreMsg("C", nil, msg, 0)
 		}
 
-		n, err = fs.PurgeEx("B", 0, 0)
+		n, err = fs.PurgeEx("B", 0, 0, false)
 		require_NoError(t, err)
 		require_Equal(t, n, 5)
 
@@ -5462,7 +5462,7 @@ func TestFileStoreFullStatePurge(t *testing.T) {
 		}
 
 		// Purge with keep.
-		n, err = fs.PurgeEx(_EMPTY_, 0, 2)
+		n, err = fs.PurgeEx(_EMPTY_, 0, 2, false)
 		require_NoError(t, err)
 		require_Equal(t, n, 3)
 
@@ -6370,7 +6370,7 @@ func TestFileStorePurgeExBufPool(t *testing.T) {
 		fs.StoreMsg("foo.bar", nil, msg, 0)
 	}
 
-	p, err := fs.PurgeEx("foo.bar", 1, 0)
+	p, err := fs.PurgeEx("foo.bar", 1, 0, false)
 	require_NoError(t, err)
 	require_Equal(t, p, 1000)
 
@@ -6409,7 +6409,7 @@ func TestFileStoreFSSMeta(t *testing.T) {
 	// Let cache's expire before PurgeEx which will load them back in.
 	time.Sleep(250 * time.Millisecond)
 
-	p, err := fs.PurgeEx("A", 1, 0)
+	p, err := fs.PurgeEx("A", 1, 0, false)
 	require_NoError(t, err)
 	require_Equal(t, p, 2)
 
@@ -6827,7 +6827,7 @@ func TestFileStoreWriteFullStateAfterPurgeEx(t *testing.T) {
 	fs.RemoveMsg(9)
 	fs.RemoveMsg(10)
 
-	n, err := fs.PurgeEx(">", 8, 0)
+	n, err := fs.PurgeEx(">", 8, 0, false)
 	require_NoError(t, err)
 	require_Equal(t, n, 7)
 
@@ -7767,7 +7767,7 @@ func TestFileStoreRestoreDeleteTombstonesExceedingMaxBlkSize(t *testing.T) {
 		require_NoError(t, err)
 		defer fs.Stop()
 
-		n, err := fs.PurgeEx(_EMPTY_, 1_000_000_000, 0)
+		n, err := fs.PurgeEx(_EMPTY_, 1_000_000_000, 0, false)
 		require_NoError(t, err)
 		require_Equal(t, n, 0)
 
@@ -8703,7 +8703,7 @@ func TestFileStoreSubjectDeleteMarkersOnPurgeEx(t *testing.T) {
 		require_NoError(t, err)
 	}
 
-	_, err = fs.PurgeEx("test.*", 1, 0)
+	_, err = fs.PurgeEx("test.*", 1, 0, false)
 	require_NoError(t, err)
 
 	for i := uint64(0); i < 10; i++ {
@@ -8715,6 +8715,33 @@ func TestFileStoreSubjectDeleteMarkersOnPurgeEx(t *testing.T) {
 	}
 }
 
+func TestFileStoreSubjectDeleteMarkersOnPurgeExNoMarkers(t *testing.T) {
+	storeDir := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{
+			Name: "zzz", Subjects: []string{"test.*"}, Storage: FileStorage,
+			MaxAge: time.Second, AllowMsgTTL: true,
+			SubjectDeleteMarkerTTL: time.Second,
+		},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	for i := 0; i < 10; i++ {
+		_, _, err := fs.StoreMsg(fmt.Sprintf("test.%d", i), nil, nil, 0)
+		require_NoError(t, err)
+	}
+
+	_, err = fs.PurgeEx("test.*", 1, 0, true)
+	require_NoError(t, err)
+
+	for i := uint64(0); i < 10; i++ {
+		_, err := fs.LoadMsg(11+i, nil)
+		require_Error(t, err)
+	}
+}
+
 func TestFileStoreSubjectDeleteMarkersOnCompact(t *testing.T) {
 	storeDir := t.TempDir()
 	fs, err := newFileStore(
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 307177a8..d5c144bd 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -25396,3 +25396,113 @@ func TestJetStreamSubjectDeleteMarkersWithMirror(t *testing.T) {
 	})
 	require_Error(t, err)
 }
+
+func TestJetStreamSubjectDeleteMarkersAfterPurge(t *testing.T) {
+	for _, storage := range []StorageType{FileStorage, MemoryStorage} {
+		t.Run(storage.String(), func(t *testing.T) {
+			s := RunBasicJetStreamServer(t)
+			defer s.Shutdown()
+
+			nc, js := jsClientConnect(t, s)
+			defer nc.Close()
+
+			jsStreamCreate(t, nc, &StreamConfig{
+				Name:                   "TEST",
+				Storage:                storage,
+				Subjects:               []string{"test.>"},
+				MaxAge:                 time.Second,
+				AllowMsgTTL:            true,
+				SubjectDeleteMarkerTTL: time.Second,
+			})
+
+			sub, err := js.SubscribeSync("test.>")
+			require_NoError(t, err)
+
+			for i := 0; i < 5; i++ {
+				_, err = js.Publish(fmt.Sprintf("test.%d", i), nil)
+				require_NoError(t, err)
+			}
+
+			for i := 0; i < 5; i++ {
+				_, err := sub.NextMsg(time.Second)
+				require_NoError(t, err)
+			}
+
+			body, err := json.Marshal(JSApiStreamPurgeRequest{
+				Sequence:  6,
+				NoMarkers: false,
+			})
+			require_NoError(t, err)
+
+			resp, err := nc.Request(fmt.Sprintf(JSApiStreamPurgeT, "TEST"), body, time.Second)
+			require_NoError(t, err)
+
+			var apiresp JSApiStreamPurgeResponse
+			require_NoError(t, json.Unmarshal(resp.Data, &apiresp))
+			require_True(t, apiresp.Success)
+
+			si, err := js.StreamInfo("TEST")
+			require_NoError(t, err)
+			require_Equal(t, si.State.FirstSeq, 6)
+			require_Equal(t, si.State.Msgs, 5)
+
+			for i := 0; i < 5; i++ {
+				msg, err := sub.NextMsg(time.Second)
+				require_NoError(t, err)
+				require_Equal(t, msg.Header.Get(JSMarkerReason), JSMarkerReasonPurge)
+			}
+		})
+	}
+}
+
+func TestJetStreamSubjectDeleteMarkersAfterPurgeNoMarkers(t *testing.T) {
+	for _, storage := range []StorageType{FileStorage, MemoryStorage} {
+		t.Run(storage.String(), func(t *testing.T) {
+			s := RunBasicJetStreamServer(t)
+			defer s.Shutdown()
+
+			nc, js := jsClientConnect(t, s)
+			defer nc.Close()
+
+			jsStreamCreate(t, nc, &StreamConfig{
+				Name:                   "TEST",
+				Storage:                storage,
+				Subjects:               []string{"test.>"},
+				MaxAge:                 time.Second,
+				AllowMsgTTL:            true,
+				SubjectDeleteMarkerTTL: time.Second,
+			})
+
+			sub, err := js.SubscribeSync("test.>")
+			require_NoError(t, err)
+
+			for i := 0; i < 5; i++ {
+				_, err = js.Publish(fmt.Sprintf("test.%d", i), nil)
+				require_NoError(t, err)
+			}
+
+			for i := 0; i < 5; i++ {
+				_, err := sub.NextMsg(time.Second)
+				require_NoError(t, err)
+			}
+
+			body, err := json.Marshal(JSApiStreamPurgeRequest{
+				Sequence:  6,
+				NoMarkers: true,
+			})
+			require_NoError(t, err)
+
+			resp, err := nc.Request(fmt.Sprintf(JSApiStreamPurgeT, "TEST"), body, time.Second)
+			require_NoError(t, err)
+
+			var apiresp JSApiStreamPurgeResponse
+			require_NoError(t, json.Unmarshal(resp.Data, &apiresp))
+			require_True(t, apiresp.Success)
+
+			si, err := js.StreamInfo("TEST")
+			require_NoError(t, err)
+			require_Equal(t, si.State.FirstSeq, 6)
+			require_Equal(t, si.State.Msgs, 0)
+		})
+	}
+}
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 8d3fdae2..419b59df 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -449,7 +449,7 @@ func TestMemStorePurgeExWithSubject(t *testing.T) {
 	}
 
 	// This should purge all.
-	ms.PurgeEx("foo", 1, 0)
+	ms.PurgeEx("foo", 1, 0, false)
 	require_True(t, ms.State().Msgs == 0)
 }
 
@@ -1043,7 +1043,7 @@ func TestMemStorePurgeExWithDeletedMsgs(t *testing.T) {
 	ms.RemoveMsg(2)
 	ms.RemoveMsg(9) // This was the bug
 
-	n, err := ms.PurgeEx(_EMPTY_, 9, 0)
+	n, err := ms.PurgeEx(_EMPTY_, 9, 0, true)
 	require_NoError(t, err)
 	require_Equal(t, n, 7)
 
@@ -1275,7 +1275,7 @@ func TestMemStoreSubjectDeleteMarkersOnPurgeEx(t *testing.T) {
 		require_NoError(t, err)
 	}
 
-	_, err = ms.PurgeEx("test.*", 1, 0)
+	_, err = ms.PurgeEx("test.*", 1, 0, false)
 	require_NoError(t, err)
 
 	for i := uint64(0); i < 10; i++ {
@@ -1287,6 +1287,31 @@ func TestMemStoreSubjectDeleteMarkersOnPurgeEx(t *testing.T) {
 	}
 }
 
+func TestMemStoreSubjectDeleteMarkersOnPurgeExNoMarkers(t *testing.T) {
+	ms, err := newMemStore(
+		&StreamConfig{
+			Name: "zzz", Subjects: []string{"test.*"}, Storage: MemoryStorage,
+			MaxAge: time.Second, AllowMsgTTL: true,
+			SubjectDeleteMarkerTTL: time.Second,
+		},
+	)
+	require_NoError(t, err)
+	defer ms.Stop()
+
+	for i := 0; i < 10; i++ {
+		_, _, err := ms.StoreMsg(fmt.Sprintf("test.%d", i), nil, nil, 0)
+		require_NoError(t, err)
+	}
+
+	_, err = ms.PurgeEx("test.*", 1, 0, true)
+	require_NoError(t, err)
+
+	for i := uint64(0); i < 10; i++ {
+		_, err := ms.LoadMsg(11+i, nil)
+		require_Error(t, err)
+	}
+}
+
 func TestMemStoreSubjectDeleteMarkersOnCompact(t *testing.T) {
 	ms, err := newMemStore(
 		&StreamConfig{
diff --git a/server/norace_test.go b/server/norace_test.go
index f0fa8375..dcbe0cb8 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -11326,7 +11326,7 @@ func TestNoRaceStoreReverseWalkWithDeletesPerf(t *testing.T) {
 		require_Equal(t, ss.Msgs, 1_000_002)
 
 		// Create a bunch of interior deletes.
-		p, err := store.PurgeEx("foo.B", 1, 0)
+		p, err := store.PurgeEx("foo.B", 1, 0, true)
 		require_NoError(t, err)
 		require_Equal(t, p, 1_000_000)