Misc refactorings extracted from dapr#5170 (dapr#5609)

.gitignore

@@ -15,6 +15,7 @@ github.com/
 google
 
 test_report*
+coverage.txt
 
 # Go Workspaces (introduced in Go 1.18+)
 go.work

pkg/actors/actors.go

@@ -329,7 +329,10 @@ type lookupActorRes struct {
 }
 
 func (a *actorsRuntime) Call(ctx context.Context, req *invokev1.InvokeMethodRequest) (*invokev1.InvokeMethodResponse, error) {
-	a.placement.WaitUntilPlacementTableIsReady()
+	err := a.placement.WaitUntilPlacementTableIsReady(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to wait for placement table readiness: %w", err)
+	}
 
 	actor := req.Actor()
 	// Retry here to allow placement table dissemination/rebalancing to happen.
@@ -1031,13 +1034,14 @@ func (a *actorsRuntime) executeReminder(reminder *Reminder) error {
 }
 
 func (a *actorsRuntime) reminderRequiresUpdate(req *CreateReminderRequest, reminder *Reminder) bool {
-	if reminder.ActorID == req.ActorID && reminder.ActorType == req.ActorType && reminder.Name == req.Name &&
-		(!reflect.DeepEqual(reminder.Data, req.Data) || reminder.DueTime != req.DueTime || reminder.Period != req.Period ||
-			len(req.TTL) != 0 || (len(reminder.ExpirationTime) != 0 && len(req.TTL) == 0)) {
-		return true
-	}
-
-	return false
+	return reminder.ActorID == req.ActorID &&
+		reminder.ActorType == req.ActorType &&
+		reminder.Name == req.Name &&
+		(!reflect.DeepEqual(reminder.Data, req.Data) ||
+			reminder.DueTime != req.DueTime ||
+			reminder.Period != req.Period ||
+			len(req.TTL) != 0 ||
+			(len(reminder.ExpirationTime) != 0 && len(req.TTL) == 0))
 }
 
 func (a *actorsRuntime) getReminder(reminderName string, actorType string, actorID string) (*Reminder, bool) {

pkg/actors/internal/placement.go

@@ -14,6 +14,7 @@ limitations under the License.
 package internal
 
 import (
+	"context"
 	"net"
 	"sync"
 	"sync/atomic"
@@ -224,9 +225,15 @@ func (p *ActorPlacement) Stop() {
 }
 
 // WaitUntilPlacementTableIsReady waits until placement table is until table lock is unlocked.
-func (p *ActorPlacement) WaitUntilPlacementTableIsReady() {
-	if p.tableIsBlocked.Load() {
-		<-p.unblockSignal
+func (p *ActorPlacement) WaitUntilPlacementTableIsReady(ctx context.Context) error {
+	if !p.tableIsBlocked.Load() {
+		return nil
+	}
+	select {
+	case <-p.unblockSignal:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
 	}
 }
 

pkg/actors/internal/placement_test.go

@@ -14,6 +14,7 @@ limitations under the License.
 package internal
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net"
@@ -24,6 +25,7 @@ import (
 
 	"github.com/phayes/freeport"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -201,24 +203,69 @@ func TestWaitUntilPlacementTableIsReady(t *testing.T) {
 		[]string{"actorOne", "actorTwo"},
 		appHealthFunc, tableUpdateFunc)
 
-	testPlacement.onPlacementOrder(&placementv1pb.PlacementOrder{Operation: "lock"})
+	t.Run("already unlocked", func(t *testing.T) {
+		require.False(t, testPlacement.tableIsBlocked.Load())
 
-	asserted := atomic.Bool{}
-	asserted.Store(false)
-	go func() {
-		testPlacement.WaitUntilPlacementTableIsReady()
-		asserted.Store(true)
-	}()
+		err := testPlacement.WaitUntilPlacementTableIsReady(context.Background())
+		assert.NoError(t, err)
+	})
+
+	t.Run("wait until ready", func(t *testing.T) {
+		testPlacement.onPlacementOrder(&placementv1pb.PlacementOrder{Operation: "lock"})
+
+		testSuccessCh := make(chan struct{})
+		go func() {
+			err := testPlacement.WaitUntilPlacementTableIsReady(context.Background())
+			if assert.NoError(t, err) {
+				testSuccessCh <- struct{}{}
+			}
+		}()
+
+		time.Sleep(50 * time.Millisecond)
+		require.True(t, testPlacement.tableIsBlocked.Load())
+
+		// unlock
+		testPlacement.onPlacementOrder(&placementv1pb.PlacementOrder{Operation: "unlock"})
+
+		// ensure that it is unlocked
+		select {
+		case <-time.After(500 * time.Millisecond):
+			t.Fatal("placement table not unlocked in 500ms")
+		case <-testSuccessCh:
+			// all good
+		}
 
-	time.Sleep(50 * time.Millisecond)
-	assert.False(t, asserted.Load())
+		assert.False(t, testPlacement.tableIsBlocked.Load())
+	})
 
-	// unlock
-	testPlacement.onPlacementOrder(&placementv1pb.PlacementOrder{Operation: "unlock"})
+	t.Run("abort on context canceled", func(t *testing.T) {
+		testPlacement.onPlacementOrder(&placementv1pb.PlacementOrder{Operation: "lock"})
+
+		testSuccessCh := make(chan struct{})
+		ctx, cancel := context.WithCancel(context.Background())
+		go func() {
+			err := testPlacement.WaitUntilPlacementTableIsReady(ctx)
+			if assert.ErrorIs(t, err, context.Canceled) {
+				testSuccessCh <- struct{}{}
+			}
+		}()
+
+		time.Sleep(50 * time.Millisecond)
+		require.True(t, testPlacement.tableIsBlocked.Load())
+
+		// cancel context
+		cancel()
+
+		// ensure that it is still locked
+		select {
+		case <-time.After(500 * time.Millisecond):
+			t.Fatal("did not return in 500ms")
+		case <-testSuccessCh:
+			// all good
+		}
 
-	// ensure that it is unlocked
-	time.Sleep(50 * time.Millisecond)
-	assert.True(t, asserted.Load())
+		assert.True(t, testPlacement.tableIsBlocked.Load())
+	})
 }
 
 func TestLookupActor(t *testing.T) {

pkg/channel/grpc/grpc_channel.go

@@ -114,14 +114,12 @@ func (g *Channel) invokeMethodV1(ctx context.Context, req *invokev1.InvokeMethod
 
 	var header, trailer grpcMetadata.MD
 
-	var opts []grpc.CallOption
-	opts = append(
-		opts,
+	opts := []grpc.CallOption{
 		grpc.Header(&header),
 		grpc.Trailer(&trailer),
-		grpc.MaxCallSendMsgSize(g.maxRequestBodySizeMB<<20),
-		grpc.MaxCallRecvMsgSize(g.maxRequestBodySizeMB<<20),
-	)
+		grpc.MaxCallSendMsgSize(g.maxRequestBodySizeMB << 20),
+		grpc.MaxCallRecvMsgSize(g.maxRequestBodySizeMB << 20),
+	}
 
 	resp, err := g.appCallbackClient.OnInvoke(ctx, req.Message(), opts...)
 
@@ -139,9 +137,11 @@ func (g *Channel) invokeMethodV1(ctx context.Context, req *invokev1.InvokeMethod
 		rsp = invokev1.NewInvokeMethodResponse(int32(codes.OK), "", nil)
 	}
 
-	rsp.WithHeaders(header).WithTrailers(trailer)
+	rsp.WithHeaders(header).
+		WithTrailers(trailer).
+		WithMessage(resp)
 
-	return rsp.WithMessage(resp), nil
+	return rsp, nil
 }
 
 // HealthProbe performs a health probe.

pkg/channel/http/http_channel_test.go

@@ -21,11 +21,11 @@ import (
 	"net/http/httptest"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/atomic"
 
 	"github.com/dapr/dapr/pkg/config"
 	invokev1 "github.com/dapr/dapr/pkg/messaging/v1"
@@ -41,13 +41,13 @@ type testConcurrencyHandler struct {
 }
 
 func (t *testConcurrencyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	cur := t.currentCalls.Inc()
+	cur := t.currentCalls.Add(1)
 
 	if cur > t.maxCalls {
 		t.testFailed = true
 	}
 
-	t.currentCalls.Dec()
+	t.currentCalls.Add(-1)
 	io.WriteString(w, r.URL.RawQuery)
 }
 
@@ -365,7 +365,7 @@ func TestInvokeMethodMaxConcurrency(t *testing.T) {
 	t.Run("single concurrency", func(t *testing.T) {
 		handler := testConcurrencyHandler{
 			maxCalls:     1,
-			currentCalls: atomic.NewInt32(0),
+			currentCalls: &atomic.Int32{},
 		}
 		server := httptest.NewServer(&handler)
 		c := Channel{
@@ -398,7 +398,7 @@ func TestInvokeMethodMaxConcurrency(t *testing.T) {
 	t.Run("10 concurrent calls", func(t *testing.T) {
 		handler := testConcurrencyHandler{
 			maxCalls:     10,
-			currentCalls: atomic.NewInt32(0),
+			currentCalls: &atomic.Int32{},
 		}
 		server := httptest.NewServer(&handler)
 		c := Channel{
@@ -431,7 +431,7 @@ func TestInvokeMethodMaxConcurrency(t *testing.T) {
 	t.Run("introduce failures", func(t *testing.T) {
 		handler := testConcurrencyHandler{
 			maxCalls:     5,
-			currentCalls: atomic.NewInt32(0),
+			currentCalls: &atomic.Int32{},
 		}
 		server := httptest.NewServer(&handler)
 		c := Channel{

pkg/components/bindings/input_pluggable_test.go

@@ -21,25 +21,20 @@ import (
 	"os"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"testing"
 
+	guuid "github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
 	"github.com/dapr/components-contrib/bindings"
 	contribMetadata "github.com/dapr/components-contrib/metadata"
-
 	"github.com/dapr/dapr/pkg/components/pluggable"
 	proto "github.com/dapr/dapr/pkg/proto/components/v1"
 	testingGrpc "github.com/dapr/dapr/pkg/testing/grpc"
-
-	guuid "github.com/google/uuid"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"go.uber.org/atomic"
-
 	"github.com/dapr/kit/logger"
-
-	"google.golang.org/grpc"
 )
 
 type inputBindingServer struct {