Nov 3, 2025 · Nov 19, 2025
diff --git a/server/filestore.go b/server/filestore.go
 mb.ttls++
 }
 if mb.fs.scheduling != nil {
 if schedule, ok :=getMessageSchedule(hdr); ok && !schedule.IsZero() {
 if schedule, ok :=nextMessageSchedule(hdr, ts); ok && !schedule.IsZero() {
 mb.fs.scheduling.add(seq, string(subj), schedule.UnixNano())
 mb.schedules++
 }
 if len(msg.hdr) == 0 {
 continue
 }
 if schedule, ok :=getMessageSchedule(sm.hdr); ok && !schedule.IsZero() {
 if schedule, ok :=nextMessageSchedule(sm.hdr, sm.ts); ok && !schedule.IsZero() {
 fs.scheduling.init(seq, sm.subj, schedule.UnixNano())
 }
 }

 // Message scheduling.
 if fs.scheduling != nil {
 if schedule, ok :=getMessageSchedule(hdr); ok && !schedule.IsZero() {
 if schedule, ok :=nextMessageSchedule(hdr, ts); ok && !schedule.IsZero() {
 fs.scheduling.add(seq, subj, schedule.UnixNano())
 fs.lmb.schedules++
 } else {
 fs.scheduling.removeSubject(subj)
 }

 // Check for a repeating schedule and update such that it triggers again.
 if scheduleNext := bytesToString(sliceHeader(JSScheduleNext, hdr)); scheduleNext != _EMPTY_ && scheduleNext != JSScheduleNextPurge {
 scheduler := getMessageScheduler(hdr)
 if next, err := time.Parse(time.RFC3339Nano, scheduleNext); err == nil && scheduler != _EMPTY_ {
 fs.scheduling.update(scheduler, next.UnixNano())
 }
 }
 }

 return nil
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
 }
 }

 func TestJetStreamClusterScheduledIntervalMessage(t *testing.T) {
 for _, replicas := range []int{1, 3} {
 for _, storage := range []StorageType{FileStorage, MemoryStorage} {
 t.Run(fmt.Sprintf("R%d/%s", replicas, storage), func(t *testing.T) {
 c := createJetStreamClusterExplicit(t, "R3S", 3)
 defer c.shutdown()

 nc, js := jsClientConnect(t, c.randomServer())
 defer nc.Close()

 cfg := &StreamConfig{
 Name:              "SchedulesEnabled",
 Subjects:          []string{"foo.*"},
 Storage:           storage,
 Replicas:          replicas,
 AllowMsgSchedules: true,
 }
 _, err := jsStreamCreate(t, nc, cfg)
 require_NoError(t, err)

 // Schedule with an interval that triggers often.
 m := nats.NewMsg("foo.schedule")
 m.Header.Set("Nats-Schedule", "@every 1s")
 m.Header.Set("Nats-Schedule-Target", "foo.publish")
 pubAck, err := js.PublishMsg(m)
 require_NoError(t, err)
 require_Equal(t, pubAck.Sequence, 1)

 sl := c.streamLeader(globalAccountName, "SchedulesEnabled")
 mset, err := sl.globalAccount().lookupStream("SchedulesEnabled")
 require_NoError(t, err)

 // Waiting for the repeated message to be published enough times.
 checkFor(t, 4*time.Second, 200*time.Millisecond, func() error {
 total := mset.Store().SubjectsTotals("foo.publish")["foo.publish"]
 if total < 3 {
 return fmt.Errorf("expected at least 3 messages, got %d", total)
 }
 return nil
 })

 // Servers should be synced.
 checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
 return checkState(t, c, globalAccountName, "SchedulesEnabled")
 })

 // If we remove the schedule, the scheduling should stop.
 var removed bool
 total := mset.Store().SubjectsTotals("foo.publish")["foo.publish"]
 for i := 1; ; i++ {
 before := total
 time.Sleep(1200 * time.Millisecond)
 total = mset.Store().SubjectsTotals("foo.publish")["foo.publish"]
 if before == total {
 // If the scheduling stopped before we removed, that's a problem.
 require_True(t, removed)
 break
 }
 if !removed {
 removed = true
 require_NoError(t, js.DeleteMsg("SchedulesEnabled", pubAck.Sequence))
 }
 if i > 2 {
 t.Fatal("Scheduling didn't stop")
 }
 }

 // Servers should be synced.
 checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
 return checkState(t, c, globalAccountName, "SchedulesEnabled")
 })
 })
 }
 }
 }

 func TestJetStreamClusterOfflineStreamAndConsumerAfterAssetCreateOrUpdate(t *testing.T) {
 clusterName := "R3S"
 c := createJetStreamClusterExplicit(t, clusterName, 3)
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
 }
 }

 func TestJetStreamScheduledMessageParse(t *testing.T) {
 // @at <ts>
 ts := time.Now().UTC()
 sts, repeat, ok := parseMsgSchedule(fmt.Sprintf("@at %s", ts.Format(time.RFC3339Nano)), 0)
 require_True(t, ok)
 require_False(t, repeat)
 require_Equal(t, ts, sts)

 // @every <duration>
 now := time.Now().UTC().Round(time.Second)
 sts, repeat, ok = parseMsgSchedule("@every 5s", now.UnixNano())
 require_True(t, ok)
 require_True(t, repeat)
 require_Equal(t, now.Add(5*time.Second), sts)

 // A schedule on an interval should not spam loads of times if it hasn't run in a long while.
 sts, repeat, ok = parseMsgSchedule("@every 5s", 0)
 require_True(t, ok)
 require_True(t, repeat)
 require_True(t, sts.After(time.Unix(0, 0).UTC().Add(5*time.Second)))

 // A schedule can only run at least once every second.
 _, _, ok = parseMsgSchedule("@every 999ms", 0)
 require_False(t, ok)
 }

 func TestJetStreamDirectGetBatchParallelWriteDeadlock(t *testing.T) {
 s := RunBasicJetStreamServer(t)
 defer s.Shutdown()
diff --git a/server/memstore.go b/server/memstore.go
 if len(sm.hdr) == 0 {
 continue
 }
 if schedule, ok :=getMessageSchedule(sm.hdr); ok && !schedule.IsZero() {
 if schedule, ok :=nextMessageSchedule(sm.hdr, sm.ts); ok && !schedule.IsZero() {
 ms.scheduling.init(seq, sm.subj, schedule.UnixNano())
 }
 }

 // Message scheduling.
 if ms.scheduling != nil {
 if schedule, ok :=getMessageSchedule(hdr); ok && !schedule.IsZero() {
 if schedule, ok :=nextMessageSchedule(hdr, ts); ok && !schedule.IsZero() {
 ms.scheduling.add(seq, subj, schedule.UnixNano())
 } else {
 ms.scheduling.removeSubject(subj)
 }

 // Check for a repeating schedule and update such that it triggers again.
 if scheduleNext := bytesToString(sliceHeader(JSScheduleNext, hdr)); scheduleNext != _EMPTY_ && scheduleNext != JSScheduleNextPurge {
 scheduler := getMessageScheduler(hdr)
 if next, err := time.Parse(time.RFC3339Nano, scheduleNext); err == nil && scheduler != _EMPTY_ {
 ms.scheduling.update(scheduler, next.UnixNano())
 }
 }
 }

 return nil
diff --git a/server/scheduler.go b/server/scheduler.go
 "io"
 "math"
 "slices"
 "strings"
 "time"

 "github.com/nats-io/nats-server/v2/server/thw"
 delete(ms.inflight,subj)
 }

 func (ms*MsgScheduling)update(subjstring,tsint64) {
 ifsched,ok:=ms.schedules[subj];ok {
 // Remove and add separately, it's for the same sequence, but if replicated
 // this server could not know the previous timestamp yet.
 ms.ttls.Remove(sched.seq,sched.ts)
 ms.ttls.Add(sched.seq,ts)
 sched.ts=ts
 delete(ms.inflight,subj)
 ms.resetTimer()
 }
 }

 func (ms*MsgScheduling)markInflight(subjstring) {
 if_,ok:=ms.schedules[subj];ok {
 ms.inflight[subj]=struct{}{}
 returnfalse
 }
 // Validate the contents are correct if not, we just remove it from THW.
 pattern:=bytesToString(sliceHeader(JSSchedulePattern,sm.hdr))
 ifpattern==_EMPTY_ {
 ms.remove(seq)
 returntrue
 }
 next,repeat,ok:=parseMsgSchedule(pattern,ts)
 if!ok {
 ms.remove(seq)
 returntrue
 }
 ttl,ok:=getMessageScheduleTTL(sm.hdr)
 if!ok {
 ms.remove(seq)

 // Add headers for the scheduled message.
 hdr=genHeader(hdr,JSScheduler,sm.subj)
 hdr=genHeader(hdr,JSScheduleNext,JSScheduleNextPurge)// Purge the schedule message itself.
 if!repeat {
 hdr=genHeader(hdr,JSScheduleNext,JSScheduleNextPurge)// Purge the schedule message itself.
 }else {
 hdr=genHeader(hdr,JSScheduleNext,next.Format(time.RFC3339))// Next time the schedule fires.
 }
 ifttl!=_EMPTY_ {
 hdr=genHeader(hdr,JSMessageTTL,ttl)
 }
 }
 returnstamp,nil
 }

 // parseMsgSchedule parses a message schedule pattern and returns the time
 // to fire, whether it is a repeating schedule, and whether the pattern was valid.
 funcparseMsgSchedule(patternstring,tsint64) (time.Time,bool,bool) {
 ifpattern==_EMPTY_ {
 return time.Time{},false,true
 }
 // Exact time.
 ifstrings.HasPrefix(pattern,"@at ") {
 t,err:=time.Parse(time.RFC3339,pattern[4:])
 returnt,false,err==nil
 }
 // Repeating on a simple interval.
 ifstrings.HasPrefix(pattern,"@every ") {
 dur,err:=time.ParseDuration(pattern[7:])
 iferr!=nil {
 return time.Time{},false,false
 }
 // Only allow intervals of at least a second.
 ifdur.Seconds()<1 {
 return time.Time{},false,false
 }
 // If this schedule would trigger multiple times, for example after a restart, skip ahead and only fire once.
 next:=time.Unix(0,ts).UTC().Round(time.Second).Add(dur)
 ifnow:=time.Now().UTC().Round(time.Second);next.Before(now) {
 next=now
 }
 returnnext,true,true
 }
 return time.Time{},false,false

 }
diff --git a/server/stream.go b/server/stream.go
 return time.Time{}, true
 }
 val := bytesToString(sliceHeader(JSSchedulePattern, hdr))
 if val == _EMPTY_ {
 schedule, _, ok := parseMsgSchedule(val, time.Now().UTC().UnixNano())
 return schedule, ok
 }

 // Fast lookup and calculation of next message schedule.
 func nextMessageSchedule(hdr []byte, ts int64) (time.Time, bool) {
 if len(hdr) == 0 {
 return time.Time{}, true
 }
 if !strings.HasPrefix(val, "@at ") {
 return time.Time{}, false
 }
 t, err := time.Parse(time.RFC3339, val[4:])
 return t, err == nil
 val := bytesToString(sliceHeader(JSSchedulePattern, hdr))
 schedule, _, ok := parseMsgSchedule(val, ts)
 return schedule, ok
 }

 // Fast lookup of the message schedule TTL from headers.
