Repository: nats-io/nats-server
Pull request #6140 (2024-11-18 09:41:59)

[FIXED] (2.11) Propose message delete for clustered interest stream
For an Interest or WorkQueue stream messages will be removed once all consumers that need to receive a message have acked it.

For a clustered stream each consumer would ack and remove a message by themselves. This can be problematic since that introduces different ordering between servers. For example when using DiscardOld with MaxMsgs, which could result in stream desync. Proposing the message removal ensures ordering between servers.

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

MauriceVanVeen: Putting on draft for now, to look into what happens if we receive it as pre-ack and if it then still correctly does the message delete.
MauriceVanVeen: We've discussed this and agreed on the direction. Will need to look into the upgrade path for this at least.

Patch:
diff --git a/conf/parse.go b/conf/parse.go
index 3e4d7a33..f30812cb 100644
--- a/conf/parse.go
+++ b/conf/parse.go
@@ -26,6 +26,7 @@ package conf
 // see parse_test.go for more examples.
 
 import (
+	"crypto/sha256"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -36,6 +37,8 @@ import (
 	"unicode"
 )
 
+const _EMPTY_ = ""
+
 type parser struct {
 	mapping map[string]any
 	lx      *lexer
@@ -108,6 +111,44 @@ func ParseFileWithChecks(fp string) (map[string]any, error) {
 	return p.mapping, nil
 }
 
+// cleanupUsedEnvVars will recursively remove all already used
+// environment variables which might be in the parsed tree.
+func cleanupUsedEnvVars(m map[string]any) {
+	for k, v := range m {
+		t := v.(*token)
+		if t.usedVariable {
+			delete(m, k)
+			continue
+		}
+		// Cleanup any other env var that is still in the map.
+		if tm, ok := t.value.(map[string]any); ok {
+			cleanupUsedEnvVars(tm)
+		}
+	}
+}
+
+// ParseFileWithChecksDigest returns the processed config and a digest
+// that represents the configuration.
+func ParseFileWithChecksDigest(fp string) (map[string]any, string, error) {
+	data, err := os.ReadFile(fp)
+	if err != nil {
+		return nil, _EMPTY_, err
+	}
+	p, err := parse(string(data), fp, true)
+	if err != nil {
+		return nil, _EMPTY_, err
+	}
+	// Filter out any environment variables before taking the digest.
+	cleanupUsedEnvVars(p.mapping)
+	digest := sha256.New()
+	e := json.NewEncoder(digest)
+	err = e.Encode(p.mapping)
+	if err != nil {
+		return nil, _EMPTY_, err
+	}
+	return p.mapping, fmt.Sprintf("sha256:%x", digest.Sum(nil)), nil
+}
+
 type token struct {
 	item         item
 	value        any
diff --git a/main.go b/main.go
index 657e2020..ee6c105b 100644
--- a/main.go
+++ b/main.go
@@ -110,7 +110,7 @@ func main() {
 	if err != nil {
 		server.PrintAndDie(fmt.Sprintf("%s: %s", exe, err))
 	} else if opts.CheckConfig {
-		fmt.Fprintf(os.Stderr, "%s: configuration file %s is valid\n", exe, opts.ConfigFile)
+		fmt.Fprintf(os.Stderr, "%s: configuration file %s is valid (%s)\n", exe, opts.ConfigFile, opts.ConfigDigest())
 		os.Exit(0)
 	}
 
@@ -120,7 +120,7 @@ func main() {
 		server.PrintAndDie(fmt.Sprintf("%s: %s", exe, err))
 	}
 
-	// Configure the logger based on the flags
+	// Configure the logger based on the flags.
 	s.ConfigureLogger()
 
 	// Start things up. Block here until done.
diff --git a/server/consumer.go b/server/consumer.go
index f9cb4c23..e835c591 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -5977,6 +5977,7 @@ func (o *consumer) isMonitorRunning() bool {
 
 // If we detect that our ackfloor is higher than the stream's last sequence, return this error.
 var errAckFloorHigherThanLastSeq = errors.New("consumer ack floor is higher than streams last sequence")
+var errAckFloorInvalid = errors.New("consumer ack floor is invalid")
 
 // If we are a consumer of an interest or workqueue policy stream, process that state and make sure consistent.
 func (o *consumer) checkStateForInterestStream(ss *StreamState) error {
@@ -6006,7 +6007,7 @@ func (o *consumer) checkStateForInterestStream(ss *StreamState) error {
 	asflr := state.AckFloor.Stream
 	// Protect ourselves against rolling backwards.
 	if asflr&(1<<63) != 0 {
-		return nil
+		return errAckFloorInvalid
 	}
 
 	// Check if the underlying stream's last sequence is less than our floor.
@@ -6025,6 +6026,7 @@ func (o *consumer) checkStateForInterestStream(ss *StreamState) error {
 		fseq = chkfloor
 	}
 
+	var retryAsflr uint64
 	for seq = fseq; asflr > 0 && seq <= asflr; seq++ {
 		if filters != nil {
 			_, nseq, err = store.LoadNextMsgMulti(filters, seq, &smv)
@@ -6037,15 +6039,24 @@ func (o *consumer) checkStateForInterestStream(ss *StreamState) error {
 		}
 		// Only ack though if no error and seq <= ack floor.
 		if err == nil && seq <= asflr {
-			mset.ackMsg(o, seq)
+			didRemove := mset.ackMsg(o, seq)
+			// Removing the message could fail. For example if clustered since we need to propose it.
+			// Overwrite retry floor (only the first time) to allow us to check next time if the removal was successful.
+			if didRemove && retryAsflr == 0 {
+				retryAsflr = seq
+			}
 		}
 	}
+	// If retry floor was not overwritten, set to ack floor+1, we don't need to account for any retries below it.
+	if retryAsflr == 0 {
+		retryAsflr = asflr + 1
+	}
 
 	o.mu.Lock()
 	// Update our check floor.
 	// Check floor must never be greater than ack floor+1, otherwise subsequent calls to this function would skip work.
-	if asflr+1 > o.chkflr {
-		o.chkflr = asflr + 1
+	if retryAsflr > o.chkflr {
+		o.chkflr = retryAsflr
 	}
 	// See if we need to process this update if our parent stream is not a limits policy stream.
 	state, _ = o.store.State()
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index a0e7cb58..38c0c7f0 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -2510,8 +2510,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	var cist *time.Ticker
 	var cistc <-chan time.Time
 
-	// 2 minutes plus up to 30s jitter.
-	checkInterestInterval := 2*time.Minute + time.Duration(rand.Intn(30))*time.Second
+	checkInterestInterval := checkInterestStateT + time.Duration(rand.Intn(checkInterestStateJ))*time.Second
 
 	if mset != nil && mset.isInterestRetention() {
 		// Wait on our consumers to be assigned and running before proceeding.
diff --git a/server/memstore.go b/server/memstore.go
index 48defb80..fcee4ba6 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -568,8 +568,9 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 
 // SubjectsState returns a map of SimpleState for all matching subjects.
 func (ms *memStore) SubjectsState(subject string) map[string]SimpleState {
-	ms.mu.RLock()
-	defer ms.mu.RUnlock()
+	// This needs to be a write lock, as we can mutate the per-subject state.
+	ms.mu.Lock()
+	defer ms.mu.Unlock()
 
 	if ms.fss.Size() == 0 {
 		return nil
@@ -1076,7 +1077,9 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
 		ms.state.FirstSeq = seq
 		ms.state.FirstTime = time.Time{}
 		ms.state.LastSeq = seq - 1
+		// Reset msgs and fss.
 		ms.msgs = make(map[uint64]*StoreMsg)
+		ms.fss = stree.NewSubjectTree[SimpleState]()
 	}
 	ms.mu.Unlock()
 
diff --git a/server/monitor.go b/server/monitor.go
index ca65a670..fcf8242a 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1217,6 +1217,7 @@ type Varz struct {
 	Subscriptions         uint32                 `json:"subscriptions"`
 	HTTPReqStats          map[string]uint64      `json:"http_req_stats"`
 	ConfigLoadTime        time.Time              `json:"config_load_time"`
+	ConfigDigest          string                 `json:"config_digest"`
 	Tags                  jwt.TagList            `json:"tags,omitempty"`
 	TrustedOperatorsJwt   []string               `json:"trusted_operators_jwt,omitempty"`
 	TrustedOperatorsClaim []*jwt.OperatorClaims  `json:"trusted_operators_claim,omitempty"`
@@ -1671,6 +1672,7 @@ func (s *Server) updateVarzConfigReloadableFields(v *Varz) {
 	v.TLSTimeout = opts.TLSTimeout
 	v.WriteDeadline = opts.WriteDeadline
 	v.ConfigLoadTime = s.configTime.UTC()
+	v.ConfigDigest = opts.configDigest
 	// Update route URLs if applicable
 	if s.varzUpdateRouteURLs {
 		v.Cluster.URLs = urlsToStrings(opts.Routes)
diff --git a/server/opts.go b/server/opts.go
index e5794040..17237725 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -449,6 +449,9 @@ type Options struct {
 
 	// Used to mark that we had a top level authorization block.
 	authBlockDefined bool
+
+	// configDigest represents the state of configuration.
+	configDigest string
 }
 
 // WebsocketOpts are options for websocket
@@ -891,10 +894,11 @@ func (o *Options) ProcessConfigFile(configFile string) error {
 	if configFile == _EMPTY_ {
 		return nil
 	}
-	m, err := conf.ParseFileWithChecks(configFile)
+	m, digest, err := conf.ParseFileWithChecksDigest(configFile)
 	if err != nil {
 		return err
 	}
+	o.configDigest = digest
 
 	return o.processConfigFile(configFile, m)
 }
@@ -910,6 +914,11 @@ func (o *Options) ProcessConfigString(data string) error {
 	return o.processConfigFile(_EMPTY_, m)
 }
 
+// ConfigDigest returns the digest representing the configuration.
+func (o *Options) ConfigDigest() string {
+	return o.configDigest
+}
+
 func (o *Options) processConfigFile(configFile string, m map[string]any) error {
 	// Collect all errors and warnings and report them all together.
 	errors := make([]error, 0)
diff --git a/server/reload.go b/server/reload.go
index 03fb06c9..2bc54d7e 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -994,6 +994,13 @@ func (s *Server) Reload() error {
 		// TODO: Dump previous good config to a .bak file?
 		return err
 	}
+
+	// Use the digest from the configuration to detect whether unnecessary to apply reload.
+	if s.getOpts().ConfigDigest() != "" && newOpts.ConfigDigest() == s.getOpts().ConfigDigest() {
+		s.Noticef("Config reload skipped. No changes detected.")
+		return nil
+	}
+
 	return s.ReloadOptions(newOpts)
 }
 
@@ -1627,6 +1634,10 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			if new != old {
 				diffOpts = append(diffOpts, &profBlockRateReload{newValue: new})
 			}
+		case "configdigest":
+			// skip changes in config digest, this is handled already while
+			// processing the config.
+			continue
 		default:
 			// TODO(ik): Implement String() on those options to have a nice print.
 			// %v is difficult to figure what's what, %+v print private fields and
@@ -1774,8 +1785,11 @@ func (s *Server) applyOptions(ctx *reloadContext, opts []option) {
 	if err := s.reloadOCSP(); err != nil {
 		s.Warnf("Can't restart OCSP features: %v", err)
 	}
-
-	s.Noticef("Reloaded server configuration")
+	var cd string
+	if newOpts.configDigest != "" {
+		cd = fmt.Sprintf("(%s)", newOpts.configDigest)
+	}
+	s.Noticef("Reloaded server configuration %s", cd)
 }
 
 // This will send a reset to the internal send loop.
diff --git a/server/server.go b/server/server.go
index f5ca4adc..20f22819 100644
--- a/server/server.go
+++ b/server/server.go
@@ -2243,7 +2243,11 @@ func (s *Server) Start() {
 	}
 
 	if opts.ConfigFile != _EMPTY_ {
-		s.Noticef("Using configuration file: %s", opts.ConfigFile)
+		var cd string
+		if opts.configDigest != "" {
+			cd = fmt.Sprintf("(%s)", opts.configDigest)
+		}
+		s.Noticef("Using configuration file: %s %s", opts.ConfigFile, cd)
 	}
 
 	hasOperators := len(opts.TrustedOperators) > 0
diff --git a/server/stream.go b/server/stream.go
index cb82020a..a1694f95 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -4993,13 +4993,11 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	} else {
 		// Make sure to take into account any message assignments that we had to skip (clfs).
 		seq = lseq + 1 - clfs
-		// Check for preAcks and the need to skip vs store.
+		// Check for preAcks and the need to clear it.
 		if mset.hasAllPreAcks(seq, subject) {
 			mset.clearAllPreAcks(seq)
-			store.SkipMsg()
-		} else {
-			err = store.StoreRawMsg(subject, hdr, msg, seq, ts)
 		}
+		err = store.StoreRawMsg(subject, hdr, msg, seq, ts)
 	}
 
 	if err != nil {
@@ -5679,6 +5677,17 @@ func (mset *stream) getPublicConsumers() []*consumer {
 	return obs
 }
 
+// 2 minutes plus up to 30s jitter.
+const (
+	defaultCheckInterestStateT = 2 * time.Minute
+	defaultCheckInterestStateJ = 30
+)
+
+var (
+	checkInterestStateT = defaultCheckInterestStateT // Interval
+	checkInterestStateJ = defaultCheckInterestStateJ // Jitter (secs)
+)
+
 // Will check for interest retention and make sure messages
 // that have been acked are processed and removed.
 // This will check the ack floors of all consumers, and adjust our first sequence accordingly.
@@ -6026,16 +6035,18 @@ func (mset *stream) clearPreAck(o *consumer, seq uint64) {
 }
 
 // ackMsg is called into from a consumer when we have a WorkQueue or Interest Retention Policy.
-func (mset *stream) ackMsg(o *consumer, seq uint64) {
+// Returns whether the message at seq was removed as a result of the ACK.
+// (Or should be removed in the case of clustered streams, since it requires a message delete proposal)
+func (mset *stream) ackMsg(o *consumer, seq uint64) bool {
 	if seq == 0 {
-		return
+		return false
 	}
 
 	// Don't make this RLock(). We need to have only 1 running at a time to gauge interest across all consumers.
 	mset.mu.Lock()
 	if mset.closed.Load() || mset.cfg.Retention == LimitsPolicy {
 		mset.mu.Unlock()
-		return
+		return false
 	}
 
 	store := mset.store
@@ -6046,7 +6057,9 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
 	if seq > state.LastSeq {
 		mset.registerPreAck(o, seq)
 		mset.mu.Unlock()
-		return
+		// We have not removed the message, but should still signal so we could retry later
+		// since we potentially need to remove it then.
+		return true
 	}
 
 	// Always clear pre-ack if here.
@@ -6055,7 +6068,7 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
 	// Make sure this sequence is not below our first sequence.
 	if seq < state.FirstSeq {
 		mset.mu.Unlock()
-		return
+		return false
 	}
 
 	var shouldRemove bool
@@ -6067,18 +6080,35 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
 	case InterestPolicy:
 		shouldRemove = mset.noInterest(seq, o)
 	}
-	mset.mu.Unlock()
 
 	// If nothing else to do.
 	if !shouldRemove {
-		return
+		mset.mu.Unlock()
+		return false
+	}
+
+	if !mset.isClustered() {
+		mset.mu.Unlock()
+		// If we are here we should attempt to remove.
+		if _, err := store.RemoveMsg(seq); err == ErrStoreEOF {
+			// This should not happen, but being pedantic.
+			mset.registerPreAckLock(o, seq)
+		}
+		return true
 	}
 
-	// If we are here we should attempt to remove.
-	if _, err := store.RemoveMsg(seq); err == ErrStoreEOF {
-		// This should not happen, but being pedantic.
-		mset.registerPreAckLock(o, seq)
+	// Only propose message deletion to the stream if we're consumer leader, otherwise all followers would also propose.
+	// We must be the consumer leader, since we know for sure we've stored the message and don't register as pre-ack.
+	if o != nil && !o.IsLeader() {
+		mset.mu.Unlock()
+		// Must still mark as removal if follower. If we become leader later, we must be able to retry the proposal.
+		return true
 	}
+
+	md := streamMsgDelete{Seq: seq, NoErase: true, Stream: mset.cfg.Name}
+	mset.node.ForwardProposal(encodeMsgDelete(&md))
+	mset.mu.Unlock()
+	return true
 }
 
 // Snapshot creates a snapshot for the stream and possibly consumers.


Test patch:
diff --git a/conf/parse_test.go b/conf/parse_test.go
index a434a618..b7bb7bcd 100644
--- a/conf/parse_test.go
+++ b/conf/parse_test.go
@@ -874,3 +874,143 @@ func TestBlocks(t *testing.T) {
 		})
 	}
 }
+
+func TestParseDigest(t *testing.T) {
+	for _, test := range []struct {
+		input    string
+		includes map[string]string
+		digest   string
+	}{
+		{
+			`foo = bar`,
+			nil,
+			"sha256:226e49e13d16e5e8aa0d62e58cd63361bf097d3e2b2444aa3044334628a2e8de",
+		},
+		{
+			`# Comments and whitespace have no effect
+                        foo = bar
+                        `,
+			nil,
+			"sha256:226e49e13d16e5e8aa0d62e58cd63361bf097d3e2b2444aa3044334628a2e8de",
+		},
+		{
+			`# Syntax changes have no effect
+                        'foo': 'bar'
+                        `,
+			nil,
+			"sha256:226e49e13d16e5e8aa0d62e58cd63361bf097d3e2b2444aa3044334628a2e8de",
+		},
+		{
+			`# Syntax changes have no effect
+                        { 'foo': 'bar' }
+                        `,
+			nil,
+			"sha256:226e49e13d16e5e8aa0d62e58cd63361bf097d3e2b2444aa3044334628a2e8de",
+		},
+		{
+			`# substitutions
+                        BAR_USERS = { users = [ {user = "bar"} ]}
+                        hello = 'world'
+                        accounts {
+                          QUUX_USERS = [ { user: quux }]
+                          bar = $BAR_USERS
+                          quux = { users = $QUUX_USERS }
+                        }
+                        very { nested { env { VAR = 'NESTED', quux = $VAR }}}
+                        `,
+			nil,
+			"sha256:34f8faf3f269fe7509edc4742f20c8c4a7ad51fe21f8b361764314b533ac3ab5",
+		},
+		{
+			`# substitutions, same as previous one without env vars.
+                        hello = 'world'
+                        accounts {
+                          bar  = { users = [ { user = "bar" } ]}
+                          quux = { users = [ { user: quux   } ]}
+                        }
+                        very { nested { env { quux = 'NESTED' }}}
+                        `,
+			nil,
+			"sha256:34f8faf3f269fe7509edc4742f20c8c4a7ad51fe21f8b361764314b533ac3ab5",
+		},
+		{
+			`# substitutions
+                        BAR_USERS = { users = [ {user = "foo"} ]}
+                        bar = $BAR_USERS
+                        accounts {
+                          users = $BAR_USERS
+                        }
+                        `,
+			nil,
+			"sha256:f5d943b4ed22b80c6199203f8a7eaa8eb68ef7b2d46ef6b1b26f05e21f8beb13",
+		},
+		{
+			`# substitutions
+                        bar = { users = [ {user = "foo"} ]}
+                        accounts {
+                          users = { users = [ {user = "foo"} ]}
+                        }
+                        `,
+			nil,
+			"sha256:f5d943b4ed22b80c6199203f8a7eaa8eb68ef7b2d46ef6b1b26f05e21f8beb13",
+		},
+		{
+			`# includes
+			accounts {
+                          foo { include 'foo.conf'}
+                          bar { users = [{user = "bar"}] }
+                          quux { include 'quux.conf'}
+                        }
+                        `,
+			map[string]string{
+				"foo.conf":  ` users = [{user = "foo"}]`,
+				"quux.conf": ` users = [{user = "quux"}]`,
+			},
+			"sha256:e72d70c91b64b0f880f86decb95ec2600cbdcf8bdcd2355fce5ebc54a84a77e9",
+		},
+		{
+			`# includes
+			accounts {
+                          foo { include 'foo.conf'}
+                          bar { include 'bar.conf'}
+                          quux { include 'quux.conf'}
+                        }
+                        `,
+			map[string]string{
+				"foo.conf":  ` users = [{user = "foo"}]`,
+				"bar.conf":  ` users = [{user = "bar"}]`,
+				"quux.conf": ` users = [{user = "quux"}]`,
+			},
+			"sha256:e72d70c91b64b0f880f86decb95ec2600cbdcf8bdcd2355fce5ebc54a84a77e9",
+		},
+	} {
+		t.Run("", func(t *testing.T) {
+			sdir := t.TempDir()
+			f, err := os.CreateTemp(sdir, "nats.conf-")
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := os.WriteFile(f.Name(), []byte(test.input), 066); err != nil {
+				t.Error(err)
+			}
+			if test.includes != nil {
+				for includeFile, contents := range test.includes {
+					inf, err := os.Create(filepath.Join(sdir, includeFile))
+					if err != nil {
+						t.Fatal(err)
+					}
+					if err := os.WriteFile(inf.Name(), []byte(contents), 066); err != nil {
+						t.Error(err)
+					}
+				}
+			}
+			_, digest, err := ParseFileWithChecksDigest(f.Name())
+			if err != nil {
+				t.Fatal(err)
+			}
+			if digest != test.digest {
+				t.Errorf("\ngot: %s\nexpected: %s", digest, test.digest)
+			}
+		})
+	}
+}
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 3453b85c..71fccdba 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -784,6 +784,13 @@ func TestJetStreamClusterConsumerPauseSurvivesRestart(t *testing.T) {
 }
 
 func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
+	checkInterestStateT = 4 * time.Second
+	checkInterestStateJ = 1
+	defer func() {
+		checkInterestStateT = defaultCheckInterestStateT
+		checkInterestStateJ = defaultCheckInterestStateJ
+	}()
+
 	type testParams struct {
 		restartAny       bool
 		restartLeader    bool
@@ -1208,7 +1215,7 @@ func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
 		checkMsgsEqual := func(t *testing.T) {
 			// These have already been checked to be the same for all streams.
 			state := getStreamDetails(t, c.streamLeader("js", sc.Name)).State
-			// Gather all the streams.
+			// Gather the stream mset from each replica.
 			var msets []*stream
 			for _, s := range c.servers {
 				acc, err := s.LookupAccount("js")
@@ -1218,13 +1225,26 @@ func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
 				msets = append(msets, mset)
 			}
 			for seq := state.FirstSeq; seq <= state.LastSeq; seq++ {
+				var expectedErr error
 				var msgId string
 				var smv StoreMsg
 				for _, mset := range msets {
 					mset.mu.RLock()
 					sm, err := mset.store.LoadMsg(seq, &smv)
 					mset.mu.RUnlock()
-					require_NoError(t, err)
+					if err != nil || expectedErr != nil {
+						// If one of the msets reports an error for LoadMsg for this
+						// particular sequence, then the same error should be reported
+						// by all msets for that seq to prove consistency across replicas.
+						// If one of the msets either returns no error or doesn't return
+						// the same error, then that replica has drifted.
+						if expectedErr == nil {
+							expectedErr = err
+						} else {
+							require_Error(t, err, expectedErr)
+						}
+						continue
+					}
 					if msgId == _EMPTY_ {
 						msgId = string(sm.hdr)
 					} else if msgId != string(sm.hdr) {
@@ -4935,3 +4955,135 @@ func TestJetStreamClusterStreamConsumerStateResetAfterRecreate(t *testing.T) {
 	}
 	wg.Wait()
 }
+
+func TestJetStreamClusterStreamAckMsgR1SignalsRemovedMsg(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Subjects:  []string{"foo"},
+		Retention: nats.WorkQueuePolicy,
+		Replicas:  1,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "CONSUMER",
+		Replicas:  1,
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	_, err = js.Publish("foo", nil)
+	require_NoError(t, err)
+
+	s := c.streamLeader(globalAccountName, "TEST")
+	acc, err := s.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+	mset, err := acc.lookupStream("TEST")
+	require_NoError(t, err)
+	o := mset.lookupConsumer("CONSUMER")
+	require_NotNil(t, o)
+
+	// Too high sequence, should register pre-ack and return true allowing for retries.
+	require_True(t, mset.ackMsg(o, 100))
+
+	var smv StoreMsg
+	sm, err := mset.store.LoadMsg(1, &smv)
+	require_NoError(t, err)
+	require_Equal(t, sm.subj, "foo")
+
+	// Now do a proper ack, should immediately remove the message since it's R1.
+	require_True(t, mset.ackMsg(o, 1))
+	_, err = mset.store.LoadMsg(1, &smv)
+	require_Error(t, err, ErrStoreMsgNotFound)
+}
+
+func TestJetStreamClusterStreamAckMsgR3SignalsRemovedMsg(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Subjects:  []string{"foo"},
+		Retention: nats.WorkQueuePolicy,
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "CONSUMER",
+		Replicas:  3,
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	_, err = js.Publish("foo", nil)
+	require_NoError(t, err)
+
+	getStreamAndConsumer := func(s *Server) (*stream, *consumer, error) {
+		t.Helper()
+		acc, err := s.lookupAccount(globalAccountName)
+		if err != nil {
+			return nil, nil, err
+		}
+		mset, err := acc.lookupStream("TEST")
+		if err != nil {
+			return nil, nil, err
+		}
+		o := mset.lookupConsumer("CONSUMER")
+		if err != nil {
+			return nil, nil, err
+		}
+		return mset, o, nil
+	}
+
+	sl := c.consumerLeader(globalAccountName, "TEST", "CONSUMER")
+	sf := c.randomNonConsumerLeader(globalAccountName, "TEST", "CONSUMER")
+
+	msetL, ol, err := getStreamAndConsumer(sl)
+	require_NoError(t, err)
+	msetF, of, err := getStreamAndConsumer(sf)
+	require_NoError(t, err)
+
+	// Too high sequence, should register pre-ack and return true allowing for retries.
+	require_True(t, msetL.ackMsg(ol, 100))
+	require_True(t, msetF.ackMsg(of, 100))
+
+	// Ack message on follower, should not remove message as that's proposed by the leader.
+	// But should still signal message removal.
+	require_True(t, msetF.ackMsg(of, 1))
+
+	// Confirm all servers have the message.
+	var smv StoreMsg
+	for _, s := range c.servers {
+		mset, _, err := getStreamAndConsumer(s)
+		require_NoError(t, err)
+		sm, err := mset.store.LoadMsg(1, &smv)
+		require_NoError(t, err)
+		require_Equal(t, sm.subj, "foo")
+	}
+
+	// Now do a proper ack, should propose the message removal since it's R3.
+	require_True(t, msetL.ackMsg(ol, 1))
+	checkFor(t, 5*time.Second, 200*time.Millisecond, func() error {
+		for _, s := range c.servers {
+			mset, _, err := getStreamAndConsumer(s)
+			if err != nil {
+				return err
+			}
+			_, err = mset.store.LoadMsg(1, &smv)
+			if err != ErrStoreMsgNotFound {
+				return fmt.Errorf("expected error, but got: %v", err)
+			}
+		}
+		return nil
+	})
+}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 0c14b78c..e362be50 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -24688,3 +24688,36 @@ func TestJetStreamDelayedAPIResponses(t *testing.T) {
 	s.sendDelayedAPIErrResponse(nil, acc, "I", _EMPTY_, "request9", "response9", nil, 100*time.Millisecond)
 	check("request9", "response9")
 }
+
+func TestJetStreamMemoryPurgeClearsSubjectsState(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Storage:  nats.MemoryStorage,
+	})
+	require_NoError(t, err)
+
+	pa, err := js.Publish("foo", nil)
+	require_NoError(t, err)
+	require_Equal(t, pa.Sequence, 1)
+
+	// When requesting stream info, we expect one subject foo with one entry.
+	si, err := js.StreamInfo("TEST", &nats.StreamInfoRequest{SubjectsFilter: ">"})
+	require_NoError(t, err)
+	require_Len(t, len(si.State.Subjects), 1)
+	require_Equal(t, si.State.Subjects["foo"], 1)
+
+	// After purging, moving the sequence up, the subjects state should be cleared.
+	err = js.PurgeStream("TEST", &nats.StreamPurgeRequest{Sequence: 100})
+	require_NoError(t, err)
+
+	si, err = js.StreamInfo("TEST", &nats.StreamInfoRequest{SubjectsFilter: ">"})
+	require_NoError(t, err)
+	require_Len(t, len(si.State.Subjects), 0)
+}
diff --git a/server/norace_test.go b/server/norace_test.go
index 19f83c91..ff9917e1 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -7271,17 +7271,25 @@ func TestNoRaceJetStreamInterestStreamCheckInterestRaceBug(t *testing.T) {
 		return nil
 	})
 
-	for _, s := range c.servers {
-		mset, err := s.GlobalAccount().lookupStream("TEST")
-		require_NoError(t, err)
+	checkFor(t, 5*time.Second, time.Second, func() error {
+		for _, s := range c.servers {
+			mset, err := s.GlobalAccount().lookupStream("TEST")
+			if err != nil {
+				return err
+			}
 
-		mset.mu.RLock()
-		defer mset.mu.RUnlock()
+			mset.mu.RLock()
+			defer mset.mu.RUnlock()
 
-		state := mset.state()
-		require_True(t, state.Msgs == 0)
-		require_True(t, state.FirstSeq == uint64(numToSend+1))
-	}
+			state := mset.state()
+			if state.Msgs != 0 {
+				return fmt.Errorf("too many messages: %d", state.Msgs)
+			} else if state.FirstSeq != uint64(numToSend+1) {
+				return fmt.Errorf("wrong FirstSeq: %d, expected: %d", state.FirstSeq, numToSend+1)
+			}
+		}
+		return nil
+	})
 }
 
 func TestNoRaceJetStreamClusterInterestStreamConsistencyAfterRollingRestart(t *testing.T) {
@@ -9817,6 +9825,13 @@ func TestNoRaceJetStreamSnapshotsWithSlowAckDontSlowConsumer(t *testing.T) {
 }
 
 func TestNoRaceJetStreamWQSkippedMsgsOnScaleUp(t *testing.T) {
+	checkInterestStateT = 4 * time.Second
+	checkInterestStateJ = 1
+	defer func() {
+		checkInterestStateT = defaultCheckInterestStateT
+		checkInterestStateJ = defaultCheckInterestStateJ
+	}()
+
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -11189,7 +11204,7 @@ func TestNoRaceJetStreamClusterCheckInterestStatePerformanceInterest(t *testing.
 	}
 
 	require_Equal(t, checkFloor(mset.lookupConsumer("A")), 1)
-	require_Equal(t, checkFloor(mset.lookupConsumer("B")), 100_001)
+	require_Equal(t, checkFloor(mset.lookupConsumer("B")), 90_001)
 	require_Equal(t, checkFloor(mset.lookupConsumer("C")), 100_001)
 
 	// This checks the chkflr state. For this test this should be much faster,
diff --git a/server/opts_test.go b/server/opts_test.go
index 9ae32043..81159129 100644
--- a/server/opts_test.go
+++ b/server/opts_test.go
@@ -121,6 +121,7 @@ func TestConfigFile(t *testing.T) {
 		LameDuckDuration:      4 * time.Minute,
 		ConnectErrorReports:   86400,
 		ReconnectErrorReports: 5,
+		configDigest:          "sha256:314adbd9997c1183f028f5b620362daa45893da76bac746136bfb48b2fd14996",
 		authBlockDefined:      true,
 	}
 
@@ -142,6 +143,7 @@ func TestTLSConfigFile(t *testing.T) {
 		AuthTimeout:      1.0,
 		TLSTimeout:       2.0,
 		authBlockDefined: true,
+		configDigest:     "sha256:cec986d37d7c09c86916d1ba4990cea1b8dadd49c86f9f782b455b47d07c2ac8",
 	}
 	opts, err := ProcessConfigFile("./configs/tls.conf")
 	if err != nil {
@@ -287,6 +289,7 @@ func TestMergeOverrides(t *testing.T) {
 		ConnectErrorReports:   86400,
 		ReconnectErrorReports: 5,
 		authBlockDefined:      true,
+		configDigest:          "sha256:314adbd9997c1183f028f5b620362daa45893da76bac746136bfb48b2fd14996",
 	}
 	fopts, err := ProcessConfigFile("./configs/test.conf")
 	if err != nil {
@@ -362,8 +365,9 @@ func TestRouteFlagOverride(t *testing.T) {
 			Password:    "top_secret",
 			AuthTimeout: 0.5,
 		},
-		Routes:    []*url.URL{rurl},
-		RoutesStr: routeFlag,
+		Routes:       []*url.URL{rurl},
+		RoutesStr:    routeFlag,
+		configDigest: "sha256:fe3c13f82637723989a9bbd0ad6d064b95d48971666af440d4196d9c0d3af979",
 	}
 
 	fopts, err := ProcessConfigFile("./configs/srv_a.conf")
@@ -403,7 +407,8 @@ func TestClusterFlagsOverride(t *testing.T) {
 			Password:    "top_secret",
 			AuthTimeout: 0.5,
 		},
-		Routes: []*url.URL{rurl},
+		Routes:       []*url.URL{rurl},
+		configDigest: "sha256:fe3c13f82637723989a9bbd0ad6d064b95d48971666af440d4196d9c0d3af979",
 	}
 
 	fopts, err := ProcessConfigFile("./configs/srv_a.conf")
@@ -438,8 +443,9 @@ func TestRouteFlagOverrideWithMultiple(t *testing.T) {
 			Password:    "top_secret",
 			AuthTimeout: 0.5,
 		},
-		Routes:    rurls,
-		RoutesStr: routeFlag,
+		Routes:       rurls,
+		RoutesStr:    routeFlag,
+		configDigest: "sha256:fe3c13f82637723989a9bbd0ad6d064b95d48971666af440d4196d9c0d3af979",
 	}
 
 	fopts, err := ProcessConfigFile("./configs/srv_a.conf")
diff --git a/server/reload_test.go b/server/reload_test.go
index 8c3f3f8c..ff393c86 100644
--- a/server/reload_test.go
+++ b/server/reload_test.go
@@ -151,7 +151,8 @@ func TestConfigReloadUnsupported(t *testing.T) {
 			Host: "127.0.0.1",
 			Port: -1,
 		},
-		NoSigs: true,
+		NoSigs:       true,
+		configDigest: "sha256:3c5c4141f56274bcfa801f2d7326ec64d5eabddd9e6d4f9a06ed167315a57f55",
 	}
 	setBaselineOptions(golden)
 
@@ -223,7 +224,8 @@ func TestConfigReloadInvalidConfig(t *testing.T) {
 			Host: "127.0.0.1",
 			Port: -1,
 		},
-		NoSigs: true,
+		NoSigs:       true,
+		configDigest: "sha256:3c5c4141f56274bcfa801f2d7326ec64d5eabddd9e6d4f9a06ed167315a57f55",
 	}
 	setBaselineOptions(golden)
 
@@ -286,7 +288,8 @@ func TestConfigReload(t *testing.T) {
 			Host: "127.0.0.1",
 			Port: server.ClusterAddr().Port,
 		},
-		NoSigs: true,
+		NoSigs:       true,
+		configDigest: "sha256:3c5c4141f56274bcfa801f2d7326ec64d5eabddd9e6d4f9a06ed167315a57f55",
 	}
 	setBaselineOptions(golden)
 
@@ -4054,6 +4057,11 @@ func TestConfigReloadAndVarz(t *testing.T) {
 	if v.MaxConn != DEFAULT_MAX_CONNECTIONS {
 		t.Fatalf("MaxConn should be %v, got %v", DEFAULT_MAX_CONNECTIONS, v.MaxConn)
 	}
+	got := v.ConfigDigest
+	expected := "sha256:7ea4cc5c6864139d814ce940a40ee6546b7ac5285eec3c390a4ce11e34dc1102"
+	if got != expected {
+		t.Fatalf("got: %v, expected: %v", got, expected)
+	}
 
 	changeCurrentConfigContentWithNewContent(t, conf, []byte(fmt.Sprintf(template, "max_connections: 10")))
 
@@ -4071,6 +4079,11 @@ func TestConfigReloadAndVarz(t *testing.T) {
 	if v.MaxConn != 10 {
 		t.Fatalf("MaxConn should be 10, got %v", v.MaxConn)
 	}
+	got = v.ConfigDigest
+	expected = "sha256:508a26309068f62c3022aa8951e712ed8ff5dd6f2360f727ad8242a2a233176e"
+	if got != expected {
+		t.Fatalf("got: %v, expected: %v", got, expected)
+	}
 }
 
 func TestConfigReloadConnectErrReports(t *testing.T) {
diff --git a/server/routes_test.go b/server/routes_test.go
index 671a5e85..01f51be0 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -106,6 +106,7 @@ func TestRouteConfig(t *testing.T) {
 			ConnectRetries: 2,
 		},
 		PidFile:          "/tmp/nats-server/nats_cluster_test.pid",
+		configDigest:     "sha256:1a95b87d99ff3950ff3eb220ef6ffb5387c95fa606ed6976023ade266329c7b5",
 		authBlockDefined: true,
 	}
 
diff --git a/server/signal_test.go b/server/signal_test.go
index 2eab8d4d..0f2480d5 100644
--- a/server/signal_test.go
+++ b/server/signal_test.go
@@ -76,7 +76,14 @@ func TestSignalToReOpenLogFile(t *testing.T) {
 }
 
 func TestSignalToReloadConfig(t *testing.T) {
-	opts, err := ProcessConfigFile("./configs/reload/basic.conf")
+	tmpl := `
+		listen: 127.0.0.1:-1
+		accounts: {
+			A: { users: [ { user: %s, password: foo } ] }
+		}
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, "foo")))
+	opts, err := ProcessConfigFile(conf)
 	if err != nil {
 		t.Fatalf("Error processing config file: %v", err)
 	}
@@ -84,10 +91,21 @@ func TestSignalToReloadConfig(t *testing.T) {
 	s := RunServer(opts)
 	defer s.Shutdown()
 
+	// Check that the reload time does not change when there are no changes.
+	loaded := s.ConfigTime()
+	time.Sleep(500 * time.Millisecond)
+	syscall.Kill(syscall.Getpid(), syscall.SIGHUP)
+	if reloaded := s.ConfigTime(); reloaded.After(loaded) {
+		t.Fatalf("ConfigTime is incorrect.\nexpected no change: %s\ngot: %s", loaded, reloaded)
+	}
+
 	// Repeat test to make sure that server services signals more than once...
 	for i := 0; i < 2; i++ {
 		loaded := s.ConfigTime()
-
+		user := fmt.Sprintf("foo:%d", i)
+		if err := os.WriteFile(conf, []byte(fmt.Sprintf(tmpl, user)), 0666); err != nil {
+			t.Fatalf("Error creating config file: %v", err)
+		}
 		// Wait a bit to ensure ConfigTime changes.
 		time.Sleep(5 * time.Millisecond)