events/queue: read ScheduledTime under the lock and match it in execute

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,21 +185,20 @@ 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
 		}
 
 		t = p.clock.NewTimer(deadline)
 		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
 	}

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")
+	}
+}

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
 }