Original file line number	Diff line number	Diff line change
Expand Up		@@ -1646,7 +1646,7 @@ func (mb msgBlock) rebuildStateLocked() (LostStreamData, []uint64, error) {
		mb.ttls++
		}
		if mb.fs.scheduling != nil {
		if schedule, ok :=getMessageSchedule(hdr); ok && !schedule.IsZero() {
		if schedule, ok :=nextMessageSchedule(hdr, ts); ok && !schedule.IsZero() {
		mb.fs.scheduling.add(seq, string(subj), schedule.UnixNano())
		mb.schedules++
		}
Expand DownExpand Up		@@ -2170,7 +2170,7 @@ func (fs *fileStore) recoverMsgSchedulingState() error {
		if len(msg.hdr) == 0 {
		continue
		}
		if schedule, ok :=getMessageSchedule(sm.hdr); ok && !schedule.IsZero() {
		if schedule, ok :=nextMessageSchedule(sm.hdr, sm.ts); ok && !schedule.IsZero() {
		fs.scheduling.init(seq, sm.subj, schedule.UnixNano())
		}
		}
Expand DownExpand Up		@@ -4566,12 +4566,20 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts, t

		// Message scheduling.
		if fs.scheduling != nil {
		if schedule, ok :=getMessageSchedule(hdr); ok && !schedule.IsZero() {
		if schedule, ok :=nextMessageSchedule(hdr, ts); ok && !schedule.IsZero() {
		fs.scheduling.add(seq, subj, schedule.UnixNano())
		fs.lmb.schedules++
		} else {
		fs.scheduling.removeSubject(subj)
		}

		// Check for a repeating schedule and update such that it triggers again.
		if scheduleNext := bytesToString(sliceHeader(JSScheduleNext, hdr)); scheduleNext != _EMPTY_ && scheduleNext != JSScheduleNextPurge {
		scheduler := getMessageScheduler(hdr)
		if next, err := time.Parse(time.RFC3339Nano, scheduleNext); err == nil && scheduler != _EMPTY_ {
		fs.scheduling.update(scheduler, next.UnixNano())
		}
		}
		}

		return nil
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -9524,6 +9524,82 @@ func TestJetStreamClusterScheduledDelayedMessageReversedHeaderOrder(t *testing.T
		}
		}

		func TestJetStreamClusterScheduledIntervalMessage(t *testing.T) {
		for _, replicas := range []int{1, 3} {
		for _, storage := range []StorageType{FileStorage, MemoryStorage} {
		t.Run(fmt.Sprintf("R%d/%s", replicas, storage), func(t *testing.T) {
		c := createJetStreamClusterExplicit(t, "R3S", 3)
		defer c.shutdown()

		nc, js := jsClientConnect(t, c.randomServer())
		defer nc.Close()

		cfg := &StreamConfig{
		Name: "SchedulesEnabled",
		Subjects: []string{"foo.*"},
		Storage: storage,
		Replicas: replicas,
		AllowMsgSchedules: true,
		}
		_, err := jsStreamCreate(t, nc, cfg)
		require_NoError(t, err)

		// Schedule with an interval that triggers often.
		m := nats.NewMsg("foo.schedule")
		m.Header.Set("Nats-Schedule", "@every 1s")
		m.Header.Set("Nats-Schedule-Target", "foo.publish")
		pubAck, err := js.PublishMsg(m)
		require_NoError(t, err)
		require_Equal(t, pubAck.Sequence, 1)

		sl := c.streamLeader(globalAccountName, "SchedulesEnabled")
		mset, err := sl.globalAccount().lookupStream("SchedulesEnabled")
		require_NoError(t, err)

		// Waiting for the repeated message to be published enough times.
		checkFor(t, 4time.Second, 200time.Millisecond, func() error {
		total := mset.Store().SubjectsTotals("foo.publish")["foo.publish"]
		if total < 3 {
		return fmt.Errorf("expected at least 3 messages, got %d", total)
		}
		return nil
		})

		// Servers should be synced.
		checkFor(t, 2time.Second, 200time.Millisecond, func() error {
		return checkState(t, c, globalAccountName, "SchedulesEnabled")
		})

		// If we remove the schedule, the scheduling should stop.
		var removed bool
		total := mset.Store().SubjectsTotals("foo.publish")["foo.publish"]
		for i := 1; ; i++ {
		before := total
		time.Sleep(1200 * time.Millisecond)
		total = mset.Store().SubjectsTotals("foo.publish")["foo.publish"]
		if before == total {
		// If the scheduling stopped before we removed, that's a problem.
		require_True(t, removed)
		break
		}
		if !removed {
		removed = true
		require_NoError(t, js.DeleteMsg("SchedulesEnabled", pubAck.Sequence))
		}
		if i > 2 {
		t.Fatal("Scheduling didn't stop")
		}
		}

		// Servers should be synced.
		checkFor(t, 2time.Second, 200time.Millisecond, func() error {
		return checkState(t, c, globalAccountName, "SchedulesEnabled")
		})
		})
		}
		}
		}

		func TestJetStreamClusterOfflineStreamAndConsumerAfterAssetCreateOrUpdate(t *testing.T) {
		clusterName := "R3S"
		c := createJetStreamClusterExplicit(t, clusterName, 3)
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -21978,6 +21978,32 @@ func TestJetStreamScheduledMessageNotDeactivated(t *testing.T) {
		}
		}

		func TestJetStreamScheduledMessageParse(t *testing.T) {
		// @at <ts>
		ts := time.Now().UTC()
		sts, repeat, ok := parseMsgSchedule(fmt.Sprintf("@at %s", ts.Format(time.RFC3339Nano)), 0)
		require_True(t, ok)
		require_False(t, repeat)
		require_Equal(t, ts, sts)

		// @every <duration>
		now := time.Now().UTC().Round(time.Second)
		sts, repeat, ok = parseMsgSchedule("@every 5s", now.UnixNano())
		require_True(t, ok)
		require_True(t, repeat)
		require_Equal(t, now.Add(5*time.Second), sts)

		// A schedule on an interval should not spam loads of times if it hasn't run in a long while.
		sts, repeat, ok = parseMsgSchedule("@every 5s", 0)
		require_True(t, ok)
		require_True(t, repeat)
		require_True(t, sts.After(time.Unix(0, 0).UTC().Add(5*time.Second)))

		// A schedule can only run at least once every second.
		_, _, ok = parseMsgSchedule("@every 999ms", 0)
		require_False(t, ok)
		}

		func TestJetStreamDirectGetBatchParallelWriteDeadlock(t *testing.T) {
		s := RunBasicJetStreamServer(t)
		defer s.Shutdown()
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -185,7 +185,7 @@ func (ms *memStore) recoverMsgSchedulingState() {
		if len(sm.hdr) == 0 {
		continue
		}
		if schedule, ok :=getMessageSchedule(sm.hdr); ok && !schedule.IsZero() {
		if schedule, ok :=nextMessageSchedule(sm.hdr, sm.ts); ok && !schedule.IsZero() {
		ms.scheduling.init(seq, sm.subj, schedule.UnixNano())
		}
		}
Expand DownExpand Up		@@ -310,11 +310,19 @@ func (ms *memStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts, tt

		// Message scheduling.
		if ms.scheduling != nil {
		if schedule, ok :=getMessageSchedule(hdr); ok && !schedule.IsZero() {
		if schedule, ok :=nextMessageSchedule(hdr, ts); ok && !schedule.IsZero() {
		ms.scheduling.add(seq, subj, schedule.UnixNano())
		} else {
		ms.scheduling.removeSubject(subj)
		}

		// Check for a repeating schedule and update such that it triggers again.
		if scheduleNext := bytesToString(sliceHeader(JSScheduleNext, hdr)); scheduleNext != _EMPTY_ && scheduleNext != JSScheduleNextPurge {
		scheduler := getMessageScheduler(hdr)
		if next, err := time.Parse(time.RFC3339Nano, scheduleNext); err == nil && scheduler != _EMPTY_ {
		ms.scheduling.update(scheduler, next.UnixNano())
		}
		}
		}

		return nil
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -19,6 +19,7 @@ import (
		"io"
		"math"
		"slices"
		"strings"
		"time"

		"github.com/nats-io/nats-server/v2/server/thw"
Expand DownExpand Up		@@ -77,6 +78,18 @@ func (ms *MsgScheduling) init(seq uint64, subj string, ts int64) {
		delete(ms.inflight,subj)
		}

		func (ms*MsgScheduling)update(subjstring,tsint64) {
		ifsched,ok:=ms.schedules[subj];ok {
		// Remove and add separately, it's for the same sequence, but if replicated
		// this server could not know the previous timestamp yet.
		ms.ttls.Remove(sched.seq,sched.ts)
		ms.ttls.Add(sched.seq,ts)
		sched.ts=ts
		delete(ms.inflight,subj)
		ms.resetTimer()
		}
		}

		func (ms*MsgScheduling)markInflight(subjstring) {
		if_,ok:=ms.schedules[subj];ok {
		ms.inflight[subj]=struct{}{}
Expand DownExpand Up		@@ -159,6 +172,16 @@ func (ms MsgScheduling) getScheduledMessages(loadMsg func(seq uint64, smv Stor
		returnfalse
		}
		// Validate the contents are correct if not, we just remove it from THW.
		pattern:=bytesToString(sliceHeader(JSSchedulePattern,sm.hdr))
		ifpattern==_EMPTY_ {
		ms.remove(seq)
		returntrue
		}
		next,repeat,ok:=parseMsgSchedule(pattern,ts)
		if!ok {
		ms.remove(seq)
		returntrue
		}
		ttl,ok:=getMessageScheduleTTL(sm.hdr)
		if!ok {
		ms.remove(seq)
Expand All		@@ -184,7 +207,11 @@ func (ms MsgScheduling) getScheduledMessages(loadMsg func(seq uint64, smv Stor

		// Add headers for the scheduled message.
		hdr=genHeader(hdr,JSScheduler,sm.subj)
		hdr=genHeader(hdr,JSScheduleNext,JSScheduleNextPurge)// Purge the schedule message itself.
		if!repeat {
		hdr=genHeader(hdr,JSScheduleNext,JSScheduleNextPurge)// Purge the schedule message itself.
		}else {
		hdr=genHeader(hdr,JSScheduleNext,next.Format(time.RFC3339))// Next time the schedule fires.
		}
		ifttl!=_EMPTY_ {
		hdr=genHeader(hdr,JSMessageTTL,ttl)
		}
Expand DownExpand Up		@@ -261,3 +288,35 @@ func (ms *MsgScheduling) decode(b []byte) (uint64, error) {
		}
		returnstamp,nil
		}

		// parseMsgSchedule parses a message schedule pattern and returns the time
		// to fire, whether it is a repeating schedule, and whether the pattern was valid.
		funcparseMsgSchedule(patternstring,tsint64) (time.Time,bool,bool) {
		ifpattern==_EMPTY_ {
		return time.Time{},false,true
		}
		// Exact time.
		ifstrings.HasPrefix(pattern,"@at ") {
		t,err:=time.Parse(time.RFC3339,pattern[4:])
		returnt,false,err==nil
		}
		// Repeating on a simple interval.
		ifstrings.HasPrefix(pattern,"@every ") {
		dur,err:=time.ParseDuration(pattern[7:])
		iferr!=nil {
		return time.Time{},false,false
		}
		// Only allow intervals of at least a second.
		ifdur.Seconds()<1 {
		return time.Time{},false,false
		}
		// If this schedule would trigger multiple times, for example after a restart, skip ahead and only fire once.
		next:=time.Unix(0,ts).UTC().Round(time.Second).Add(dur)
		ifnow:=time.Now().UTC().Round(time.Second);next.Before(now) {
		next=now
		}
		returnnext,true,true
		}
		return time.Time{},false,false

		}
Original file line number	Diff line number	Diff line change
Expand Up		@@ -4876,14 +4876,18 @@ func getMessageSchedule(hdr []byte) (time.Time, bool) {
		return time.Time{}, true
		}
		val := bytesToString(sliceHeader(JSSchedulePattern, hdr))
		if val == _EMPTY_ {
		schedule, _, ok := parseMsgSchedule(val, time.Now().UTC().UnixNano())
		return schedule, ok
		}

		// Fast lookup and calculation of next message schedule.
		func nextMessageSchedule(hdr []byte, ts int64) (time.Time, bool) {
		if len(hdr) == 0 {
		return time.Time{}, true
		}
		if !strings.HasPrefix(val, "@at ") {
		return time.Time{}, false
		}
		t, err := time.Parse(time.RFC3339, val[4:])
		return t, err == nil
		val := bytesToString(sliceHeader(JSSchedulePattern, hdr))
		schedule, _, ok := parseMsgSchedule(val, ts)
		return schedule, ok
		}

		// Fast lookup of the message schedule TTL from headers.
Expand Down