Multiple fixes in actors package

pkg/actors/actor.go

@@ -22,6 +22,7 @@ import (
 	"k8s.io/utils/clock"
 
 	diag "github.com/dapr/dapr/pkg/diagnostics"
+	"github.com/dapr/kit/ptr"
 )
 
 // ErrActorDisposed is the error when runtime tries to hold the lock of the disposed actor.
@@ -39,11 +40,12 @@ type actor struct {
 	// pendingActorCalls is the number of the current pending actor calls by turn-based concurrency.
 	pendingActorCalls atomic.Int32
 
-	// When consistent hashing tables are updated, actor runtime drains actor to rebalance actors
-	// across actor hosts after drainOngoingCallTimeout or until all pending actor calls are completed.
-	// lastUsedTime is the time when the last actor call holds lock. This is used to calculate
-	// the duration of ongoing calls to time out.
-	lastUsedTime time.Time
+	// idleTimeout is the configured max idle time for actors of this kind.
+	idleTimeout time.Duration
+
+	// idleAt is the time after which this actor is considered to be idle.
+	// When the actor is locked, idleAt is updated by adding the idleTimeout to the current time.
+	idleAt atomic.Pointer[time.Time]
 
 	// disposeLock guards disposed and disposeCh.
 	disposeLock sync.RWMutex
@@ -56,17 +58,33 @@ type actor struct {
 	clock clock.Clock
 }
 
-func newActor(actorType, actorID string, maxReentrancyDepth *int, cl clock.Clock) *actor {
+func newActor(actorType, actorID string, maxReentrancyDepth *int, idleTimeout time.Duration, cl clock.Clock) *actor {
 	if cl == nil {
 		cl = &clock.RealClock{}
 	}
-	return &actor{
-		actorType:    actorType,
-		actorID:      actorID,
-		actorLock:    NewActorLock(int32(*maxReentrancyDepth)),
-		clock:        cl,
-		lastUsedTime: cl.Now().UTC(),
+
+	a := &actor{
+		actorType:   actorType,
+		actorID:     actorID,
+		actorLock:   NewActorLock(int32(*maxReentrancyDepth)),
+		clock:       cl,
+		idleTimeout: idleTimeout,
 	}
+	a.idleAt.Store(ptr.Of(cl.Now().Add(idleTimeout)))
+
+	return a
+}
+
+// Key returns the key for this unique actor.
+// This is implemented to comply with the queueable interface.
+func (a *actor) Key() string {
+	return a.actorType + daprSeparator + a.actorID
+}
+
+// ScheduledTime returns the time the actor becomes idle at.
+// This is implemented to comply with the queueable interface.
+func (a *actor) ScheduledTime() time.Time {
+	return *a.idleAt.Load()
 }
 
 // isBusy returns true when pending actor calls are ongoing.
@@ -115,7 +133,8 @@ func (a *actor) lock(reentrancyID *string) error {
 		a.unlock()
 		return ErrActorDisposed
 	}
-	a.lastUsedTime = a.clock.Now().UTC()
+
+	a.idleAt.Store(ptr.Of(a.clock.Now().Add(a.idleTimeout)))
 	return nil
 }
 

pkg/actors/actor_test.go

@@ -26,20 +26,20 @@ import (
 var reentrancyStackDepth = 32
 
 func TestIsBusy(t *testing.T) {
-	testActor := newActor("testType", "testID", &reentrancyStackDepth, nil)
+	testActor := newActor("testType", "testID", &reentrancyStackDepth, time.Second, nil)
 
 	testActor.lock(nil)
 	assert.Equal(t, true, testActor.isBusy())
 	testActor.unlock()
 }
 
 func TestTurnBasedConcurrencyLocks(t *testing.T) {
-	testActor := newActor("testType", "testID", &reentrancyStackDepth, nil)
+	testActor := newActor("testType", "testID", &reentrancyStackDepth, time.Second, nil)
 
 	// first lock
 	testActor.lock(nil)
 	assert.Equal(t, true, testActor.isBusy())
-	firstLockTime := testActor.lastUsedTime
+	firstIdleAt := *testActor.idleAt.Load()
 
 	waitCh := make(chan bool)
 
@@ -57,7 +57,7 @@ func TestTurnBasedConcurrencyLocks(t *testing.T) {
 	time.Sleep(10 * time.Millisecond)
 	assert.Equal(t, int32(2), testActor.pendingActorCalls.Load())
 	assert.True(t, testActor.isBusy())
-	assert.Equal(t, firstLockTime, testActor.lastUsedTime)
+	assert.Equal(t, firstIdleAt, *testActor.idleAt.Load())
 
 	// unlock the first lock
 	testActor.unlock()
@@ -70,12 +70,12 @@ func TestTurnBasedConcurrencyLocks(t *testing.T) {
 
 	assert.Equal(t, int32(0), testActor.pendingActorCalls.Load())
 	assert.False(t, testActor.isBusy())
-	assert.True(t, testActor.lastUsedTime.Sub(firstLockTime) >= 10*time.Millisecond)
+	assert.True(t, testActor.idleAt.Load().Sub(firstIdleAt) >= 10*time.Millisecond)
 }
 
 func TestDisposedActor(t *testing.T) {
 	t.Run("not disposed", func(t *testing.T) {
-		testActor := newActor("testType", "testID", &reentrancyStackDepth, nil)
+		testActor := newActor("testType", "testID", &reentrancyStackDepth, time.Second, nil)
 
 		testActor.lock(nil)
 		testActor.unlock()
@@ -86,7 +86,7 @@ func TestDisposedActor(t *testing.T) {
 	})
 
 	t.Run("disposed", func(t *testing.T) {
-		testActor := newActor("testType", "testID", &reentrancyStackDepth, nil)
+		testActor := newActor("testType", "testID", &reentrancyStackDepth, time.Second, nil)
 
 		testActor.lock(nil)
 		ch := testActor.channel()
@@ -102,7 +102,7 @@ func TestDisposedActor(t *testing.T) {
 
 func TestPendingActorCalls(t *testing.T) {
 	t.Run("no pending actor call with new actor object", func(t *testing.T) {
-		testActor := newActor("testType", "testID", &reentrancyStackDepth, nil)
+		testActor := newActor("testType", "testID", &reentrancyStackDepth, time.Second, nil)
 		channelClosed := false
 
 		select {
@@ -117,7 +117,7 @@ func TestPendingActorCalls(t *testing.T) {
 	})
 
 	t.Run("close channel before timeout", func(t *testing.T) {
-		testActor := newActor("testType", "testID", &reentrancyStackDepth, nil)
+		testActor := newActor("testType", "testID", &reentrancyStackDepth, time.Second, nil)
 		testActor.lock(nil)
 
 		channelClosed := atomic.Bool{}
@@ -139,7 +139,7 @@ func TestPendingActorCalls(t *testing.T) {
 
 	t.Run("multiple listeners", func(t *testing.T) {
 		clock := clocktesting.NewFakeClock(time.Now())
-		testActor := newActor("testType", "testID", &reentrancyStackDepth, clock)
+		testActor := newActor("testType", "testID", &reentrancyStackDepth, time.Second, clock)
 		testActor.lock(nil)
 
 		nListeners := 10