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 }