Repository: nats-io/nats-server

Issue:
Sending an InProgress acknowledgement does not count towards consumer activity
### Observed behavior

I have a work queue scenario where a worker is fetching messages one by one to process them.

The time it takes to process a message is variable and some messages can take a long time to process.
An inactive threshold is set on the consumer to have the NATS server automatically clean up the consumer.

I originally opened an issue for the consumer getting removed while actively processing a message:  https://github.com/nats-io/nats.go/issues/1900. This was fixed by https://github.com/nats-io/nats-server/pull/7052

To ensure messages are redelivered quickly when processing fails, the `AckWait` is deliberately configured lower than the max processing time. `InProgress` is called to extend the processing window while a message is getting worked on.

The consumer is getting detected as inactive and gets removed while working on a message.

### Expected behavior

Pending acks are now counted toward consumer activity, I'd expect sending an `InProgress` acknowledgment to similarly count and prevent the consumer's removal.

### Server and client version

Server: Build of `26797a84c2c691176511ea3b3b0e9e841c64fed8`
Client: github.com/nats-io/nats.go v1.43.0

### Host environment

_No response_

### Steps to reproduce

Go script to reproduce:

```go
package main

import (
	"context"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	url := os.Getenv("NATS_URL")
	if url == "" {
		url = nats.DefaultURL
	}

	nc, err := nats.Connect(url)
	if err != nil {
		log.Fatal(err)
	}
	defer nc.Drain()

	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatal(err)
	}

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer cancel()

	stream, err := js.CreateStream(ctx, jetstream.StreamConfig{
		Name:      "queue1",
		Subjects:  []string{"queue1.>"},
		Retention: jetstream.WorkQueuePolicy,
	})
	if err != nil {
		log.Fatal(err)
	}

	consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Name:              "queue1-consumer",
		FilterSubject:     "queue1.>",
		InactiveThreshold: time.Second * 5,
		AckWait:           time.Second * 10,
	})
	if err != nil {
		log.Fatal(err)
	}

	sub, err := consumer.Consume(func(msg jetstream.Msg) {
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()

		meta, err := msg.Metadata()
		if err != nil {
			log.Printf("Error getting metadata: %v", err)
			return
		}
		log.Printf("Received message %d, delivered %d\n", meta.Sequence.Stream, meta.NumDelivered)

		go func() {
			interval := consumer.CachedInfo().Config.AckWait * 70 / 100
			ticker := time.NewTicker(interval)
			defer ticker.Stop()

			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					log.Printf("Extending ack wait for message %d\n", meta.Sequence.Stream)
					if err := msg.InProgress(); err != nil {
						log.Printf("Error in calling in progress: %v", err)
					}
				}
			}
		}()

		time.Sleep(time.Second * 20)

		msg.Ack()
		log.Printf("Acked message %d\n", meta.Sequence.Stream)
	}, jetstream.ConsumeErrHandler(func(consumeCtx jetstream.ConsumeContext, err error) {
		log.Printf("Error consuming message: %v", err)
	}), jetstream.PullMaxMessages(1))
	if err != nil {
		log.Fatal(err)
	}

	<-ctx.Done()

	sub.Stop()
	<-sub.Closed()
}
```

Publish a couple of message using the nats CLI:

```sh
for i in {1..5}; do
  nats publish queue1.msg "msg"
done
```

Hints from issue comments:
@MauriceVanVeen: > Pending acks are now counted toward consumer activity, I'd expect sending an InProgress acknowledgment to similarly count and prevent the consumer's removal.

This is already the case. The repro does not fail when I'm running it.

However, for these specific `InactiveThreshold: 5s, AckWait: 10s` and the `msg.InProgress()` being sent every 7 seconds (due to `10s*7/10`). You could get redeliveries or be marked inactive if any of those `InProgress` messages are lost.

`msg.InProgress()` is a "fire-and-forget" message, meaning it could be lost. I'd recommend to send `InProgress` messages more often. Sending every 3 seconds for example would allow some messages to be lost, while still being marked active if at least one goes through before you pull new messages.

There was a small race condition though, that will be fixed as part of https://github.com/nats-io/nats-server/pull/7107. I doubt you're hitting this based on the shared repro though.@welteki: Thanks for looking into this.

I ran the repro again and did not encounter the issue this time. It appears there may have been a misconfiguration on my end during the initial testing. My apologies for that.

> msg.InProgress() is a "fire-and-forget" message, meaning it could be lost. I'd recommend to send InProgress messages more often. Sending every 3 seconds for example would allow some messages to be lost, while still being marked active if at least one goes through before you pull new messages.

Thanks for the recommendation. I will adjust the message frequency.

