hanzo/pubsub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(2.14) [FIXED] Don't allow purging schedule if not a schedule

Browse Source 
Signed-off-by: Maurice van Veen <github@mauricevanveen.com>
This commit is contained in:
Maurice van Veen
2026-05-12 11:21:12 +02:00
parent a68ed818e1
commit b0b210ae34
4 changed files with 189 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/jetstream_batching.go
+25 -1
View File
@@ -451,6 +451,7 @@ func (diff *batchStagedDiff) commit(mset *stream) {
if c, ok := mset.inflight[subj]; ok {
c.bytes += i.bytes
c.ops += i.ops
c.schedule = i.schedule
} else {
mset.inflight[subj] = i
}
@@ -530,6 +531,7 @@ func checkMsgHeadersPreClusteredProposal(
discard DiscardPolicy, discardNewPer bool, maxMsgSize int, maxMsgs int64, maxMsgsPer int64, maxBytes int64,
) ([]byte, []byte, uint64, *ApiError, error) {
var incr *big.Int
var hasSchedule bool
// Some header checks must be checked pre proposal.
if len(hdr) > 0 {
@@ -810,6 +812,7 @@ func checkMsgHeadersPreClusteredProposal(
}
return hdr, msg, 0, apiErr, apiErr
} else if !schedule.IsZero() {
hasSchedule = true
if !allowMsgSchedules {
apiErr := NewJSMessageSchedulesDisabledError()
return hdr, msg, 0, apiErr, apiErr
@@ -877,6 +880,26 @@ func checkMsgHeadersPreClusteredProposal(
} else if !allowMsgSchedules {
apiErr := NewJSMessageSchedulesDisabledError()
return hdr, msg, 0, apiErr, apiErr
} else {
// Check that the to-be-purged subject is a schedule message.
// We still allow this message through if there exists no message for this subject,
// to remain backward-compatible. An "expected at sequence" check can still be
// performed to make this stricter.
schedSubj := bytesToString(scheduler)
var invalid bool
if i, ok := diff.inflight[schedSubj]; ok {
invalid = !i.schedule
} else if i, ok = mset.inflight[schedSubj]; ok {
invalid = !i.schedule
} else {
var smv StoreMsg
sm, _ := mset.store.LoadLastMsg(schedSubj, &smv)
invalid = sm != nil && len(sliceHeader(JSSchedulePattern, sm.hdr)) == 0
}
if invalid {
apiErr := NewJSMessageSchedulesSchedulerInvalidError()
return hdr, msg, 0, apiErr, apiErr
}
}
} else if !sourced && len(sliceHeader(JSScheduler, hdr)) > 0 {
// Clients may only use Nats-Scheduler alongside Nats-Schedule-Next.
@@ -930,8 +953,9 @@ func checkMsgHeadersPreClusteredProposal(
if i, ok = diff.inflight[subject]; ok {
i.bytes += sz
i.ops++
i.schedule = hasSchedule
} else {
i = &inflightSubjectRunningTotal{bytes: sz, ops: 1}
i = &inflightSubjectRunningTotal{bytes: sz, ops: 1, schedule: hasSchedule}
diff.inflight[subject] = i
}
server/jetstream_batching_test.go
+106
View File
@@ -1172,6 +1172,112 @@ func TestJetStreamAtomicBatchPublishStageAndCommit(t *testing.T) {
},
},
},
{
title: "msg-schedules-purge-staged-not-schedule",
allowMsgSchedules: true,
allowRollup: true,
batch: []BatchItem{
{subject: "foo"},
{
subject: "bar",
header: nats.Header{
JSScheduleNext: {JSScheduleNextPurge},
JSScheduler: {"foo"},
},
err: NewJSMessageSchedulesSchedulerInvalidError(),
},
},
},
{
title: "msg-schedules-purge-staged-schedule",
allowMsgSchedules: true,
allowRollup: true,
batch: []BatchItem{
{
subject: "foo",
header: nats.Header{
JSSchedulePattern: {"@at 1970-01-01T00:00:00Z"},
JSScheduleTarget: {"baz"},
},
},
{
subject: "bar",
header: nats.Header{
JSScheduleNext: {JSScheduleNextPurge},
JSScheduler: {"foo"},
},
},
},
validate: func(mset *stream, commit bool) {
if !commit {
return
}
require_NotNil(t, mset.inflight["foo"])
require_True(t, mset.inflight["foo"].schedule)
},
},
{
title: "msg-schedules-purge-staged-overwritten",
allowMsgSchedules: true,
allowRollup: true,
batch: []BatchItem{
{
subject: "foo",
header: nats.Header{
JSSchedulePattern: {"@at 1970-01-01T00:00:00Z"},
JSScheduleTarget: {"baz"},
},
},
{subject: "foo"},
{
subject: "bar",
header: nats.Header{
JSScheduleNext: {JSScheduleNextPurge},
JSScheduler: {"foo"},
},
err: NewJSMessageSchedulesSchedulerInvalidError(),
},
},
},
{
title: "msg-schedules-purge-inflight-not-schedule",
allowMsgSchedules: true,
allowRollup: true,
init: func(mset *stream) {
mset.inflight = map[string]*inflightSubjectRunningTotal{
"foo": {bytes: 10, ops: 1, schedule: false},
}
},
batch: []BatchItem{
{
subject: "bar",
header: nats.Header{
JSScheduleNext: {JSScheduleNextPurge},
JSScheduler: {"foo"},
},
err: NewJSMessageSchedulesSchedulerInvalidError(),
},
},
},
{
title: "msg-schedules-purge-inflight-schedule",
allowMsgSchedules: true,
allowRollup: true,
init: func(mset *stream) {
mset.inflight = map[string]*inflightSubjectRunningTotal{
"foo": {bytes: 10, ops: 1, schedule: true},
}
},
batch: []BatchItem{
{
subject: "bar",
header: nats.Header{
JSScheduleNext: {JSScheduleNextPurge},
JSScheduler: {"foo"},
},
},
},
},
{
title: "discard-new",
discardNew: true,
server/jetstream_cluster_1_test.go
+38 -3
View File
@@ -9966,10 +9966,19 @@ func TestJetStreamClusterPurgeScheduleWithInvalidScheduler(t *testing.T) {
_, err := jsStreamCreate(t, nc, cfg)
require_NoError(t, err)
_, err = js.Publish("foo.schedule", []byte("previous"))
// Publish a normal non-schedule message.
_, err = js.Publish("foo.normal", []byte("previous"))
require_NoError(t, err)
// Publish a schedule message.
m := nats.NewMsg("foo.schedule")
m.Header.Set("Nats-Schedule", "@every 1h")
m.Header.Set("Nats-Schedule-Target", "foo.target")
m.Data = []byte("previous")
_, err = js.PublishMsg(m)
require_NoError(t, err)
m = nats.NewMsg("foo.schedule")
m.Header.Set("Nats-Schedule-Next", "purge")
// Missing scheduler
_, err = js.PublishMsg(m)
@@ -9985,22 +9994,48 @@ func TestJetStreamClusterPurgeScheduleWithInvalidScheduler(t *testing.T) {
_, err = js.PublishMsg(m)
require_Error(t, err, NewJSMessageSchedulesSchedulerInvalidError())
// We're allowed to purge a schedule, as long as the subject we're publishing to doesn't equal.
// Should error if the scheduler message is not actually a schedule.
// Otherwise, non-schedule messages could be purged this way.
m.Header.Set("Nats-Scheduler", "foo.normal")
m.Subject = "foo.last"
_, err = js.PublishMsg(m)
require_Error(t, err, NewJSMessageSchedulesSchedulerInvalidError())
// We're allowed to purge a schedule, as long as the subject we're
// publishing to doesn't equal and it is a schedule.
m.Header.Set("Nats-Scheduler", "foo.schedule")
m.Subject = "foo.last"
_, err = js.PublishMsg(m)
require_NoError(t, err)
// Purge the normal message separately.
require_NoError(t, js.PurgeStream("SchedulesEnabled", &nats.StreamPurgeRequest{Subject: "foo.normal"}))
// Only the last message should exist.
checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
state, err := checkStateAndErr(t, c, globalAccountName, "SchedulesEnabled")
if err != nil {
return err
}
if state.Msgs != 1 || state.FirstSeq != 2 || state.LastSeq != 2 {
if state.Msgs != 1 || state.FirstSeq != 3 || state.LastSeq != 3 {
return fmt.Errorf("expected 1 msg, got %v", state)
}
return nil
})
// If the scheduler message does not exist, we should preserve backward-compatibility
// and allow the message to be persisted with a noop purge.
m.Header.Set("Nats-Scheduler", "foo.none")
m.Subject = "foo.last"
_, err = js.PublishMsg(m)
require_NoError(t, err)
// Should still be able to perform an "expected at sequence" check to make sure the
// schedule actually exists before persisting the message and performing a purge.
m.Header.Set("Nats-Expected-Last-Subject-Sequence", "1")
m.Header.Set("Nats-Expected-Last-Subject-Sequence-Subject", "foo.none")
_, err = js.PublishMsg(m)
require_Error(t, err, NewJSStreamWrongLastSequenceError(0))
})
}
}
server/stream.go
+20 -2
View File
@@ -582,8 +582,9 @@ type stream struct {
// inflightSubjectRunningTotal stores a running total of inflight messages for a specific subject.
type inflightSubjectRunningTotal struct {
bytes uint64 // Running total of inflight bytes for inflight messages.
ops uint64 // Inflight operations, i.e. inflight messages for this subject. If this reaches zero, we can remove the running total.
bytes uint64 // Running total of inflight bytes for inflight messages.
ops uint64 // Inflight operations, i.e. inflight messages for this subject. If this reaches zero, we can remove the running total.
schedule bool // Marks whether the last message is a schedule.
}
// msgCounterRunningTotal stores a running total and a number of inflight
@@ -6540,6 +6541,23 @@ func (mset *stream) processJetStreamMsgWithBatch(subject, reply string, hdr, msg
outq.sendMsg(reply, b)
}
return apiErr
} else {
// Check that the to-be-purged subject is a schedule message.
// We still allow this message through if there exists no message for this subject,
// to remain backward-compatible. An "expected at sequence" check can still be
// performed to make this stricter.
var smv StoreMsg
sm, _ := store.LoadLastMsg(bytesToString(scheduler), &smv)
if sm != nil && len(sliceHeader(JSSchedulePattern, sm.hdr)) == 0 {
apiErr := NewJSMessageSchedulesSchedulerInvalidError()
if canRespond {
resp.PubAck = &PubAck{Stream: name}
resp.Error = apiErr
b, _ := json.Marshal(resp)
outq.sendMsg(reply, b)
}
return apiErr
}
}
} else if !sourced && len(sliceHeader(JSScheduler, hdr)) > 0 {
// Clients may only use Nats-Scheduler alongside Nats-Schedule-Next.