[v1.15] Actors: Don't sent Reentrancy ID header when disabled (#8542)

* [v1.15] Actors: Don't sent Reentrancy ID header when disabled

Fix sending Actor Reentrancy ID HTTP header when disabled

Problem
Calling `this.StateManager.TryGetStateAsync()` in the DotNet Actor SDK would return stale data during some invocation scenarios.

Impact
The latest Actor state data was not being correctly returned during some Actor invocation scenarios using the DotNet Actor SDK.

Root cause
When Reentrancy was disabled, the Actor Reentrancy ID HTTP header was still being sent to Actor HTTP servers.
The DotNet SDK uses the existence of this HTTP header in logic to determine what state should be returned to the Actor.

Solution
Don't send the Actor Reentrancy ID HTTP header (`"Dapr-Reentrancy-Id"`) when Reentrancy is disabled.

Signed-off-by: joshvanl <[email protected]>

* Fix actor call placement lock due to reentrancy ID HTTP header.

Remove reentancy ID HTTP header check from placement lock during actor
calls. Since we always added the reentrancy ID header in v1.15.0,
`isReentrancy` bool was always true, so `!isReentrancy` was always
false- a no-op to the OR statement. This was added before other changes
were made to the actor locking system, but since it was a no-op, it was
never caught and removed. Now the reentrancy ID header is (correctly)
optionally added to keep backwards compatibility, and the previous actor
engine state this check was catering for no longer exists, this check
actually breaks the runtime.

Signed-off-by: joshvanl <[email protected]>

---------

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

docs/release_notes/v1.15.1.md

@@ -1,11 +1,15 @@
 # Dapr 1.15.1
 
-This update includes a fix for https://github.com/dapr/dapr/issues/8537.
+This update includes bug fixes:
 