Solution patch:
diff --git a/server/consumer.go b/server/consumer.go
index 1e1fd353..d15b6c46 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -4465,16 +4465,19 @@ func (o *consumer) processInboundAcks(qch chan struct{}) {
 	for {
 		select {
 		case <-o.ackMsgs.ch:
+			// If we have an inactiveThreshold set, mark our activity.
+			// Do this before processing acks, otherwise we might race if there are no pending messages
+			// anymore and the inactivity threshold kicks in before we're able to mark activity.
+			if hasInactiveThresh {
+				o.suppressDeletion()
+			}
+
 			acks := o.ackMsgs.pop()
 			for _, ack := range acks {
 				o.processAck(ack.subject, ack.reply, ack.hdr, ack.msg)
 				ack.returnToPool()
 			}
 			o.ackMsgs.recycle(&acks)
-			// If we have an inactiveThreshold set, mark our activity.
-			if hasInactiveThresh {
-				o.suppressDeletion()
-			}
 		case <-ticker.C:
 			o.checkAckFloor()
 		case <-qch:
diff --git a/server/filestore.go b/server/filestore.go
index 9936949a..9a11c80b 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -306,8 +306,8 @@ const (
 	consumerDir = "obs"
 	// Index file for a consumer.
 	consumerState = "o.dat"
-	// The suffix that will be given to a new temporary block during compression.
-	compressTmpSuffix = ".tmp"
+	// The suffix that will be given to a new temporary block for compression or when rewriting the full file.
+	blkTmpSuffix = ".tmp"
 	// This is where we keep state on templates.
 	tmplsDir = "templates"
 	// Maximum size of a write buffer we may consider for re-use.
@@ -1223,7 +1223,9 @@ func (fs *fileStore) rebuildStateLocked(ld *LostStreamData) {
 				fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
 			}
 		}
-		if lseq := atomic.LoadUint64(&mb.last.seq); lseq > fs.state.LastSeq {
+		// Preserve last time, could have erased the last message in one block, and then
+		// have a tombstone with the proper timestamp afterward in another block
+		if lseq := atomic.LoadUint64(&mb.last.seq); lseq >= fs.state.LastSeq {
 			fs.state.LastSeq = lseq
 			if mb.last.ts == 0 {
 				fs.state.LastTime = time.Time{}
@@ -1331,6 +1333,10 @@ func (mb *msgBlock) convertToEncrypted() error {
 	if err != nil {
 		return err
 	}
+	// Check for compression.
+	if buf, err = mb.decompressIfNeeded(buf); err != nil {
+		return err
+	}
 	if err := mb.indexCacheBuf(buf); err != nil {
 		// This likely indicates this was already encrypted or corrupt.
 		mb.cache = nil
@@ -1401,15 +1407,9 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, []uint64, error) {
 	firstNeedsSet := true
 
 	// Check if we need to decrypt.
-	if mb.bek != nil && len(buf) > 0 {
-		// Recreate to reset counter.
-		mb.bek, err = genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
-		if err != nil {
-			return nil, nil, err
-		}
-		mb.bek.XORKeyStream(buf, buf)
+	if err = mb.encryptOrDecryptIfNeeded(buf); err != nil {
+		return nil, nil, err
 	}
-
 	// Check for compression.
 	if buf, err = mb.decompressIfNeeded(buf); err != nil {
 		return nil, nil, err
@@ -2011,12 +2011,9 @@ func (mb *msgBlock) lastChecksum() []byte {
 	}
 	if mb.bek != nil {
 		if buf, _ := mb.loadBlock(nil); len(buf) >= checksumSize {
-			bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
-			if err != nil {
+			if err = mb.encryptOrDecryptIfNeeded(buf); err != nil {
 				return nil
 			}
-			mb.bek = bek
-			mb.bek.XORKeyStream(buf, buf)
 			copy(lchk[0:], buf[len(buf)-checksumSize:])
 		}
 	} else {
@@ -2110,7 +2107,9 @@ func (fs *fileStore) recoverMsgs() error {
 					fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
 				}
 			}
-			if lseq := atomic.LoadUint64(&mb.last.seq); lseq > fs.state.LastSeq {
+			// Preserve last time, could have erased the last message in one block, and then
+			// have a tombstone with the proper timestamp afterward in another block
+			if lseq := atomic.LoadUint64(&mb.last.seq); lseq >= fs.state.LastSeq {
 				fs.state.LastSeq = lseq
 				if mb.last.ts == 0 {
 					fs.state.LastTime = time.Time{}
@@ -3974,7 +3973,7 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 		// Flush any pending messages.
 		lmb.flushPendingMsgsLocked()
 		// Determine if we can reclaim any resources here.
-		lmb.closeFDsLocked()
+		lmb.closeFDsLockedNoCheck()
 		if lmb.cache != nil {
 			// Reset write timestamp and see if we can expire this cache.
 			rbuf = lmb.tryExpireWriteCache()
@@ -3984,7 +3983,11 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 		if fs.fcfg.Compression != NoCompression {
 			// We've now reached the end of this message block, if we want
 			// to compress blocks then now's the time to do it.
-			go lmb.recompressOnDiskIfNeeded()
+			go func() {
+				lmb.mu.Lock()
+				defer lmb.mu.Unlock()
+				lmb.recompressOnDiskIfNeeded()
+			}()
 		}
 	}
 
@@ -4248,20 +4251,19 @@ func (fs *fileStore) StoreMsg(subj string, hdr, msg []byte, ttl int64) (uint64,
 // we will place an empty record marking the sequence as used. The
 // sequence will be marked erased.
 // fs lock should be held.
-func (mb *msgBlock) skipMsg(seq uint64, now time.Time) {
+func (mb *msgBlock) skipMsg(seq uint64, now int64) {
 	if mb == nil {
 		return
 	}
 	var needsRecord bool
-	nowts := now.UnixNano()
 
 	mb.mu.Lock()
 	// If we are empty can just do meta.
 	if mb.msgs == 0 {
 		atomic.StoreUint64(&mb.last.seq, seq)
-		mb.last.ts = nowts
+		mb.last.ts = now
 		atomic.StoreUint64(&mb.first.seq, seq+1)
-		mb.first.ts = nowts
+		mb.first.ts = 0
 		needsRecord = mb == mb.fs.lmb
 		if needsRecord && mb.rbytes > 0 {
 			// We want to make sure since we have no messages
@@ -4280,7 +4282,7 @@ func (mb *msgBlock) skipMsg(seq uint64, now time.Time) {
 	mb.mu.Unlock()
 
 	if needsRecord {
-		mb.writeMsgRecord(emptyRecordLen, seq|ebit, _EMPTY_, nil, nil, nowts, true)
+		mb.writeMsgRecord(emptyRecordLen, seq|ebit, _EMPTY_, nil, nil, now, true)
 	} else {
 		mb.kickFlusher()
 	}
@@ -4298,18 +4300,18 @@ func (fs *fileStore) SkipMsg() uint64 {
 	}
 
 	// Grab time and last seq.
-	now, seq := time.Now(), fs.state.LastSeq+1
+	now, seq := ats.AccessTime(), fs.state.LastSeq+1
 
 	// Write skip msg.
 	mb.skipMsg(seq, now)
 
 	// Update fs state.
-	fs.state.LastSeq, fs.state.LastTime = seq, now
+	fs.state.LastSeq, fs.state.LastTime = seq, time.Unix(0, now).UTC()
 	if fs.state.Msgs == 0 {
-		fs.state.FirstSeq, fs.state.FirstTime = seq, now
+		fs.state.FirstSeq, fs.state.FirstTime = seq, time.Time{}
 	}
 	if seq == fs.state.FirstSeq {
-		fs.state.FirstSeq, fs.state.FirstTime = seq+1, now
+		fs.state.FirstSeq, fs.state.FirstTime = seq+1, time.Time{}
 	}
 	// Mark as dirty for stream state.
 	fs.dirty++
@@ -4346,17 +4348,16 @@ func (fs *fileStore) SkipMsgs(seq uint64, num uint64) error {
 	}
 
 	// Insert into dmap all entries and place last as marker.
-	now := time.Now()
-	nowts := now.UnixNano()
+	now := ats.AccessTime()
 	lseq := seq + num - 1
 
 	mb.mu.Lock()
 	// If we are empty update meta directly.
 	if mb.msgs == 0 {
 		atomic.StoreUint64(&mb.last.seq, lseq)
-		mb.last.ts = nowts
+		mb.last.ts = now
 		atomic.StoreUint64(&mb.first.seq, lseq+1)
-		mb.first.ts = nowts
+		mb.first.ts = 0
 	} else {
 		for ; seq <= lseq; seq++ {
 			mb.dmap.Insert(seq)
@@ -4365,13 +4366,13 @@ func (fs *fileStore) SkipMsgs(seq uint64, num uint64) error {
 	mb.mu.Unlock()
 
 	// Write out our placeholder.
-	mb.writeMsgRecord(emptyRecordLen, lseq|ebit, _EMPTY_, nil, nil, nowts, true)
+	mb.writeMsgRecord(emptyRecordLen, lseq|ebit, _EMPTY_, nil, nil, now, true)
 
 	// Now update FS accounting.
 	// Update fs state.
-	fs.state.LastSeq, fs.state.LastTime = lseq, now
+	fs.state.LastSeq, fs.state.LastTime = lseq, time.Unix(0, now).UTC()
 	if fs.state.Msgs == 0 {
-		fs.state.FirstSeq, fs.state.FirstTime = lseq+1, now
+		fs.state.FirstSeq, fs.state.FirstTime = lseq+1, time.Time{}
 	}
 
 	// Mark as dirty for stream state.
@@ -4801,18 +4802,19 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 	mb.removeSeqPerSubject(sm.subj, seq)
 	fs.removePerSubject(sm.subj)
 
-	if secure {
+	fifo := seq == atomic.LoadUint64(&mb.first.seq)
+	isLastBlock := mb == fs.lmb
+	isEmpty := mb.msgs == 0
+
+	// If erase but block is empty, we can simply remove the block later.
+	if secure && !isEmpty {
 		// Grab record info.
 		ri, rl, _, _ := mb.slotInfo(int(seq - mb.cache.fseq))
-		if err := mb.eraseMsg(seq, int(ri), int(rl)); err != nil {
+		if err := mb.eraseMsg(seq, int(ri), int(rl), isLastBlock); err != nil {
 			return false, err
 		}
 	}
 
-	fifo := seq == atomic.LoadUint64(&mb.first.seq)
-	isLastBlock := mb == fs.lmb
-	isEmpty := mb.msgs == 0
-
 	if fifo {
 		mb.selectNextFirst()
 		if !isEmpty {
@@ -5203,7 +5205,7 @@ func (mb *msgBlock) flushLoop(fch, qch chan struct{}) {
 }
 
 // Lock should be held.
-func (mb *msgBlock) eraseMsg(seq uint64, ri, rl int) error {
+func (mb *msgBlock) eraseMsg(seq uint64, ri, rl int, isLastBlock bool) error {
 	var le = binary.LittleEndian
 	var hdr [msgHdrSize]byte
 
@@ -5245,35 +5247,23 @@ func (mb *msgBlock) eraseMsg(seq uint64, ri, rl int) error {
 
 	// Disk
 	if mb.cache.off+mb.cache.wp > ri {
-		<-dios
-		mfd, err := os.OpenFile(mb.mfn, os.O_RDWR, defaultFilePerms)
-		dios <- struct{}{}
-		if err != nil {
-			return err
-		}
-		defer mfd.Close()
-		if _, err = mfd.WriteAt(nbytes, int64(ri)); err == nil {
-			mfd.Sync()
-		}
-		if err != nil {
+		if err := mb.atomicOverwriteFile(mb.cache.buf, !isLastBlock); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-// Truncate this message block to the storedMsg.
-func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
-	mb.mu.Lock()
-	defer mb.mu.Unlock()
-
+// Truncate this message block to the tseq and ts.
+// Lock should be held.
+func (mb *msgBlock) truncate(tseq uint64, ts int64) (nmsgs, nbytes uint64, err error) {
 	// Make sure we are loaded to process messages etc.
 	if err := mb.loadMsgsWithLock(); err != nil {
 		return 0, 0, err
 	}
 
 	// Calculate new eof using slot info from our new last sm.
-	ri, rl, _, err := mb.slotInfo(int(sm.seq - mb.cache.fseq))
+	ri, rl, _, err := mb.slotInfo(int(tseq - mb.cache.fseq))
 	if err != nil {
 		return 0, 0, err
 	}
@@ -5298,7 +5288,7 @@ func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 	checkDmap := mb.dmap.Size() > 0
 	var smv StoreMsg
 
-	for seq := atomic.LoadUint64(&mb.last.seq); seq > sm.seq; seq-- {
+	for seq := atomic.LoadUint64(&mb.last.seq); seq > tseq; seq-- {
 		if checkDmap {
 			if mb.dmap.Exists(seq) {
 				// Delete and skip to next.
@@ -5332,46 +5322,19 @@ func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 		if err != nil {
 			return 0, 0, fmt.Errorf("failed to load block from disk: %w", err)
 		}
-		if mb.bek != nil && len(buf) > 0 {
-			bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
-			if err != nil {
-				return 0, 0, err
-			}
-			mb.bek = bek
-			mb.bek.XORKeyStream(buf, buf)
+		if err = mb.encryptOrDecryptIfNeeded(buf); err != nil {
+			return 0, 0, err
 		}
-		buf, err = mb.decompressIfNeeded(buf)
-		if err != nil {
+		if buf, err = mb.decompressIfNeeded(buf); err != nil {
 			return 0, 0, fmt.Errorf("failed to decompress block: %w", err)
 		}
 		buf = buf[:eof]
 		copy(mb.lchk[0:], buf[:len(buf)-checksumSize])
-		buf, err = mb.cmp.Compress(buf)
-		if err != nil {
-			return 0, 0, fmt.Errorf("failed to recompress block: %w", err)
-		}
-		meta := &CompressionInfo{
-			Algorithm:    mb.cmp,
-			OriginalSize: uint64(eof),
-		}
-		buf = append(meta.MarshalMetadata(), buf...)
-		if mb.bek != nil && len(buf) > 0 {
-			bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
-			if err != nil {
-				return 0, 0, err
-			}
-			mb.bek = bek
-			mb.bek.XORKeyStream(buf, buf)
-		}
-		n, err := mb.writeAt(buf, 0)
-		if err != nil {
-			return 0, 0, fmt.Errorf("failed to rewrite compressed block: %w", err)
-		}
-		if n != len(buf) {
-			return 0, 0, fmt.Errorf("short write (%d != %d)", n, len(buf))
+		// We did decompress but don't recompress the truncated buffer here since we're the last block
+		// and would otherwise have compressed data and allow to write uncompressed data in the same block.
+		if err = mb.atomicOverwriteFile(buf, false); err != nil {
+			return 0, 0, err
 		}
-		mb.mfd.Truncate(int64(len(buf)))
-		mb.mfd.Sync()
 	} else if mb.mfd != nil {
 		mb.mfd.Truncate(eof)
 		mb.mfd.Sync()
@@ -5384,8 +5347,8 @@ func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 	}
 
 	// Update our last msg.
-	atomic.StoreUint64(&mb.last.seq, sm.seq)
-	mb.last.ts = sm.ts
+	atomic.StoreUint64(&mb.last.seq, tseq)
+	mb.last.ts = ts
 
 	// Clear our cache.
 	mb.clearCacheAndOffset()
@@ -5449,7 +5412,11 @@ func (fs *fileStore) selectNextFirst() {
 		mb := fs.blks[0]
 		mb.mu.RLock()
 		fs.state.FirstSeq = atomic.LoadUint64(&mb.first.seq)
-		fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
+		if mb.first.ts == 0 {
+			fs.state.FirstTime = time.Time{}
+		} else {
+			fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
+		}
 		mb.mu.RUnlock()
 	} else {
 		// Could not find anything, so treat like purge
@@ -6233,13 +6200,9 @@ func (fs *fileStore) writeTombstoneNoFlush(seq uint64, ts int64) error {
 	return lmb.writeTombstoneNoFlush(seq, ts)
 }
 
+// Lock should be held.
 func (mb *msgBlock) recompressOnDiskIfNeeded() error {
 	alg := mb.fs.fcfg.Compression
-	mb.mu.Lock()
-	defer mb.mu.Unlock()
-
-	origFN := mb.mfn                    // The original message block on disk.
-	tmpFN := mb.mfn + compressTmpSuffix // The compressed block will be written here.
 
 	// Open up the file block and read in the entire contents into memory.
 	// One of two things will happen:
@@ -6248,7 +6211,7 @@ func (mb *msgBlock) recompressOnDiskIfNeeded() error {
 	// 2. The block will be uncompressed, in which case we will compress it
 	//    and then write it back out to disk, re-encrypting if necessary.
 	<-dios
-	origBuf, err := os.ReadFile(origFN)
+	origBuf, err := os.ReadFile(mb.mfn)
 	dios <- struct{}{}
 
 	if err != nil {
@@ -6260,13 +6223,8 @@ func (mb *msgBlock) recompressOnDiskIfNeeded() error {
 	// compression can be as efficient as possible on the raw data, whereas
 	// the encrypted ciphertext will not compress anywhere near as well.
 	// The block encryption also covers the optional compression metadata.
-	if mb.bek != nil && len(origBuf) > 0 {
-		bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
-		if err != nil {
-			return err
-		}
-		mb.bek = bek
-		mb.bek.XORKeyStream(origBuf, origBuf)
+	if err = mb.encryptOrDecryptIfNeeded(origBuf); err != nil {
+		return err
 	}
 
 	meta := &CompressionInfo{}
@@ -6291,6 +6249,19 @@ func (mb *msgBlock) recompressOnDiskIfNeeded() error {
 		}
 	}
 
+	return mb.atomicOverwriteFile(origBuf, true)
+}
+
+// Lock should be held.
+func (mb *msgBlock) atomicOverwriteFile(buf []byte, allowCompress bool) error {
+	if mb.mfd != nil {
+		mb.closeFDsLockedNoCheck()
+		defer mb.enableForWriting(mb.fs.fip)
+	}
+
+	origFN := mb.mfn               // The original message block on disk.
+	tmpFN := mb.mfn + blkTmpSuffix // The new block will be written here.
+
 	// Rather than modifying the existing block on disk (which is a dangerous
 	// operation if something goes wrong), create a new temporary file. We will
 	// write out the new block here and then swap the files around afterwards
@@ -6309,41 +6280,37 @@ func (mb *msgBlock) recompressOnDiskIfNeeded() error {
 		return err
 	}
 
-	// The original buffer at this point is uncompressed, so we will now compress
-	// it if needed. Note that if the selected algorithm is NoCompression, the
-	// Compress function will just return the input buffer unmodified.
-	cmpBuf, err := alg.Compress(origBuf)
-	if err != nil {
-		return errorCleanup(fmt.Errorf("failed to compress block: %w", err))
-	}
+	alg := NoCompression
+	if calg := mb.fs.fcfg.Compression; calg != NoCompression && allowCompress {
+		alg = calg
+		// The original buffer at this point is uncompressed, so we will now compress
+		// it if needed. Note that if the selected algorithm is NoCompression, the
+		// Compress function will just return the input buffer unmodified.
+		if buf, err = alg.Compress(buf); err != nil {
+			return errorCleanup(fmt.Errorf("failed to compress block: %w", err))
+		}
 
-	// We only need to write out the metadata header if compression is enabled.
-	// If we're trying to uncompress the file on disk at this point, don't bother
-	// writing metadata.
-	if alg != NoCompression {
+		// We only need to write out the metadata header if compression is enabled.
+		// If we're trying to uncompress the file on disk at this point, don't bother
+		// writing metadata.
 		meta := &CompressionInfo{
 			Algorithm:    alg,
-			OriginalSize: uint64(len(origBuf)),
+			OriginalSize: uint64(len(buf)),
 		}
-		cmpBuf = append(meta.MarshalMetadata(), cmpBuf...)
+		buf = append(meta.MarshalMetadata(), buf...)
 	}
 
 	// Re-encrypt the block if necessary.
-	if mb.bek != nil && len(cmpBuf) > 0 {
-		bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
-		if err != nil {
-			return errorCleanup(err)
-		}
-		mb.bek = bek
-		mb.bek.XORKeyStream(cmpBuf, cmpBuf)
+	if err = mb.encryptOrDecryptIfNeeded(buf); err != nil {
+		return errorCleanup(err)
 	}
 
 	// Write the new block data (which might be compressed or encrypted) to the
 	// temporary file.
-	if n, err := tmpFD.Write(cmpBuf); err != nil {
+	if n, err := tmpFD.Write(buf); err != nil {
 		return errorCleanup(fmt.Errorf("failed to write to temporary file: %w", err))
-	} else if n != len(cmpBuf) {
-		return errorCleanup(fmt.Errorf("short write to temporary file (%d != %d)", n, len(cmpBuf)))
+	} else if n != len(buf) {
+		return errorCleanup(fmt.Errorf("short write to temporary file (%d != %d)", n, len(buf)))
 	}
 	if err := tmpFD.Sync(); err != nil {
 		return errorCleanup(fmt.Errorf("failed to sync temporary file: %w", err))
@@ -6363,11 +6330,12 @@ func (mb *msgBlock) recompressOnDiskIfNeeded() error {
 	mb.cmp = alg
 
 	// Also update rbytes
-	mb.rbytes = uint64(len(cmpBuf))
+	mb.rbytes = uint64(len(buf))
 
 	return nil
 }
 
+// Lock should be held.
 func (mb *msgBlock) decompressIfNeeded(buf []byte) ([]byte, error) {
 	var meta CompressionInfo
 	if n, err := meta.UnmarshalMetadata(buf); err != nil {
@@ -6388,6 +6356,19 @@ func (mb *msgBlock) decompressIfNeeded(buf []byte) ([]byte, error) {
 	}
 }
 
+// Lock should be held.
+func (mb *msgBlock) encryptOrDecryptIfNeeded(buf []byte) error {
+	if mb.bek != nil && len(buf) > 0 {
+		bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
+		if err != nil {
+			return err
+		}
+		mb.bek = bek
+		mb.bek.XORKeyStream(buf, buf)
+	}
+	return nil
+}
+
 // Lock should be held.
 func (mb *msgBlock) ensureRawBytesLoaded() error {
 	if mb.rbytes > 0 {
@@ -6691,7 +6672,7 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 				// If we have a hole fill it.
 				for dseq := mbFirstSeq + uint64(len(idx)); dseq < seq; dseq++ {
 					idx = append(idx, dbit)
-					if dms == 0 {
+					if dms == 0 && dseq != 0 {
 						mb.dmap.Insert(dseq)
 					}
 				}
@@ -6705,7 +6686,7 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 			}
 
 			// Make sure our dmap has this entry if it was erased.
-			if erased && dms == 0 {
+			if erased && dms == 0 && seq != 0 {
 				mb.dmap.Insert(seq)
 			}
 
@@ -7069,15 +7050,9 @@ checkCache:
 	mb.clearCacheAndOffset()
 
 	// Check if we need to decrypt.
-	if mb.bek != nil && len(buf) > 0 {
-		bek, err := genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
-		if err != nil {
-			return err
-		}
-		mb.bek = bek
-		mb.bek.XORKeyStream(buf, buf)
+	if err = mb.encryptOrDecryptIfNeeded(buf); err != nil {
+		return err
 	}
-
 	// Check for compression.
 	if buf, err = mb.decompressIfNeeded(buf); err != nil {
 		return err
@@ -8158,7 +8133,11 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 						firstSeqNeedsUpdate = firstSeqNeedsUpdate || seq == fs.state.FirstSeq
 					} else if seq == fs.state.FirstSeq {
 						fs.state.FirstSeq = atomic.LoadUint64(&mb.first.seq) // new one.
-						fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
+						if mb.first.ts == 0 {
+							fs.state.FirstTime = time.Time{}
+						} else {
+							fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
+						}
 					}
 				} else {
 					// Out of order delete.
@@ -8665,55 +8644,118 @@ func (fs *fileStore) Truncate(seq uint64) error {
 		return ErrStoreSnapshotInProgress
 	}
 
-	nlmb := fs.selectMsgBlock(seq)
-	if nlmb == nil {
-		fs.mu.Unlock()
-		return ErrInvalidSequence
+	var lsm *StoreMsg
+	smb := fs.selectMsgBlock(seq)
+	if smb != nil {
+		lsm, _, _ = smb.fetchMsgNoCopy(seq, nil)
 	}
-	lsm, _, _ := nlmb.fetchMsgNoCopy(seq, nil)
-	if lsm == nil {
-		fs.mu.Unlock()
-		return ErrInvalidSequence
+
+	// Reset last so new block doesn't contain truncated sequences/timestamps.
+	var lastTime int64
+	if lsm != nil {
+		lastTime = lsm.ts
+	} else if smb != nil {
+		lastTime = smb.last.ts
+	} else {
+		lastTime = fs.state.LastTime.UnixNano()
 	}
+	fs.state.LastSeq = seq
+	fs.state.LastTime = time.Unix(0, lastTime).UTC()
 
-	// Set lmb to nlmb and make sure writeable.
-	fs.lmb = nlmb
-	if err := nlmb.enableForWriting(fs.fip); err != nil {
-		fs.mu.Unlock()
+	// Always create a new write block for any tombstones.
+	// We'll truncate the selected message block as the last step, so can't write tombstones to it.
+	// If we end up not needing to write tombstones, this block will be cleaned up at the end.
+	tmb, err := fs.newMsgBlockForWrite()
+	if err != nil {
 		return err
 	}
-	// Collect all tombstones, we want to put these back so we can survive
-	// a restore without index.db properly.
-	var tombs []msgId
-	tombs = append(tombs, nlmb.tombs()...)
-
-	var purged, bytes uint64
 
-	// Truncate our new last message block.
-	nmsgs, nbytes, err := nlmb.truncate(lsm)
-	if err != nil {
-		fs.mu.Unlock()
-		return fmt.Errorf("nlmb.truncate: %w", err)
+	// If the selected block is not found or the message was deleted, we'll need to write a tombstone
+	// at the truncated sequence so we don't roll backward on our last sequence and timestamp.
+	if lsm == nil {
+		fs.writeTombstone(seq, lastTime)
 	}
-	// Account for the truncated msgs and bytes.
-	purged += nmsgs
-	bytes += nbytes
+
+	var purged, bytes uint64
 
 	// Remove any left over msg blocks.
-	getLastMsgBlock := func() *msgBlock { return fs.blks[len(fs.blks)-1] }
-	for mb := getLastMsgBlock(); mb != nlmb; mb = getLastMsgBlock() {
+	getLastMsgBlock := func() *msgBlock {
+		// Start at one before last, tmb will be the last most of the time
+		// unless a new block gets added for tombstones.
+		for i := len(fs.blks) - 2; i >= 0; i-- {
+			if mb := fs.blks[i]; mb.index < tmb.index {
+				return mb
+			}
+		}
+		return nil
+	}
+	for mb := getLastMsgBlock(); mb != nil && mb != smb; mb = getLastMsgBlock() {
 		mb.mu.Lock()
-		// We do this to load tombs.
-		tombs = append(tombs, mb.tombsLocked()...)
 		purged += mb.msgs
 		bytes += mb.bytes
+
+		// We could have tombstones for messages before the truncated sequence.
+		// Need to store those for blocks we're about to remove.
+		if tombs := mb.tombsLocked(); len(tombs) > 0 {
+			// Temporarily unlock while we write tombstones.
+			mb.mu.Unlock()
+			for _, tomb := range tombs {
+				if tomb.seq < seq {
+					fs.writeTombstone(tomb.seq, tomb.ts)
+				}
+			}
+			mb.mu.Lock()
+		}
 		fs.removeMsgBlock(mb)
 		mb.mu.Unlock()
 	}
 
+	hasWrittenTombstones := len(tmb.tombs()) > 0
+	if smb != nil {
+		// Make sure writeable.
+		smb.mu.Lock()
+		if err := smb.enableForWriting(fs.fip); err != nil {
+			smb.mu.Unlock()
+			fs.mu.Unlock()
+			return err
+		}
+
+		// Truncate our selected message block.
+		nmsgs, nbytes, err := smb.truncate(seq, lastTime)
+		if err != nil {
+			smb.mu.Unlock()
+			fs.mu.Unlock()
+			return fmt.Errorf("smb.truncate: %w", err)
+		}
+		// Account for the truncated msgs and bytes.
+		purged += nmsgs
+		bytes += nbytes
+
+		// The selected message block is not the last anymore, need to close down resources.
+		if hasWrittenTombstones {
+			// Quit our loops.
+			if smb.qch != nil {
+				close(smb.qch)
+				smb.qch = nil
+			}
+			smb.closeFDsLockedNoCheck()
+			smb.recompressOnDiskIfNeeded()
+		}
+		smb.mu.Unlock()
+	}
+
+	// If no tombstones were written, we can remove the block and
+	// purely rely on the selected block as the last block.
+	if !hasWrittenTombstones {
+		fs.lmb = smb
+		tmb.mu.Lock()
+		fs.removeMsgBlock(tmb)
+		tmb.mu.Unlock()
+	}
+
 	// Reset last.
-	fs.state.LastSeq = lsm.seq
-	fs.state.LastTime = time.Unix(0, lsm.ts).UTC()
+	fs.state.LastSeq = seq
+	fs.state.LastTime = time.Unix(0, lastTime).UTC()
 	// Update msgs and bytes.
 	if purged > fs.state.Msgs {
 		purged = fs.state.Msgs
@@ -8727,16 +8769,6 @@ func (fs *fileStore) Truncate(seq uint64) error {
 	// Reset our subject lookup info.
 	fs.resetGlobalPerSubjectInfo()
 
-	// Always create new write block.
-	fs.newMsgBlockForWrite()
-
-	// Write any tombstones as needed.
-	for _, tomb := range tombs {
-		if tomb.seq <= lsm.seq {
-			fs.writeTombstone(tomb.seq, tomb.ts)
-		}
-	}
-
 	// Any existing state file no longer applicable. We will force write a new one
 	// after we release the lock.
 	os.Remove(filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile))
diff --git a/server/memstore.go b/server/memstore.go
index 8924a9c1..19bc0bc3 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -23,6 +23,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/nats-io/nats-server/v2/server/ats"
+
 	"github.com/nats-io/nats-server/v2/server/avl"
 	"github.com/nats-io/nats-server/v2/server/gsl"
 	"github.com/nats-io/nats-server/v2/server/stree"
@@ -71,6 +73,9 @@ func newMemStore(cfg *StreamConfig) (*memStore, error) {
 		}
 	}
 
+	// Register with access time service.
+	ats.Register()
+
 	return ms, nil
 }
 
@@ -286,7 +291,7 @@ func (ms *memStore) StoreMsg(subj string, hdr, msg []byte, ttl int64) (uint64, i
 // SkipMsg will use the next sequence number but not store anything.
 func (ms *memStore) SkipMsg() uint64 {
 	// Grab time.
-	now := time.Now().UTC()
+	now := time.Unix(0, ats.AccessTime()).UTC()
 
 	ms.mu.Lock()
 	seq := ms.state.LastSeq + 1
@@ -294,7 +299,7 @@ func (ms *memStore) SkipMsg() uint64 {
 	ms.state.LastTime = now
 	if ms.state.Msgs == 0 {
 		ms.state.FirstSeq = seq + 1
-		ms.state.FirstTime = now
+		ms.state.FirstTime = time.Time{}
 	} else {
 		ms.dmap.Insert(seq)
 	}
@@ -305,7 +310,7 @@ func (ms *memStore) SkipMsg() uint64 {
 // Skip multiple msgs.
 func (ms *memStore) SkipMsgs(seq uint64, num uint64) error {
 	// Grab time.
-	now := time.Now().UTC()
+	now := time.Unix(0, ats.AccessTime()).UTC()
 
 	ms.mu.Lock()
 	defer ms.mu.Unlock()
@@ -322,7 +327,7 @@ func (ms *memStore) SkipMsgs(seq uint64, num uint64) error {
 	ms.state.LastSeq = lseq
 	ms.state.LastTime = now
 	if ms.state.Msgs == 0 {
-		ms.state.FirstSeq, ms.state.FirstTime = lseq+1, now
+		ms.state.FirstSeq, ms.state.FirstTime = lseq+1, time.Time{}
 	} else {
 		for ; seq <= lseq; seq++ {
 			ms.dmap.Insert(seq)
@@ -1297,7 +1302,9 @@ func (ms *memStore) purge(fseq uint64) (uint64, error) {
 	ms.state.FirstTime = time.Time{}
 	ms.state.Bytes = 0
 	ms.state.Msgs = 0
-	ms.msgs = make(map[uint64]*StoreMsg)
+	if ms.msgs != nil {
+		ms.msgs = make(map[uint64]*StoreMsg)
+	}
 	ms.fss = stree.NewSubjectTree[SimpleState]()
 	ms.dmap.Empty()
 	ms.sdm.empty()
@@ -1426,9 +1433,9 @@ func (ms *memStore) Truncate(seq uint64) error {
 
 	ms.mu.Lock()
 	lsm, ok := ms.msgs[seq]
-	if !ok {
-		ms.mu.Unlock()
-		return ErrInvalidSequence
+	lastTime := ms.state.LastTime
+	if ok && lsm != nil {
+		lastTime = time.Unix(0, lsm.ts).UTC()
 	}
 
 	for i := ms.state.LastSeq; i > seq; i-- {
@@ -1443,8 +1450,8 @@ func (ms *memStore) Truncate(seq uint64) error {
 		}
 	}
 	// Reset last.
-	ms.state.LastSeq = lsm.seq
-	ms.state.LastTime = time.Unix(0, lsm.ts).UTC()
+	ms.state.LastSeq = seq
+	ms.state.LastTime = lastTime
 	// Update msgs and bytes.
 	if purged > ms.state.Msgs {
 		purged = ms.state.Msgs
@@ -1936,15 +1943,24 @@ func (ms *memStore) Delete() error {
 }
 
 func (ms *memStore) Stop() error {
-	// These can't come back, so stop is same as Delete.
-	ms.Purge()
 	ms.mu.Lock()
+	if ms.msgs == nil {
+		ms.mu.Unlock()
+		return nil
+	}
 	if ms.ageChk != nil {
 		ms.ageChk.Stop()
 		ms.ageChk = nil
 	}
 	ms.msgs = nil
 	ms.mu.Unlock()
+
+	// These can't come back, so stop is same as Delete.
+	ms.Purge()
+
+	// Unregister from the access time service.
+	ats.Unregister()
+
 	return nil
 }
 
diff --git a/server/raft.go b/server/raft.go
index 4ac669a9..24f0b6bb 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -3339,21 +3339,9 @@ func (n *raft) truncateWAL(term, index uint64) {
 	}()
 
 	if err := n.wal.Truncate(index); err != nil {
-		// If we get an invalid sequence, reset our wal all together.
-		// We will not have holes, so this means we do not have this message stored anymore.
-		// This is normal when truncating back to applied/snapshot.
-		if err == ErrInvalidSequence {
-			n.debug("Clearing WAL")
-			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)
-			}
-		} else {
-			n.warn("Error truncating WAL: %v", err)
-			n.setWriteErrLocked(err)
-			return
-		}
+		n.warn("Error truncating WAL: %v", err)
+		n.setWriteErrLocked(err)
+		return
 	}
 	// Set after we know we have truncated properly.
 	n.pterm, n.pindex = term, index
@@ -4130,7 +4118,6 @@ func (n *raft) setWriteErrLocked(err error) {
 	// Ignore non-write errors.
 	if err == ErrStoreClosed ||
 		err == ErrStoreEOF ||
-		err == ErrInvalidSequence ||
 		err == ErrStoreMsgNotFound ||
 		err == errNoPending ||
 		err == errPartialCache {
diff --git a/server/store.go b/server/store.go
index bdcf758f..65a199b3 100644
--- a/server/store.go
+++ b/server/store.go
@@ -61,8 +61,6 @@ var (
 	ErrStoreWrongType = errors.New("wrong storage type")
 	// ErrNoAckPolicy is returned when trying to update a consumer's acks with no ack policy.
 	ErrNoAckPolicy = errors.New("ack policy is none")
-	// ErrInvalidSequence is returned when the sequence is not present in the stream store.
-	ErrInvalidSequence = errors.New("invalid sequence")
 	// ErrSequenceMismatch is returned when storing a raw message and the expected sequence is wrong.
 	ErrSequenceMismatch = errors.New("expected sequence does not match store")
 	// ErrCorruptStreamState



Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index de6710e8..00c543d1 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -1012,11 +1012,6 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 			t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 		}
 
-		// Check that sequence has to be interior.
-		if err := fs.Truncate(toStore + 1); err != ErrInvalidSequence {
-			t.Fatalf("Expected err of '%v', got '%v'", ErrInvalidSequence, err)
-		}
-
 		if err := fs.Truncate(tseq); err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -9058,7 +9053,7 @@ func TestFileStoreRecoverOnlyBlkFiles(t *testing.T) {
 		// Stop and write some random files, but containing ".blk", should be ignored.
 		require_NoError(t, fs.Stop())
 		require_NoError(t, os.WriteFile(filepath.Join(fs.fcfg.StoreDir, msgDir, "10.blk.random"), nil, defaultFilePerms))
-		require_NoError(t, os.WriteFile(filepath.Join(fs.fcfg.StoreDir, msgDir, fmt.Sprintf("10.blk.%s", compressTmpSuffix)), nil, defaultFilePerms))
+		require_NoError(t, os.WriteFile(filepath.Join(fs.fcfg.StoreDir, msgDir, fmt.Sprintf("10.blk.%s", blkTmpSuffix)), nil, defaultFilePerms))
 
 		fs, err = newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)
 		require_NoError(t, err)
@@ -9908,3 +9903,296 @@ func TestFileStoreAsyncFlushOnSkipMsgs(t *testing.T) {
 		})
 	}
 }
+
+func TestFileStoreCompressionAfterTruncate(t *testing.T) {
+	tests := []struct {
+		title  string
+		action func(fs *fileStore, seq uint64)
+	}{
+		{
+			title: "RemoveMsg",
+			action: func(fs *fileStore, seq uint64) {
+				removed, err := fs.RemoveMsg(seq)
+				require_NoError(t, err)
+				require_True(t, removed)
+			},
+		},
+		{
+			title: "EraseMsg",
+			action: func(fs *fileStore, seq uint64) {
+				erased, err := fs.EraseMsg(seq)
+				require_NoError(t, err)
+				require_True(t, erased)
+			},
+		},
+		{
+			title: "Tombstone",
+			action: func(fs *fileStore, seq uint64) {
+				removed, err := fs.removeMsg(seq, false, false, true)
+				require_NoError(t, err)
+				require_True(t, removed)
+			},
+		},
+	}
+	for _, test := range tests {
+		for _, recompress := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/Recompress=%v", test.title, recompress), func(t *testing.T) {
+				testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
+					cfg := StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage}
+					created := time.Now()
+					fs, err := newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)
+					require_NoError(t, err)
+					defer fs.Stop()
+
+					for range 2 {
+						_, _, err = fs.StoreMsg("foo", nil, nil, 0)
+						require_NoError(t, err)
+					}
+
+					checkCompressed := func(mb *msgBlock) (bool, error) {
+						mb.mu.Lock()
+						defer mb.mu.Unlock()
+						buf, err := mb.loadBlock(nil)
+						if err != nil {
+							return false, err
+						}
+						if err = mb.encryptOrDecryptIfNeeded(buf); err != nil {
+							return false, err
+						}
+						var meta CompressionInfo
+						if n, err := meta.UnmarshalMetadata(buf); err != nil {
+							return false, err
+						} else if n == 0 {
+							return false, nil
+						} else {
+							return meta.Algorithm != NoCompression, nil
+						}
+					}
+
+					smb := fs.getFirstBlock()
+					require_NotNil(t, smb)
+					compressed, err := checkCompressed(smb)
+					require_NoError(t, err)
+					require_False(t, compressed)
+
+					_, err = fs.newMsgBlockForWrite()
+					require_NoError(t, err)
+
+					if fcfg.Compression != NoCompression {
+						checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+							if compressed, err = checkCompressed(smb); err != nil {
+								return err
+							} else if !compressed {
+								return errors.New("block not compressed yet")
+							}
+							return nil
+						})
+					} else {
+						compressed, err = checkCompressed(smb)
+						require_NoError(t, err)
+						require_False(t, compressed)
+					}
+
+					_, _, err = fs.StoreMsg("foo", nil, nil, 0)
+					require_NoError(t, err)
+
+					test.action(fs, 2)
+
+					state := fs.State()
+					require_Equal(t, state.Msgs, 2)
+					require_Equal(t, state.FirstSeq, 1)
+					require_Equal(t, state.LastSeq, 3)
+					require_Equal(t, state.NumDeleted, 1)
+
+					require_NoError(t, fs.Truncate(2))
+					state = fs.State()
+					require_Equal(t, state.Msgs, 1)
+					require_Equal(t, state.FirstSeq, 1)
+					require_Equal(t, state.LastSeq, 2)
+					require_Equal(t, state.NumDeleted, 1)
+
+					fs.mu.RLock()
+					lmb := fs.lmb
+					fs.mu.RUnlock()
+					if smb == lmb {
+						compressed, err = checkCompressed(smb)
+						require_NoError(t, err)
+						require_False(t, compressed)
+					} else {
+						compressed, err = checkCompressed(lmb)
+						require_NoError(t, err)
+						require_False(t, compressed)
+
+						if fcfg.Compression != NoCompression {
+							checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+								if compressed, err = checkCompressed(smb); err != nil {
+									return err
+								} else if !compressed {
+									return errors.New("block not compressed yet")
+								}
+								return nil
+							})
+						}
+					}
+
+					require_NoError(t, fs.forceWriteFullState())
+
+					seq, _, err := fs.StoreMsg("foo", nil, nil, 0)
+					require_NoError(t, err)
+					require_Equal(t, seq, 3)
+					require_NoError(t, fs.forceWriteFullState())
+
+					smb.mu.Lock()
+					smb.clearCacheAndOffset()
+					smb.mu.Unlock()
+
+					require_NoError(t, smb.loadMsgsWithLock())
+					compressed, err = checkCompressed(smb)
+					require_NoError(t, err)
+					if smb == lmb {
+						require_False(t, compressed)
+					} else {
+						require_Equal(t, compressed, fcfg.Compression != NoCompression)
+					}
+
+					compressed, err = checkCompressed(lmb)
+					require_NoError(t, err)
+					require_False(t, compressed)
+				})
+			})
+		}
+	}
+}
+
+func TestFileStoreAtomicEraseMsg(t *testing.T) {
+	for _, lmb := range []bool{true, false} {
+		t.Run(fmt.Sprintf("lmb=%v", lmb), func(t *testing.T) {
+			testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
+				cfg := StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage}
+				created := time.Now()
+				fs, err := newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)
+				require_NoError(t, err)
+				defer fs.Stop()
+
+				for range 3 {
+					_, _, err = fs.StoreMsg("foo", nil, nil, 0)
+					require_NoError(t, err)
+				}
+
+				checkCompressed := func(mb *msgBlock) (bool, error) {
+					mb.mu.Lock()
+					defer mb.mu.Unlock()
+					buf, err := mb.loadBlock(nil)
+					if err != nil {
+						return false, err
+					}
+					if err := mb.checkAndLoadEncryption(); err != nil {
+						return false, err
+					}
+					if err = mb.encryptOrDecryptIfNeeded(buf); err != nil {
+						return false, err
+					}
+					var meta CompressionInfo
+					if n, err := meta.UnmarshalMetadata(buf); err != nil {
+						return false, err
+					} else if n == 0 {
+						return false, nil
+					} else {
+						return meta.Algorithm != NoCompression, nil
+					}
+				}
+
+				if !lmb {
+					smb := fs.getFirstBlock()
+					require_NotNil(t, smb)
+
+					mb, err := fs.newMsgBlockForWrite()
+					require_NoError(t, err)
+					compressed, err := checkCompressed(mb)
+					require_NoError(t, err)
+					require_False(t, compressed)
+
+					if fcfg.Compression != NoCompression {
+						checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+							if compressed, err = checkCompressed(smb); err != nil {
+								return err
+							} else if !compressed {
+								return errors.New("block not compressed yet")
+							}
+							return nil
+						})
+					}
+				}
+
+				before := fs.State()
+				require_Equal(t, before.Msgs, 3)
+				require_Equal(t, before.FirstSeq, 1)
+				require_Equal(t, before.LastSeq, 3)
+				require_Equal(t, before.NumDeleted, 0)
+
+				removed, err := fs.EraseMsg(2)
+				require_NoError(t, err)
+				require_True(t, removed)
+
+				before = fs.State()
+				require_Equal(t, before.Msgs, 2)
+				require_Equal(t, before.FirstSeq, 1)
+				require_Equal(t, before.LastSeq, 3)
+				require_Equal(t, before.NumDeleted, 1)
+
+				seq, _, err := fs.StoreMsg("foo", nil, nil, 0)
+				require_NoError(t, err)
+				require_Equal(t, seq, 4)
+				before = fs.State()
+
+				validateCompressed := func() {
+					t.Helper()
+					fs.mu.Lock()
+					defer fs.mu.Unlock()
+					for _, mb := range fs.blks {
+						compressed, err := checkCompressed(mb)
+						require_NoError(t, err)
+						if mb == fs.lmb {
+							require_False(t, compressed)
+						} else {
+							require_Equal(t, compressed, fs.fcfg.Compression != NoCompression)
+						}
+					}
+				}
+				validateCompressed()
+
+				// Restart should equal before.
+				require_NoError(t, fs.Stop())
+				fs, err = newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)
+				require_NoError(t, err)
+				defer fs.Stop()
+
+				if state := fs.State(); !reflect.DeepEqual(state, before) {
+					t.Fatalf("Expected before of:\n%+v, got:\n%+v", before, state)
+				}
+				validateCompressed()
+
+				// Stop and remove stream before file.
+				require_NoError(t, fs.Stop())
+				require_NoError(t, os.Remove(filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile)))
+
+				// Recovering based on blocks should result in the same before.
+				fs, err = newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)
+				require_NoError(t, err)
+				defer fs.Stop()
+
+				if state := fs.State(); !reflect.DeepEqual(state, before) {
+					t.Fatalf("Expected before of:\n%+v, got:\n%+v", before, state)
+				}
+				validateCompressed()
+
+				// Rebuilding before must also result in the same before.
+				fs.rebuildState(nil)
+				if state := fs.State(); !reflect.DeepEqual(state, before) {
+					t.Fatalf("Expected before of:\n%+v, got:\n%+v", before, state)
+				}
+				validateCompressed()
+			})
+		})
+	}
+}
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 9c06d37d..2ce5f668 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -404,11 +404,6 @@ func TestMemStoreStreamTruncate(t *testing.T) {
 		t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 	}
 
-	// Check that sequence has to be interior.
-	if err := ms.Truncate(toStore + 1); err != ErrInvalidSequence {
-		t.Fatalf("Expected err of '%v', got '%v'", ErrInvalidSequence, err)
-	}
-
 	if err := ms.Truncate(tseq); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
diff --git a/server/norace_1_test.go b/server/norace_1_test.go
index 19aa6bc9..75b6b70c 100644
--- a/server/norace_1_test.go
+++ b/server/norace_1_test.go
@@ -3659,7 +3659,7 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 	fs = o.raftNode().(*raft).wal.(*fileStore)
 	state = fs.State()
 	err = fs.Truncate(state.FirstSeq)
-	require_True(t, err == nil || err == ErrInvalidSequence)
+	require_NoError(t, err)
 	state = fs.State()
 
 	sub, err = js.PullSubscribe("foo", "dlc")
diff --git a/server/store_test.go b/server/store_test.go
index 92714833..8bbcc80f 100644
--- a/server/store_test.go
+++ b/server/store_test.go
@@ -17,6 +17,9 @@ package server
 
 import (
 	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
 	"testing"
 	"time"
 
@@ -585,3 +588,134 @@ func TestStoreUpdateConfigTTLState(t *testing.T) {
 		},
 	)
 }
+
+func TestStoreStreamInteriorDeleteAccounting(t *testing.T) {
+	tests := []struct {
+		title  string
+		action func(s StreamStore, lseq uint64)
+	}{
+		{
+			title: "TruncateWithRemove",
+			action: func(s StreamStore, lseq uint64) {
+				seq, _, err := s.StoreMsg("foo", nil, nil, 0)
+				require_NoError(t, err)
+				require_Equal(t, seq, lseq)
+				removed, err := s.RemoveMsg(lseq)
+				require_NoError(t, err)
+				require_True(t, removed)
+				require_NoError(t, s.Truncate(lseq))
+			},
+		},
+		{
+			title: "TruncateWithErase",
+			action: func(s StreamStore, lseq uint64) {
+				seq, _, err := s.StoreMsg("foo", nil, nil, 0)
+				require_NoError(t, err)
+				require_Equal(t, seq, lseq)
+				removed, err := s.EraseMsg(lseq)
+				require_NoError(t, err)
+				require_True(t, removed)
+				require_NoError(t, s.Truncate(lseq))
+			},
+		},
+		{
+			title: "TruncateWithTombstone",
+			action: func(s StreamStore, lseq uint64) {
+				seq, _, err := s.StoreMsg("foo", nil, nil, 0)
+				require_NoError(t, err)
+				require_Equal(t, seq, lseq)
+				if fs, ok := s.(*fileStore); ok {
+					removed, err := fs.removeMsg(lseq, false, false, true)
+					require_NoError(t, err)
+					require_True(t, removed)
+				} else {
+					removed, err := s.RemoveMsg(lseq)
+					require_NoError(t, err)
+					require_True(t, removed)
+				}
+				require_NoError(t, s.Truncate(lseq))
+			},
+		},
+		{
+			title: "SkipMsg",
+			action: func(s StreamStore, lseq uint64) {
+				s.SkipMsg()
+			},
+		},
+		{
+			title: "SkipMsgs",
+			action: func(s StreamStore, lseq uint64) {
+				require_NoError(t, s.SkipMsgs(lseq, 1))
+			},
+		},
+	}
+	for _, empty := range []bool{false, true} {
+		for _, test := range tests {
+			t.Run(fmt.Sprintf("Empty=%v/%s", empty, test.title), func(t *testing.T) {
+				cfg := StreamConfig{Name: "zzz", Subjects: []string{"foo"}}
+				testAllStoreAllPermutations(t, true, cfg, func(t *testing.T, s StreamStore) {
+					var err error
+					var lseq uint64
+					if !empty {
+						lseq, _, err = s.StoreMsg("foo", nil, nil, 0)
+						require_NoError(t, err)
+						require_Equal(t, lseq, 1)
+					}
+					lseq++
+
+					test.action(s, lseq)
+
+					// Confirm state as baseline.
+					before := s.State()
+					if empty {
+						require_Equal(t, before.Msgs, 0)
+						require_Equal(t, before.FirstSeq, 2)
+						require_Equal(t, before.LastSeq, 1)
+					} else {
+						require_Equal(t, before.Msgs, 1)
+						require_Equal(t, before.FirstSeq, 1)
+						require_Equal(t, before.LastSeq, 2)
+					}
+
+					var fs *fileStore
+					var ok bool
+					if fs, ok = s.(*fileStore); !ok {
+						return
+					}
+					cfg.Storage = FileStorage
+					fcfg := fs.fcfg
+					created := time.Time{}
+
+					// Restart should equal state.
+					require_NoError(t, fs.Stop())
+					fs, err = newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)
+					require_NoError(t, err)
+					defer fs.Stop()
+
+					if state := fs.State(); !reflect.DeepEqual(state, before) {
+						t.Fatalf("Expected state of:\n%+v, got:\n%+v", before, state)
+					}
+
+					// Stop and remove stream state file.
+					require_NoError(t, fs.Stop())
+					require_NoError(t, os.Remove(filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile)))
+
+					// Recovering based on blocks should result in the same state.
+					fs, err = newFileStoreWithCreated(fcfg, cfg, created, prf(&fcfg), nil)
+					require_NoError(t, err)
+					defer fs.Stop()
+
+					if state := fs.State(); !reflect.DeepEqual(state, before) {
+						t.Fatalf("Expected state of:\n%+v, got:\n%+v", before, state)
+					}
+
+					// Rebuilding state must also result in the same state.
+					fs.rebuildState(nil)
+					if state := fs.State(); !reflect.DeepEqual(state, before) {
+						t.Fatalf("Expected state of:\n%+v, got:\n%+v", before, state)
+					}
+				})
+			})
+		}
+	}
+}