Repository: nats-io/nats-server
Pull request #6376 (2025-01-15 17:22:43)

Stream message proposal flags, allow sourcing/mirroring to preserve TTL headers
This adds new flags to the stream message proposals so that we can distinguish how a message entered a stream, i.e. due to a publish, or because of a sourcing/mirroring arrangement.

Continue to reject TTL'd messages if they are published to a stream with TTLs disabled, but allow us to retain them in a sourcing/mirroring scenario.

Add tests to prove that TTLs being disabled and enabled on sourcing/mirroring configurations does the right thing.

Signed-off-by: Neil Twigg <neil@nats.io>


Patch:
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 500c83ac..76cd5bf0 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -2979,7 +2979,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 					}
 				}
 
-				subject, reply, hdr, msg, lseq, ts, err := decodeStreamMsg(mbuf)
+				subject, reply, hdr, msg, lseq, ts, sourced, err := decodeStreamMsg(mbuf)
 				if err != nil {
 					if node := mset.raftNode(); node != nil {
 						s.Errorf("JetStream cluster could not decode stream msg for '%s > %s' [%s]",
@@ -3036,7 +3036,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 					mt = mset.getAndDeleteMsgTrace(lseq)
 				}
 				// Process the actual message here.
-				err = mset.processJetStreamMsg(subject, reply, hdr, msg, lseq, ts, mt)
+				err = mset.processJetStreamMsg(subject, reply, hdr, msg, lseq, ts, mt, sourced)
 
 				// If we have inflight make sure to clear after processing.
 				// TODO(dlc) - technically check on inflight != nil could cause datarace.
@@ -3069,7 +3069,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 						if state.Msgs == 0 {
 							mset.store.Compact(lseq + 1)
 							// Retry
-							err = mset.processJetStreamMsg(subject, reply, hdr, msg, lseq, ts, mt)
+							err = mset.processJetStreamMsg(subject, reply, hdr, msg, lseq, ts, mt, sourced)
 						}
 						// FIXME(dlc) - We could just run a catchup with a request defining the span between what we expected
 						// and what we got.
@@ -7696,10 +7696,10 @@ func decodeConsumerAssignmentCompressed(buf []byte) (*consumerAssignment, error)
 
 var errBadStreamMsg = errors.New("jetstream cluster bad replicated stream msg")
 
-func decodeStreamMsg(buf []byte) (subject, reply string, hdr, msg []byte, lseq uint64, ts int64, err error) {
+func decodeStreamMsg(buf []byte) (subject, reply string, hdr, msg []byte, lseq uint64, ts int64, sourced bool, err error) {
 	var le = binary.LittleEndian
 	if len(buf) < 26 {
-		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, errBadStreamMsg
+		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, false, errBadStreamMsg
 	}
 	lseq = le.Uint64(buf)
 	buf = buf[8:]
@@ -7708,44 +7708,49 @@ func decodeStreamMsg(buf []byte) (subject, reply string, hdr, msg []byte, lseq u
 	sl := int(le.Uint16(buf))
 	buf = buf[2:]
 	if len(buf) < sl {
-		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, errBadStreamMsg
+		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, false, errBadStreamMsg
 	}
 	subject = string(buf[:sl])
 	buf = buf[sl:]
 	if len(buf) < 2 {
-		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, errBadStreamMsg
+		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, false, errBadStreamMsg
 	}
 	rl := int(le.Uint16(buf))
 	buf = buf[2:]
 	if len(buf) < rl {
-		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, errBadStreamMsg
+		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, false, errBadStreamMsg
 	}
 	reply = string(buf[:rl])
 	buf = buf[rl:]
 	if len(buf) < 2 {
-		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, errBadStreamMsg
+		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, false, errBadStreamMsg
 	}
 	hl := int(le.Uint16(buf))
 	buf = buf[2:]
 	if len(buf) < hl {
-		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, errBadStreamMsg
+		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, false, errBadStreamMsg
 	}
 	if hdr = buf[:hl]; len(hdr) == 0 {
 		hdr = nil
 	}
 	buf = buf[hl:]
 	if len(buf) < 4 {
-		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, errBadStreamMsg
+		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, false, errBadStreamMsg
 	}
 	ml := int(le.Uint32(buf))
 	buf = buf[4:]
 	if len(buf) < ml {
-		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, errBadStreamMsg
+		return _EMPTY_, _EMPTY_, nil, nil, 0, 0, false, errBadStreamMsg
 	}
 	if msg = buf[:ml]; len(msg) == 0 {
 		msg = nil
 	}
-	return subject, reply, hdr, msg, lseq, ts, nil
+	buf = buf[ml:]
+	if len(buf) > 0 {
+		flags, _ := binary.Uvarint(buf)
+		sourced = flags&msgFlagFromSourceOrMirror != 0
+	}
+	return subject, reply, hdr, msg, lseq, ts, sourced, nil
 }
 
 // Helper to return if compression allowed.
@@ -7755,8 +7760,13 @@ func (mset *stream) compressAllowed() bool {
 	return mset.compressOK
 }
 
-func encodeStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int64) []byte {
-	return encodeStreamMsgAllowCompress(subject, reply, hdr, msg, lseq, ts, false)
+// Flags for encodeStreamMsg/decodeStreamMsg.
+const (
+	msgFlagFromSourceOrMirror uint64 = 1 << iota
+)
+
+func encodeStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int64, sourced bool) []byte {
+	return encodeStreamMsgAllowCompress(subject, reply, hdr, msg, lseq, ts, sourced, false)
 }
 
 // Threshold for compression.
@@ -7764,7 +7774,7 @@ func encodeStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int
 const compressThreshold = 8192 // 8k
 
 // If allowed and contents over the threshold we will compress.
-func encodeStreamMsgAllowCompress(subject, reply string, hdr, msg []byte, lseq uint64, ts int64, compressOK bool) []byte {
+func encodeStreamMsgAllowCompress(subject, reply string, hdr, msg []byte, lseq uint64, ts int64, sourced bool, compressOK bool) []byte {
 	// Clip the subject, reply, header and msgs down. Operate on
 	// uint64 lengths to avoid overflowing.
 	slen := min(uint64(len(subject)), math.MaxUint16)
@@ -7775,7 +7785,12 @@ func encodeStreamMsgAllowCompress(subject, reply string, hdr, msg []byte, lseq u
 
 	shouldCompress := compressOK && total > compressThreshold
 	elen := int(1 + 8 + 8 + total)
-	elen += (2 + 2 + 2 + 4) // Encoded lengths, 4bytes
+	elen += (2 + 2 + 2 + 4 + 8) // Encoded lengths, 4bytes, flags are up to 8 bytes
+
+	var flags uint64
+	if sourced {
+		flags |= msgFlagFromSourceOrMirror
+	}
 
 	buf := make([]byte, 1, elen)
 	buf[0] = byte(streamMsgOp)
@@ -7791,6 +7806,7 @@ func encodeStreamMsgAllowCompress(subject, reply string, hdr, msg []byte, lseq u
 	buf = append(buf, hdr[:hlen]...)
 	buf = le.AppendUint32(buf, uint32(mlen))
 	buf = append(buf, msg[:mlen]...)
+	buf = binary.AppendUvarint(buf, flags)
 
 	// Check if we should compress.
 	if shouldCompress {
@@ -7904,7 +7920,7 @@ func (mset *stream) checkAllowMsgCompress(peers []string) {
 const streamLagWarnThreshold = 10_000
 
 // processClusteredInboundMsg will propose the inbound message to the underlying raft group.
-func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg []byte, mt *msgTrace) (retErr error) {
+func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg []byte, mt *msgTrace, sourced bool) (retErr error) {
 	// For possible error response.
 	var response []byte
 
@@ -7922,7 +7938,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	// We also invoke this in clustering mode for message tracing when not
 	// performing message delivery.
 	if node == nil || mt.traceOnly() {
-		return mset.processJetStreamMsg(subject, reply, hdr, msg, 0, 0, mt)
+		return mset.processJetStreamMsg(subject, reply, hdr, msg, 0, 0, mt, sourced)
 	}
 
 	// If message tracing (with message delivery), we will need to send the
@@ -8050,7 +8066,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 		}
 
 		// TTL'd messages are rejected entirely if TTLs are not enabled on the stream.
-		if ttl, _ := getMessageTTL(hdr); ttl != 0 && !allowTTL {
+		if ttl, _ := getMessageTTL(hdr); !sourced && ttl != 0 && !allowTTL {
 			if canRespond {
 				var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 				resp.Error = NewJSMessageTTLDisabledError()
@@ -8190,7 +8206,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 		}
 	}
 
-	esm := encodeStreamMsgAllowCompress(subject, reply, hdr, msg, mset.clseq, time.Now().UnixNano(), compressOK)
+	esm := encodeStreamMsgAllowCompress(subject, reply, hdr, msg, mset.clseq, time.Now().UnixNano(), sourced, compressOK)
 	var mtKey uint64
 	if mt != nil {
 		mtKey = mset.clseq
@@ -8711,7 +8727,7 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 		}
 	}
 
-	subj, _, hdr, msg, seq, ts, err := decodeStreamMsg(mbuf)
+	subj, _, hdr, msg, seq, ts, _, err := decodeStreamMsg(mbuf)
 	if err != nil {
 		return 0, errCatchupBadMsg
 	}
@@ -9181,7 +9197,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 		sendDR := func() {
 			if dr.Num == 1 {
 				// Send like a normal skip msg.
-				sendEM(encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, dr.First, 0))
+				sendEM(encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, dr.First, 0, false))
 			} else {
 				// We have a run, send a gap record. We send these without reply or tracking.
 				s.sendInternalMsgLocked(sendSubject, _EMPTY_, nil, encodeDeleteRange(&dr))
@@ -9261,7 +9277,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 					sendDR()
 				}
 				// Send the normal message now.
-				sendEM(encodeStreamMsgAllowCompress(sm.subj, _EMPTY_, sm.hdr, sm.msg, sm.seq, sm.ts, compressOk))
+				sendEM(encodeStreamMsgAllowCompress(sm.subj, _EMPTY_, sm.hdr, sm.msg, sm.seq, sm.ts, false, compressOk))
 			} else {
 				if drOk {
 					if dr.First == 0 {
@@ -9271,7 +9287,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 					}
 				} else {
 					// Skip record for deleted msg.
-					sendEM(encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq, 0))
+					sendEM(encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq, 0, false))
 				}
 			}
 
diff --git a/server/stream.go b/server/stream.go
index 9b182687..39a212e3 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -2531,10 +2531,10 @@ func (mset *stream) processInboundMirrorMsg(m *inMsg) bool {
 			s.resourcesExceededError()
 			err = ApiErrors[JSInsufficientResourcesErr]
 		} else {
-			err = node.Propose(encodeStreamMsg(m.subj, _EMPTY_, m.hdr, m.msg, sseq-1, ts))
+			err = node.Propose(encodeStreamMsg(m.subj, _EMPTY_, m.hdr, m.msg, sseq-1, ts, true))
 		}
 	} else {
-		err = mset.processJetStreamMsg(m.subj, _EMPTY_, m.hdr, m.msg, sseq-1, ts, nil)
+		err = mset.processJetStreamMsg(m.subj, _EMPTY_, m.hdr, m.msg, sseq-1, ts, nil, true)
 	}
 	if err != nil {
 		if strings.Contains(err.Error(), "no space left") {
@@ -2613,7 +2613,7 @@ func (mset *stream) skipMsgs(start, end uint64) {
 	// With syncRequest was easy to add bool into request.
 	var entries []*Entry
 	for seq := start; seq <= end; seq++ {
-		entries = append(entries, newEntry(EntryNormal, encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq-1, 0)))
+		entries = append(entries, newEntry(EntryNormal, encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq-1, 0, false)))
 		// So a single message does not get too big.
 		if len(entries) > 10_000 {
 			node.ProposeMulti(entries)
@@ -3386,7 +3386,7 @@ func (mset *stream) sendFlowControlReply(reply string) {
 func (mset *stream) handleFlowControl(m *inMsg) {
 	// If we are clustered we will send the flow control message through the replication stack.
 	if mset.isClustered() {
-		mset.node.Propose(encodeStreamMsg(_EMPTY_, m.rply, m.hdr, nil, 0, 0))
+		mset.node.Propose(encodeStreamMsg(_EMPTY_, m.rply, m.hdr, nil, 0, 0, false))
 	} else {
 		mset.outq.sendMsg(m.rply, nil)
 	}
@@ -3492,9 +3492,9 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
 	var err error
 	// If we are clustered we need to propose this message to the underlying raft group.
 	if node != nil {
-		err = mset.processClusteredInboundMsg(m.subj, _EMPTY_, hdr, msg, nil)
+		err = mset.processClusteredInboundMsg(m.subj, _EMPTY_, hdr, msg, nil, true)
 	} else {
-		err = mset.processJetStreamMsg(m.subj, _EMPTY_, hdr, msg, 0, 0, nil)
+		err = mset.processJetStreamMsg(m.subj, _EMPTY_, hdr, msg, 0, 0, nil, true)
 	}
 
 	if err != nil {
@@ -4666,7 +4666,7 @@ var (
 )
 
 // processJetStreamMsg is where we try to actually process the stream msg.
-func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int64, mt *msgTrace) (retErr error) {
+func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int64, mt *msgTrace, sourced bool) (retErr error) {
 	if mt != nil {
 		// Only the leader/standalone will have mt!=nil. On exit, send the
 		// message trace event.
@@ -4804,7 +4804,7 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 		// TTL'd messages are rejected entirely if TTLs are not enabled on the stream.
 		// Shouldn't happen in clustered mode since we should have already caught this
 		// in processClusteredInboundMsg, but needed here for non-clustered etc.
-		if ttl, _ := getMessageTTL(hdr); ttl != 0 && !mset.cfg.AllowMsgTTL {
+		if ttl, _ := getMessageTTL(hdr); !sourced && ttl != 0 && !mset.cfg.AllowMsgTTL {
 			mset.mu.Unlock()
 			bumpCLFS()
 			if canRespond {
@@ -5504,9 +5504,9 @@ func (mset *stream) internalLoop() {
 			for _, im := range ims {
 				// If we are clustered we need to propose this message to the underlying raft group.
 				if isClustered {
-					mset.processClusteredInboundMsg(im.subj, im.rply, im.hdr, im.msg, im.mt)
+					mset.processClusteredInboundMsg(im.subj, im.rply, im.hdr, im.msg, im.mt, false)
 				} else {
-					mset.processJetStreamMsg(im.subj, im.rply, im.hdr, im.msg, 0, 0, im.mt)
+					mset.processJetStreamMsg(im.subj, im.rply, im.hdr, im.msg, 0, 0, im.mt, false)
 				}
 				im.returnToPool()
 			}


Test patch:
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index e6c245e5..821c5e1d 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -6801,7 +6801,7 @@ func TestJetStreamClusterCatchupLoadNextMsgTooManyDeletes(t *testing.T) {
 			msg, err := sub.NextMsg(time.Second)
 			require_NoError(t, err)
 			require_Equal(t, entryOp(msg.Data[0]), streamMsgOp)
-			subj, _, _, _, seq, ts, err := decodeStreamMsg(msg.Data[1:])
+			subj, _, _, _, seq, ts, _, err := decodeStreamMsg(msg.Data[1:])
 			require_NoError(t, err)
 			require_Equal(t, seq, 5)
 			require_Equal(t, subj, _EMPTY_)
@@ -6856,7 +6856,7 @@ func TestJetStreamClusterCatchupMustStallWhenBehindOnApplies(t *testing.T) {
 	msg, err := sub.NextMsg(time.Second)
 	require_NoError(t, err)
 	require_Equal(t, entryOp(msg.Data[0]), streamMsgOp)
-	subj, _, _, _, seq, _, err := decodeStreamMsg(msg.Data[1:])
+	subj, _, _, _, seq, _, _, err := decodeStreamMsg(msg.Data[1:])
 	require_NoError(t, err)
 	require_Equal(t, seq, 1)
 	require_Equal(t, subj, "foo")
@@ -6893,7 +6893,7 @@ func TestJetStreamClusterCatchupMustStallWhenBehindOnApplies(t *testing.T) {
 	msg, err = sub.NextMsg(time.Second)
 	require_NoError(t, err)
 	require_Equal(t, entryOp(msg.Data[0]), streamMsgOp)
-	subj, _, _, _, seq, _, err = decodeStreamMsg(msg.Data[1:])
+	subj, _, _, _, seq, _, _, err = decodeStreamMsg(msg.Data[1:])
 	require_NoError(t, err)
 	require_Equal(t, seq, 1)
 	require_Equal(t, subj, "foo")
diff --git a/server/jetstream_cluster_2_test.go b/server/jetstream_cluster_2_test.go
index 28e30035..e5d041fb 100644
--- a/server/jetstream_cluster_2_test.go
+++ b/server/jetstream_cluster_2_test.go
@@ -7063,7 +7063,7 @@ func TestJetStreamClusterLeaderAbortsCatchupOnFollowerError(t *testing.T) {
 	}
 
 	// Now send a message with a wrong sequence and expect to receive an error.
-	em := encodeStreamMsg("foo", _EMPTY_, nil, []byte("fail"), 102, time.Now().UnixNano())
+	em := encodeStreamMsg("foo", _EMPTY_, nil, []byte("fail"), 102, time.Now().UnixNano(), false)
 	leader.sendInternalMsgLocked(sreqSubj, syncRepl.Subject, nil, em)
 	msg = natsNexMsg(t, syncRepl, time.Second)
 	if len(msg.Data) == 0 {
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 4086ae24..394a8e3d 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -4259,7 +4259,7 @@ func TestJetStreamClusterPreserveWALDuringCatchupWithMatchingTerm(t *testing.T)
 						break
 					}
 
-					esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, i, ts, true)
+					esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, i, ts, true, false)
 					entries := []*Entry{newEntry(EntryNormal, esm)}
 					rn.Lock()
 					ae := rn.buildAppendEntry(entries)
@@ -4692,7 +4692,7 @@ func TestJetStreamClusterDontInstallSnapshotWhenStoppingStream(t *testing.T) {
 	validateStreamState(snap)
 
 	// Simulate a message being stored, but not calling Applied yet.
-	err = mset.processJetStreamMsg("foo", _EMPTY_, nil, nil, 1, time.Now().UnixNano(), nil)
+	err = mset.processJetStreamMsg("foo", _EMPTY_, nil, nil, 1, time.Now().UnixNano(), nil, false)
 	require_NoError(t, err)
 
 	// Simulate the stream being stopped before we're able to call Applied.
@@ -5333,6 +5333,140 @@ func TestJetStreamClusterRoutedAPIRecoverPerformance(t *testing.T) {
 	t.Logf("Took %s to clear %d items", time.Since(start), count)
 }
 
+func TestJetStreamClusterMessageTTLWhenSourcing(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:        "Origin",
+		Storage:     FileStorage,
+		Subjects:    []string{"test"},
+		AllowMsgTTL: true,
+		Replicas:    3,
+	})
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:    "TTLEnabled",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			{Name: "Origin"},
+		},
+		AllowMsgTTL: true,
+		Replicas:    3,
+	})
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:    "TTLDisabled",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			{Name: "Origin"},
+		},
+		AllowMsgTTL: false,
+		Replicas:    3,
+	})
+
+	hdr := nats.Header{}
+	hdr.Add("Nats-TTL", "1s")
+
+	_, err := js.PublishMsg(&nats.Msg{
+		Subject: "test",
+		Header:  hdr,
+	})
+	require_NoError(t, err)
+
+	for _, stream := range []string{"TTLEnabled", "TTLDisabled"} {
+		t.Run(stream, func(t *testing.T) {
+			sc, err := js.PullSubscribe("test", "consumer", nats.BindStream(stream))
+			require_NoError(t, err)
+
+			msgs, err := sc.Fetch(1)
+			require_NoError(t, err)
+			require_Len(t, len(msgs), 1)
+			require_Equal(t, msgs[0].Header.Get(JSMessageTTL), "1s")
+
+			time.Sleep(time.Second)
+
+			si, err := js.StreamInfo(stream)
+			require_NoError(t, err)
+			if stream == "TTLDisabled" {
+				require_Equal(t, si.State.Msgs, 1)
+			} else {
+				require_Equal(t, si.State.Msgs, 0)
+			}
+		})
+	}
+}
+
+func TestJetStreamClusterMessageTTLWhenMirroring(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:        "Origin",
+		Storage:     FileStorage,
+		Subjects:    []string{"test"},
+		AllowMsgTTL: true,
+		Replicas:    3,
+	})
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:    "TTLEnabled",
+		Storage: FileStorage,
+		Mirror: &StreamSource{
+			Name: "Origin",
+		},
+		AllowMsgTTL: true,
+		Replicas:    3,
+	})
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:    "TTLDisabled",
+		Storage: FileStorage,
+		Mirror: &StreamSource{
+			Name: "Origin",
+		},
+		AllowMsgTTL: false,
+		Replicas:    3,
+	})
+
+	hdr := nats.Header{}
+	hdr.Add("Nats-TTL", "1s")
+
+	_, err := js.PublishMsg(&nats.Msg{
+		Subject: "test",
+		Header:  hdr,
+	})
+	require_NoError(t, err)
+
+	for _, stream := range []string{"TTLEnabled", "TTLDisabled"} {
+		t.Run(stream, func(t *testing.T) {
+			sc, err := js.PullSubscribe("test", "consumer", nats.BindStream(stream))
+			require_NoError(t, err)
+
+			msgs, err := sc.Fetch(1)
+			require_NoError(t, err)
+			require_Len(t, len(msgs), 1)
+			require_Equal(t, msgs[0].Header.Get(JSMessageTTL), "1s")
+
+			time.Sleep(time.Second)
+
+			si, err := js.StreamInfo(stream)
+			require_NoError(t, err)
+			if stream == "TTLDisabled" {
+				require_Equal(t, si.State.Msgs, 1)
+			} else {
+				require_Equal(t, si.State.Msgs, 0)
+			}
+		})
+	}
+}
+
 func TestJetStreamClusterMessageTTLDisabled(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
@@ -5358,7 +5492,6 @@ func TestJetStreamClusterMessageTTLDisabled(t *testing.T) {
 
 	// In clustered mode we should have caught this before generating the
 	// proposal, therefore the CLFS should not have been bumped by this.
-
 	for _, s := range c.servers {
 		stream, err := s.GlobalAccount().lookupStream("TEST")
 		require_NoError(t, err)
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index 167708e8..86ed6675 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -1266,6 +1266,8 @@ func jsClientConnectURL(t testing.TB, url string, opts ...nats.Option) (*nats.Co
 
 // jsStreamCreate is for sending a stream create for fields that nats.go does not know about yet.
 func jsStreamCreate(t testing.TB, nc *nats.Conn, cfg *StreamConfig) *StreamConfig {
+	t.Helper()
+
 	j, err := json.Marshal(cfg)
 	require_NoError(t, err)
 
@@ -1280,6 +1282,8 @@ func jsStreamCreate(t testing.TB, nc *nats.Conn, cfg *StreamConfig) *StreamConfi
 
 // jsStreamUpdate is for sending a stream create for fields that nats.go does not know about yet.
 func jsStreamUpdate(t testing.TB, nc *nats.Conn, cfg *StreamConfig) (*StreamConfig, error) {
+	t.Helper()
+
 	j, err := json.Marshal(cfg)
 	require_NoError(t, err)
 
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 1f16ef27..26d6ebb6 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -25142,6 +25142,133 @@ func TestJetStreamMessageTTLDisabled(t *testing.T) {
 
 	msg.Header.Set("Nats-TTL", "1s")
 	_, err := js.PublishMsg(msg)
-
 	require_Error(t, err)
 }
+
+func TestJetStreamMessageTTLWhenSourcing(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:        "Origin",
+		Storage:     FileStorage,
+		Subjects:    []string{"test"},
+		AllowMsgTTL: true,
+	})
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:    "TTLEnabled",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			{Name: "Origin"},
+		},
+		AllowMsgTTL: true,
+	})
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:    "TTLDisabled",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			{Name: "Origin"},
+		},
+		AllowMsgTTL: false,
+	})
+
+	hdr := nats.Header{}
+	hdr.Add("Nats-TTL", "1s")
+
+	_, err := js.PublishMsg(&nats.Msg{
+		Subject: "test",
+		Header:  hdr,
+	})
+	require_NoError(t, err)
+
+	for _, stream := range []string{"TTLEnabled", "TTLDisabled"} {
+		t.Run(stream, func(t *testing.T) {
+			sc, err := js.PullSubscribe("test", "consumer", nats.BindStream(stream))
+			require_NoError(t, err)
+
+			msgs, err := sc.Fetch(1)
+			require_NoError(t, err)
+			require_Len(t, len(msgs), 1)
+			require_Equal(t, msgs[0].Header.Get(JSMessageTTL), "1s")
+
+			time.Sleep(time.Second)
+
+			si, err := js.StreamInfo(stream)
+			require_NoError(t, err)
+			if stream == "TTLDisabled" {
+				require_Equal(t, si.State.Msgs, 1)
+			} else {
+				require_Equal(t, si.State.Msgs, 0)
+			}
+		})
+	}
+}
+
+func TestJetStreamMessageTTLSWhenMirroring(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:        "Origin",
+		Storage:     FileStorage,
+		Subjects:    []string{"test"},
+		AllowMsgTTL: true,
+	})
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:    "TTLEnabled",
+		Storage: FileStorage,
+		Mirror: &StreamSource{
+			Name: "Origin",
+		},
+		AllowMsgTTL: true,
+	})
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:    "TTLDisabled",
+		Storage: FileStorage,
+		Mirror: &StreamSource{
+			Name: "Origin",
+		},
+		AllowMsgTTL: false,
+	})
+
+	hdr := nats.Header{}
+	hdr.Add("Nats-TTL", "1s")
+
+	_, err := js.PublishMsg(&nats.Msg{
+		Subject: "test",
+		Header:  hdr,
+	})
+	require_NoError(t, err)
+
+	for _, stream := range []string{"TTLEnabled", "TTLDisabled"} {
+		t.Run(stream, func(t *testing.T) {
+			sc, err := js.PullSubscribe("test", "consumer", nats.BindStream(stream))
+			require_NoError(t, err)
+
+			msgs, err := sc.Fetch(1)
+			require_NoError(t, err)
+			require_Len(t, len(msgs), 1)
+			require_Equal(t, msgs[0].Header.Get(JSMessageTTL), "1s")
+
+			time.Sleep(time.Second)
+
+			si, err := js.StreamInfo(stream)
+			require_NoError(t, err)
+			if stream == "TTLDisabled" {
+				require_Equal(t, si.State.Msgs, 1)
+			} else {
+				require_Equal(t, si.State.Msgs, 0)
+			}
+		})
+	}
+}
diff --git a/server/raft_test.go b/server/raft_test.go
index c760664c..e51837ff 100644
--- a/server/raft_test.go
+++ b/server/raft_test.go
@@ -995,7 +995,7 @@ func TestNRGWALEntryWithoutQuorumMustTruncate(t *testing.T) {
 
 			// Simulate leader storing an AppendEntry in WAL but being hard killed before it can propose to its peers.
 			n := rg.leader().node().(*raft)
-			esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+			esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 			entries := []*Entry{newEntry(EntryNormal, esm)}
 			n.Lock()
 			ae := n.buildAppendEntry(entries)
@@ -1066,7 +1066,7 @@ func TestNRGTermNoDecreaseAfterWALReset(t *testing.T) {
 	l.term = 20
 	l.Unlock()
 
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 	l.Lock()
 	ae := l.buildAppendEntry(entries)
@@ -1102,7 +1102,7 @@ func TestNRGCatchupDoesNotTruncateUncommittedEntriesWithQuorum(t *testing.T) {
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1183,7 +1183,7 @@ func TestNRGCatchupCanTruncateMultipleEntriesWithoutQuorum(t *testing.T) {
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1282,7 +1282,7 @@ func TestNRGCatchupDoesNotTruncateCommittedEntriesDuringRedelivery(t *testing.T)
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1333,7 +1333,7 @@ func TestNRGCatchupFromNewLeaderWithIncorrectPterm(t *testing.T) {
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1379,7 +1379,7 @@ func TestNRGDontRemoveSnapshotIfTruncateToApplied(t *testing.T) {
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1543,7 +1543,7 @@ func TestNRGCancelCatchupWhenDetectingHigherTermDuringVoteRequest(t *testing.T)
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1590,7 +1590,7 @@ func TestNRGTruncateDownToCommitted(t *testing.T) {
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1651,7 +1651,7 @@ func TestNRGTruncateDownToCommittedWhenTruncateFails(t *testing.T) {
 	n.Unlock()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1716,7 +1716,7 @@ func TestNRGMemoryWALEmptiesSnapshotsDir(t *testing.T) {
 	defer c.shutdown()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1761,7 +1761,7 @@ func TestNRGHealthCheckWaitForCatchup(t *testing.T) {
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1820,7 +1820,7 @@ func TestNRGHealthCheckWaitForDoubleCatchup(t *testing.T) {
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"
@@ -1902,7 +1902,7 @@ func TestNRGHealthCheckWaitForPendingCommitsWhenPaused(t *testing.T) {
 	defer cleanup()
 
 	// Create a sample entry, the content doesn't matter, just that it's stored.
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true, false)
 	entries := []*Entry{newEntry(EntryNormal, esm)}
 
 	nats0 := "S1Nunr6R" // "nats-0"