-## Fixes Dapr not honoring max-body-size when invoking actors
+- [Fix Dapr not honoring max-body-size when invoking actors](#fixes-dapr-not-honoring-max-body-size-when-invoking-actors)
+- [Fix sending Actor Reentrancy ID HTTP header when disabled](#fix-sending-actor-reentrancy-id-http-header-when-disabled)
+
+## Fix Dapr not honoring max-body-size when invoking actors
 
 ### Problem
 
+Issue reported [here](https://github.com/dapr/dapr/issues/8537).
 When an actor client attempts to invoke an actor with a payload size larger than 4Mb, the call fails with `rpc error: code = ResourceExhausted desc = grpc: received message larger than max`.
 
 ### Impact
@@ -18,4 +22,23 @@ The Dapr actor gRPC client did not honor the max-body-size parameter that is pas
 
 ### Solution
 
-The DApr actor gRPC client is configured with the proper gRPC call options.
+The Dapr actor gRPC client is configured with the proper gRPC call options.
+
+## Fix sending Actor Reentrancy ID HTTP header when disabled
+
+### Problem
+
+Calling `this.StateManager.TryGetStateAsync()` in the DotNet Actor SDK would return stale data during some invocation scenarios.
+
+### Impact
+
+The latest Actor state data was not being correctly returned during some Actor invocation scenarios using the DotNet Actor SDK.
+
+### Root cause
+
+When Reentrancy was disabled, the Actor Reentrancy ID HTTP header was still being sent to Actor HTTP servers.
+The DotNet SDK uses the existence of this HTTP header in logic to determine what state should be returned to the Actor.
+
+### Solution
+
+Don't send the Actor Reentrancy ID HTTP header (`"Dapr-Reentrancy-Id"`) when Reentrancy is disabled.

pkg/actors/engine/engine.go

@@ -212,10 +212,9 @@ func (e *engine) callReminder(ctx context.Context, req *api.Reminder) error {
 func (e *engine) callActor(ctx context.Context, req *internalv1pb.InternalInvokeRequest) (*internalv1pb.InternalInvokeResponse, error) {
 	// If we are in a reentrancy which is local, skip the placement lock.
 	_, isDaprRemote := req.GetMetadata()["X-Dapr-Remote"]
-	_, isReentrancy := req.GetMetadata()["Dapr-Reentrancy-Id"]
 	_, isAPICall := req.GetMetadata()["Dapr-API-Call"]
 
-	if isAPICall || isDaprRemote || !isReentrancy {
+	if isAPICall || isDaprRemote {
 		var cancel context.CancelFunc
 		var err error
 		ctx, cancel, err = e.placement.Lock(ctx)

pkg/actors/internal/locker/lock.go

@@ -31,6 +31,8 @@ import (
 
 var ErrLockClosed = errors.New("actor lock is closed")
 
+const headerReentrancyID = "Dapr-Reentrancy-Id"
+
 type lockOptions struct {
 	actorType         string
 	reentrancyEnabled bool
@@ -229,19 +231,19 @@ func newInflight(id string) *inflight {
 }
 
 func (l *lock) idFromRequest(req *internalv1pb.InternalInvokeRequest) (string, bool) {
-	if req == nil {
+	if !l.reentrancyEnabled || req == nil {
 		return uuid.New().String(), false
 	}
 
-	if md := req.GetMetadata()["Dapr-Reentrancy-Id"]; md != nil && len(md.GetValues()) > 0 {
+	if md := req.GetMetadata()[headerReentrancyID]; md != nil && len(md.GetValues()) > 0 {
 		return md.GetValues()[0], true
 	}
 
 	id := uuid.New().String()
 	if req.Metadata == nil {
 		req.Metadata = make(map[string]*internalv1pb.ListStringValue)
 	}
-	req.Metadata["Dapr-Reentrancy-Id"] = &internalv1pb.ListStringValue{
+	req.Metadata[headerReentrancyID] = &internalv1pb.ListStringValue{
 		Values: []string{id},
 	}
 

pkg/actors/internal/locker/lock_test.go

@@ -200,3 +200,35 @@ func Test_ringid(t *testing.T) {
 		}
 	}
 }
+
+func Test_header(t *testing.T) {
+	t.Parallel()
+
+	t.Run("with header", func(t *testing.T) {
+		l := newLock(lockOptions{
+			reentrancyEnabled: true,
+			maxStackDepth:     10,
+		})
+		t.Cleanup(l.close)
+
+		req := internalv1pb.NewInternalInvokeRequest("foo")
+		cancel, err := l.lockRequest(req)
+		require.NoError(t, err)
+		t.Cleanup(cancel)
+		assert.NotEmpty(t, req.GetMetadata()["Dapr-Reentrancy-Id"])
+	})
+
+	t.Run("without header", func(t *testing.T) {
+		l := newLock(lockOptions{
+			reentrancyEnabled: false,
+			maxStackDepth:     10,
+		})
+		t.Cleanup(l.close)
+
+		req := internalv1pb.NewInternalInvokeRequest("foo")
+		cancel, err := l.lockRequest(req)
+		require.NoError(t, err)
+		t.Cleanup(cancel)
+		assert.Empty(t, req.GetMetadata()["Dapr-Reentrancy-Id"])
+	})
+}

tests/integration/suite/actors/lock/call/reentry/disabled.go

@@ -0,0 +1,82 @@
+/*
+Copyright 2024 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 implieh.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package reentry
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/dapr/dapr/tests/integration/framework"
+	"github.com/dapr/dapr/tests/integration/framework/client"
+	"github.com/dapr/dapr/tests/integration/framework/process/daprd"
+	"github.com/dapr/dapr/tests/integration/framework/process/http/app"
+	"github.com/dapr/dapr/tests/integration/framework/process/placement"
+	"github.com/dapr/dapr/tests/integration/suite"
+	"github.com/dapr/kit/ptr"
+)
+
+func init() {
+	suite.Register(new(disabled))
+}
+
+type disabled struct {
+	daprd  *daprd.Daprd
+	called atomic.Int32
+	rid    atomic.Pointer[string]
+}
+
+func (d *disabled) Setup(t *testing.T) []framework.Option {
+	app := app.New(t,
+		app.WithConfig(`{"entities":["reentrantActor"],"reentrancy":{"enabled":false}}`),
+		app.WithHandlerFunc("/actors/reentrantActor/myactorid", func(_ http.ResponseWriter, r *http.Request) {}),
+		app.WithHandlerFunc("/actors/reentrantActor/myactorid/method/foo", func(_ http.ResponseWriter, r *http.Request) {
+			d.rid.Store(ptr.Of(r.Header.Get("Dapr-Reentrancy-Id")))
+			d.called.Add(1)
+		}),
+	)
+
+	place := placement.New(t)
+	d.daprd = daprd.New(t,
+		daprd.WithInMemoryActorStateStore("mystore"),
+		daprd.WithPlacementAddresses(place.Address()),
+		daprd.WithAppPort(app.Port()),
+	)
+
+	return []framework.Option{
+		framework.WithProcesses(app, place, d.daprd),
+	}
+}
+
+func (d *disabled) Run(t *testing.T, ctx context.Context) {
+	d.daprd.WaitUntilRunning(t, ctx)
+
+	client := client.HTTP(t)
+
+	url := fmt.Sprintf("http://%s/v1.0/actors/reentrantActor/myactorid/method/foo", d.daprd.HTTPAddress())
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil)
+	require.NoError(t, err)
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+	require.NoError(t, resp.Body.Close())
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+	assert.Equal(t, int32(1), d.called.Load())
+	assert.Empty(t, *d.rid.Load())
+}

tests/integration/suite/actors/lock/call/reentry/pertype/disabled.go

@@ -0,0 +1,82 @@
+/*
+Copyright 2024 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 implieh.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package pertype
+
+import (
+	"context"
+	"fmt"
+	nethttp "net/http"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/dapr/dapr/tests/integration/framework"
+	"github.com/dapr/dapr/tests/integration/framework/client"
+	"github.com/dapr/dapr/tests/integration/framework/process/daprd"
+	"github.com/dapr/dapr/tests/integration/framework/process/http/app"
+	"github.com/dapr/dapr/tests/integration/framework/process/placement"
+	"github.com/dapr/dapr/tests/integration/suite"
+	"github.com/dapr/kit/ptr"
+)
+
+func init() {
+	suite.Register(new(disabled))
+}
+
+type disabled struct {
+	daprd  *daprd.Daprd
+	called atomic.Int32
+	rid    atomic.Pointer[string]
+}
+
+func (d *disabled) Setup(t *testing.T) []framework.Option {
+	app := app.New(t,
+		app.WithConfig(`{"entities":["reentrantActor"],"reentrancy":{"enabled":true},"entitiesConfig":[{"entities":["reentrantActor"],"reentrancy":{"enabled":false}}]}`),
+		app.WithHandlerFunc("/actors/reentrantActor/myactorid", func(_ nethttp.ResponseWriter, r *nethttp.Request) {}),
+		app.WithHandlerFunc("/actors/reentrantActor/myactorid/method/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
+			d.rid.Store(ptr.Of(r.Header.Get("Dapr-Reentrancy-Id")))
+			d.called.Add(1)
+		}),
+	)
+
+	place := placement.New(t)
+	d.daprd = daprd.New(t,
+		daprd.WithInMemoryActorStateStore("mystore"),
+		daprd.WithPlacementAddresses(place.Address()),
+		daprd.WithAppPort(app.Port()),
+	)
+
+	return []framework.Option{
+		framework.WithProcesses(app, place, d.daprd),
+	}
+}
+
+func (d *disabled) Run(t *testing.T, ctx context.Context) {
+	d.daprd.WaitUntilRunning(t, ctx)
+
+	client := client.HTTP(t)
+
+	url := fmt.Sprintf("http://%s/v1.0/actors/reentrantActor/myactorid/method/foo", d.daprd.HTTPAddress())
+	req, err := nethttp.NewRequestWithContext(ctx, nethttp.MethodPost, url, nil)
+	require.NoError(t, err)
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+	require.NoError(t, resp.Body.Close())
+	assert.Equal(t, nethttp.StatusOK, resp.StatusCode)
+
+	assert.Equal(t, int32(1), d.called.Load())
+	assert.Empty(t, *d.rid.Load())
+}