From b5deeb8f03ee0c503dbf8709d61aedc573d28325 Mon Sep 17 00:00:00 2001 From: joshvanl Date: Fri, 16 Oct 2026 16:44:03 +0100 Subject: [PATCH] events/queue: read ScheduledTime under the lock and match it in execute processLoop peeked the front item under the lock but called ScheduledTime after unlocking, so a concurrent Dequeue could hand the item back to its owner first. go-cron reset and reused such items, and the read dereferenced the reset state (nil pointer SIGSEGV in the dapr scheduler under load). The heap already calls ScheduledTime under this lock on every Enqueue, so no new lock order. execute also matched the front item by pointer only. An owner that dequeued an item and enqueued the same pointer with a later time could have it fire at the old time. execute now also requires the scheduled time read by processLoop, and restarts the loop if it changed. Signed-off-by: joshvanl --- events/queue/processor.go | 18 +-- events/queue/processor_peeklock_test.go | 139 ++++++++++++++++++++++++ events/queue/queue.go | 3 + 3 files changed, 153 insertions(+), 7 deletions(-) create mode 100644 events/queue/processor_peeklock_test.go diff --git a/events/queue/processor.go b/events/queue/processor.go index 84561a8..4dd8f72 100644 --- a/events/queue/processor.go +++ b/events/queue/processor.go @@ -165,6 +165,10 @@ func (p *Processor[K, T]) processLoop() { return } + // Read the item while it is still guaranteed to be queued: once the + // lock is released, a Dequeue can remove it and its owner may change + // or reuse it before this loop runs again. + scheduledTime = r.ScheduledTime() p.lock.Unlock() // Check if after obtaining the lock we have a stop or reset signals @@ -181,13 +185,12 @@ func (p *Processor[K, T]) processLoop() { // Nop, proceed } - scheduledTime = r.ScheduledTime() deadline = scheduledTime.Sub(p.clock.Now()) // If the deadline is less than 0.5ms away, execute it right away // This is more efficient than creating a timer if deadline < 500*time.Microsecond { - p.execute(r) + p.execute(r, scheduledTime) continue } @@ -195,7 +198,7 @@ func (p *Processor[K, T]) processLoop() { select { // Wait for when it's time to execute the item case <-t.C(): - p.execute(r) + p.execute(r, scheduledTime) // If we get a reset signal, restart the loop case <-p.resetCh: @@ -217,14 +220,15 @@ func (p *Processor[K, T]) processLoop() { } // Executes a item when it's time. -func (p *Processor[K, T]) execute(r T) { +func (p *Processor[K, T]) execute(r T, scheduledTime time.Time) { // Pop the item now that we're ready to process it // There's a small chance this is a different item than the one we peeked before p.lock.Lock() - // For safety, let's peek at the first item before popping it and make sure it's the same object - // It's unlikely, but if it's a different object then restart the loop + // The front item must be the same object with the same scheduled time: + // an owner may Dequeue an item and Enqueue the same pointer with a later + // time, which must not fire at the old time. Otherwise restart the loop. peek, ok := p.queue.Peek() - if !ok || peek != r { + if !ok || peek != r || !peek.ScheduledTime().Equal(scheduledTime) { p.lock.Unlock() return } diff --git a/events/queue/processor_peeklock_test.go b/events/queue/processor_peeklock_test.go new file mode 100644 index 0000000..e19b6b3 --- /dev/null +++ b/events/queue/processor_peeklock_test.go @@ -0,0 +1,139 @@ +/* +Copyright 2026 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package queue + +import ( + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + clocktesting "k8s.io/utils/clock/testing" +) + +// lockCheckItem counts ScheduledTime calls made while nobody holds the +// processor lock: those are the calls that can race a Dequeue. +type lockCheckItem struct { + name string + at time.Time + p *Processor[string, *lockCheckItem] + unlocked *atomic.Int64 +} + +func (i *lockCheckItem) Key() string { return i.name } + +func (i *lockCheckItem) ScheduledTime() time.Time { + if i.p.lock.TryLock() { + i.p.lock.Unlock() + i.unlocked.Add(1) + } + + return i.at +} + +// The processor must read an item's ScheduledTime under its lock. Reading it +// after unlocking let a concurrent Dequeue hand the item back to its owner, +// which reset it, and the read dereferenced the reset state (a nil pointer +// SIGSEGV in the dapr scheduler under load). +func TestProcessor_ScheduledTimeReadUnderLock(t *testing.T) { + const n = 200 + + var ( + unlocked atomic.Int64 + executed = make(chan struct{}, n) + ) + + p := NewProcessor[string, *lockCheckItem](Options[string, *lockCheckItem]{ + ExecuteFn: func(*lockCheckItem) { executed <- struct{}{} }, + }) + + t.Cleanup(func() { require.NoError(t, p.Close()) }) + + now := time.Now() + for i := range n { + p.Enqueue(&lockCheckItem{ + name: strconv.Itoa(i), + at: now.Add(time.Duration(i%5) * time.Millisecond), + p: p, + unlocked: &unlocked, + }) + } + + for range n { + select { + case <-executed: + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for items to execute") + } + } + + assert.Zero(t, unlocked.Load(), "ScheduledTime was called without the processor lock held") +} + +// hookClock runs onNow once, inside the first Now call. processLoop calls Now +// after releasing the lock and before execute, so the hook lands in that gap. +type hookClock struct { + *clocktesting.FakeClock + + once sync.Once + onNow func() +} + +func (c *hookClock) Now() time.Time { + c.once.Do(c.onNow) + return c.FakeClock.Now() +} + +// An owner may Dequeue an item and Enqueue the same pointer with a later time. +// execute matched the front item by pointer only, so it fired the re-enqueued +// item at the time processLoop had read before the Dequeue. +func TestProcessor_ReenqueuedPointerDoesNotFireEarly(t *testing.T) { + start := time.Now() + fake := clocktesting.NewFakeClock(start) + executed := make(chan *lockCheckItem, 1) + + clk := &hookClock{FakeClock: fake} + p := NewProcessor[string, *lockCheckItem](Options[string, *lockCheckItem]{ + ExecuteFn: func(r *lockCheckItem) { executed <- r }, + Clock: clk, + }) + + t.Cleanup(func() { require.NoError(t, p.Close()) }) + + item := &lockCheckItem{name: "a", at: start, p: p, unlocked: new(atomic.Int64)} + clk.onNow = func() { + p.Dequeue(item.name) + item.at = start.Add(time.Hour) + p.Enqueue(item) + } + + p.Enqueue(item) + + require.Eventually(t, func() bool { + return fake.HasWaiters() || len(executed) > 0 + }, 5*time.Second, time.Millisecond) + require.Empty(t, executed, "re-enqueued item fired at its old scheduled time") + + fake.Step(time.Hour) + + select { + case r := <-executed: + assert.Same(t, item, r) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the re-enqueued item to execute") + } +} diff --git a/events/queue/queue.go b/events/queue/queue.go index 32983bb..38e49ec 100644 --- a/events/queue/queue.go +++ b/events/queue/queue.go @@ -22,6 +22,9 @@ import ( type Queueable[T comparable] interface { comparable Key() T + // ScheduledTime is only called while the item is queued, under the + // processor's lock, so an owner may change or reuse an item once Dequeue + // for it has returned. ScheduledTime() time.Time }