mirrors/dapr
0
0 Fork 0
mirror of https://github.com/dapr/dapr.git synced 2026-10-28 08:35:25 +08:00
Code Activity

version-skew: patch the release-1.18 integration suite (#10506)

Browse Source 
- binarymetadata needs the bindings_metadataprobe build tag, which is
  set only by the test framework's own daprd build. Version skew always
  supplies daprd via DAPR_INTEGRATION_DAPRD_PATH, either a release
  tarball or `make build` output (allcomponents), so the component is
  never compiled in and daprd exits 1 on component init. Deleted in both
  modes, as svidcontext already is for state_spiffeprobe.

- actors/timers/remote/{grpc,http} and actors/lock/timers/remote expect a
  non-owning sidecar to accept a timer registration. #10380 made timer
  register and unregister owner-only, returning ERR_ACTOR_TIMER_NOT_OWNED
  (HTTP 403, gRPC PermissionDenied). Ported from master's rewrite of the
  same three files.

- serviceinvocation/grpc/connpool blocks all 150 handlers until 150 are
  simultaneously in flight. #10342 stopped the pool opening extra
  connections past the app's advertised MaxConcurrentStreams, so against
  a 100-stream app that rendezvous can never complete and every call
  deadlines. Ported from master.

- workflow/continueasnew/inboxoverload waits for the workflow actor count
  to reach zero. #10372 keeps a live workflow actor resident after an
  empty-inbox ack. Ported from master, minus the
  workflow.WithSigning(false) hunk, since that harness option arrived
  after 1.18 branched.

Signed-off-by: joshvanl <me@joshvanl.dev>
Co-authored-by: Cassie Coyle <cassie@diagrid.io>
This commit is contained in:
Josh van LeeuwenandCassie Coyle authored and GitHub committed 2026-09-17 16:20:08 +00:00
1 parent 311740998a
commit 885b68f657
5 files changed
+738
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.github/scripts/version-skew-test-patches/integration/release-1.18/control-plane-master/003-remove-binarymetadata-test.patch
+102
View File
@@ -0,0 +1,102 @@
diff --git a/tests/integration/suite/daprd/binding/output/binarymetadata.go b/tests/integration/suite/daprd/binding/output/binarymetadata.go
deleted file mode 100644
index 3668f2fd..00000000
--- a/tests/integration/suite/daprd/binding/output/binarymetadata.go
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
-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 output
-
-import (
- "context"
- "encoding/json"
- "testing"
-
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
- grpcMetadata "google.golang.org/grpc/metadata"
-
- rtv1 "github.com/dapr/dapr/pkg/proto/runtime/v1"
- "github.com/dapr/dapr/tests/integration/framework"
- "github.com/dapr/dapr/tests/integration/framework/process/daprd"
- "github.com/dapr/dapr/tests/integration/suite"
-)
-
-func init() {
- suite.Register(new(binarymetadata))
-}
-
-// binarymetadata asserts that binary gRPC metadata (keys ending in "-bin") is
-// not forwarded to output binding component metadata, while text tracing and
-// custom metadata still is. The bindings.metadataprobe component echoes the
-// metadata it receives as JSON in the response data.
-type binarymetadata struct {
- daprd *daprd.Daprd
-}
-
-func (b *binarymetadata) Setup(t *testing.T) []framework.Option {
- b.daprd = daprd.New(t,
- daprd.WithResourceFiles(`apiVersion: dapr.io/v1alpha1
-kind: Component
-metadata:
- name: probe
-spec:
- type: bindings.metadataprobe
- version: v1
-`))
-
- return []framework.Option{
- framework.WithProcesses(b.daprd),
- }
-}
-
-func (b *binarymetadata) Run(t *testing.T, ctx context.Context) {
- b.daprd.WaitUntilRunning(t, ctx)
-
- client := b.daprd.GRPCClient(t, ctx)
-
- // Valid OpenCensus binary trace context, 29 bytes. The trace ID contains
- // a horizontal TAB byte (0x09), which is what breaks components that
- // sanitize or validate string metadata.
- traceBin := []byte{
- 0x00,
- 0x00, 0x4b, 0xf9, 0x2f, 0x35, 0x77, 0xb3, 0x4d, 0xa6, 0xa3, 0xce, 0x92, 0x9d, 0x0e, 0x09, 0x47, 0x36,
- 0x01, 0x00, 0xf0, 0x67, 0xaa, 0x0b, 0xa9, 0x02, 0xb7,
- 0x02, 0x01,
- }
-
- tp := "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
- ctx = grpcMetadata.AppendToOutgoingContext(ctx,
- "grpc-trace-bin", string(traceBin),
- "foo-bin", string([]byte{0x00, 0x09, 0xff}),
- "traceparent", tp,
- "custom-key", "custom-value",
- )
-
- resp, err := client.InvokeBinding(ctx, &rtv1.InvokeBindingRequest{
- Name: "probe",
- Operation: "get",
- })
- require.NoError(t, err)
-
- var got map[string]string
- require.NoError(t, json.Unmarshal(resp.GetData(), &got))
-
- assert.NotContains(t, got, "grpc-trace-bin")
- assert.NotContains(t, got, "dapr-grpc-trace-bin")
- assert.NotContains(t, got, "foo-bin")
- assert.Equal(t, tp, got["traceparent"])
- assert.Equal(t, "custom-value", got["custom-key"])
-}
.github/scripts/version-skew-test-patches/integration/release-1.18/dapr-sidecar-master/004-remove-binarymetadata-test.patch
+102
View File
@@ -0,0 +1,102 @@
diff --git a/tests/integration/suite/daprd/binding/output/binarymetadata.go b/tests/integration/suite/daprd/binding/output/binarymetadata.go
deleted file mode 100644
index 3668f2fd..00000000
--- a/tests/integration/suite/daprd/binding/output/binarymetadata.go
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
-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 output
-
-import (
- "context"
- "encoding/json"
- "testing"
-
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
- grpcMetadata "google.golang.org/grpc/metadata"
-
- rtv1 "github.com/dapr/dapr/pkg/proto/runtime/v1"
- "github.com/dapr/dapr/tests/integration/framework"
- "github.com/dapr/dapr/tests/integration/framework/process/daprd"
- "github.com/dapr/dapr/tests/integration/suite"
-)
-
-func init() {
- suite.Register(new(binarymetadata))
-}
-
-// binarymetadata asserts that binary gRPC metadata (keys ending in "-bin") is
-// not forwarded to output binding component metadata, while text tracing and
-// custom metadata still is. The bindings.metadataprobe component echoes the
-// metadata it receives as JSON in the response data.
-type binarymetadata struct {
- daprd *daprd.Daprd
-}
-
-func (b *binarymetadata) Setup(t *testing.T) []framework.Option {
- b.daprd = daprd.New(t,
- daprd.WithResourceFiles(`apiVersion: dapr.io/v1alpha1
-kind: Component
-metadata:
- name: probe
-spec:
- type: bindings.metadataprobe
- version: v1
-`))
-
- return []framework.Option{
- framework.WithProcesses(b.daprd),
- }
-}
-
-func (b *binarymetadata) Run(t *testing.T, ctx context.Context) {
- b.daprd.WaitUntilRunning(t, ctx)
-
- client := b.daprd.GRPCClient(t, ctx)
-
- // Valid OpenCensus binary trace context, 29 bytes. The trace ID contains
- // a horizontal TAB byte (0x09), which is what breaks components that
- // sanitize or validate string metadata.
- traceBin := []byte{
- 0x00,
- 0x00, 0x4b, 0xf9, 0x2f, 0x35, 0x77, 0xb3, 0x4d, 0xa6, 0xa3, 0xce, 0x92, 0x9d, 0x0e, 0x09, 0x47, 0x36,
- 0x01, 0x00, 0xf0, 0x67, 0xaa, 0x0b, 0xa9, 0x02, 0xb7,
- 0x02, 0x01,
- }
-
- tp := "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
- ctx = grpcMetadata.AppendToOutgoingContext(ctx,
- "grpc-trace-bin", string(traceBin),
- "foo-bin", string([]byte{0x00, 0x09, 0xff}),
- "traceparent", tp,
- "custom-key", "custom-value",
- )
-
- resp, err := client.InvokeBinding(ctx, &rtv1.InvokeBindingRequest{
- Name: "probe",
- Operation: "get",
- })
- require.NoError(t, err)
-
- var got map[string]string
- require.NoError(t, json.Unmarshal(resp.GetData(), &got))
-
- assert.NotContains(t, got, "grpc-trace-bin")
- assert.NotContains(t, got, "dapr-grpc-trace-bin")
- assert.NotContains(t, got, "foo-bin")
- assert.Equal(t, tp, got["traceparent"])
- assert.Equal(t, "custom-value", got["custom-key"])
-}
.github/scripts/version-skew-test-patches/integration/release-1.18/dapr-sidecar-master/005-actor-timer-owner-only.patch
+354
View File
@@ -0,0 +1,354 @@
diff --git a/tests/integration/suite/actors/lock/timers/remote.go b/tests/integration/suite/actors/lock/timers/remote.go
index fbc02daa..6f8e14d8 100644
--- a/tests/integration/suite/actors/lock/timers/remote.go
+++ b/tests/integration/suite/actors/lock/timers/remote.go
@@ -23,6 +23,8 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
rtv1 "github.com/dapr/dapr/pkg/proto/runtime/v1"
"github.com/dapr/dapr/tests/integration/framework"
@@ -95,13 +97,22 @@ func (r *remote) Run(t *testing.T, ctx context.Context) {
Name: "foo1",
DueTime: "0s",
})
+ require.Equal(t, codes.PermissionDenied, status.Code(err))
+
+ ownerClient := r.app1.GRPCClient(t, ctx)
+ _, err = ownerClient.RegisterActorTimer(ctx, &rtv1.RegisterActorTimerRequest{
+ ActorType: "abc",
+ ActorId: strconv.Itoa(int(i.Load())),
+ Name: "foo1",
+ DueTime: "0s",
+ })
require.NoError(t, err)
assert.EventuallyWithT(t, func(c *assert.CollectT) {
assert.Equal(c, int64(2), r.called.Load())
}, time.Second*10, time.Millisecond*10)
- _, err = client.RegisterActorTimer(ctx, &rtv1.RegisterActorTimerRequest{
+ _, err = ownerClient.RegisterActorTimer(ctx, &rtv1.RegisterActorTimerRequest{
ActorType: "abc",
ActorId: strconv.Itoa(int(i.Load())),
Name: "foo2",
diff --git a/tests/integration/suite/actors/timers/remote/grpc.go b/tests/integration/suite/actors/timers/remote/grpc.go
index f28edca4..3c7c013e 100644
--- a/tests/integration/suite/actors/timers/remote/grpc.go
+++ b/tests/integration/suite/actors/timers/remote/grpc.go
@@ -24,6 +24,8 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ grpccodes "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
rtv1 "github.com/dapr/dapr/pkg/proto/runtime/v1"
"github.com/dapr/dapr/tests/integration/framework"
@@ -39,25 +41,15 @@ type grpc struct {
app1 *actors.Actors
app2 *actors.Actors
- called1, called2 atomic.Int64
- timer1, timer2 atomic.Int64
+ timer1, timer2 atomic.Int64
}
func (g *grpc) Setup(t *testing.T) []framework.Option {
g.app1 = actors.New(t,
actors.WithActorTypes("abc"),
- actors.WithHandler("/actors/abc/{id}", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
- assert.Regexp(t, "/actors/abc/.+", r.URL.Path)
- assert.Equal(t, nethttp.MethodDelete, r.Method)
- }),
- actors.WithHandler("/actors/abc/{id}/method/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
- assert.Equal(t, nethttp.MethodPut, r.Method)
- assert.Regexp(t, "/actors/abc/.+/method/foo", r.URL.Path)
- g.called1.Add(1)
- }),
+ actors.WithHandler("/actors/abc/{id}", func(nethttp.ResponseWriter, *nethttp.Request) {}),
actors.WithHandler("/actors/abc/{id}/method/timer/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
assert.Equal(t, nethttp.MethodPut, r.Method)
- assert.Regexp(t, "/actors/abc/.+/method/timer/foo", r.URL.Path)
b, err := io.ReadAll(r.Body)
assert.NoError(t, err)
assert.JSONEq(t, `{"data":"aGVsbG8=","callback":"","dueTime":"0s","period":"1s"}`, string(b))
@@ -68,18 +60,9 @@ func (g *grpc) Setup(t *testing.T) []framework.Option {
g.app2 = actors.New(t,
actors.WithPeerActor(g.app1),
actors.WithActorTypes("abc"),
- actors.WithHandler("/actors/abc/{id}", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
- assert.Regexp(t, "/actors/abc/.+", r.URL.Path)
- assert.Equal(t, nethttp.MethodDelete, r.Method)
- }),
- actors.WithHandler("/actors/abc/{id}/method/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
- assert.Equal(t, nethttp.MethodPut, r.Method)
- assert.Regexp(t, "/actors/abc/.+", r.URL.Path)
- g.called2.Add(1)
- }),
+ actors.WithHandler("/actors/abc/{id}", func(nethttp.ResponseWriter, *nethttp.Request) {}),
actors.WithHandler("/actors/abc/{id}/method/timer/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
assert.Equal(t, nethttp.MethodPut, r.Method)
- assert.Regexp(t, "/actors/abc/.+/method/timer/foo", r.URL.Path)
b, err := io.ReadAll(r.Body)
assert.NoError(t, err)
assert.JSONEq(t, `{"data":"aGVsbG8=","callback":"","dueTime":"0s","period":"1s"}`, string(b))
@@ -96,32 +79,65 @@ func (g *grpc) Run(t *testing.T, ctx context.Context) {
g.app1.WaitUntilRunning(t, ctx)
g.app2.WaitUntilRunning(t, ctx)
- client := g.app1.Daprd().GRPCClient(t, ctx)
+ client1 := g.app1.Daprd().GRPCClient(t, ctx)
+ client2 := g.app2.Daprd().GRPCClient(t, ctx)
- var i atomic.Int64
- assert.EventuallyWithT(t, func(c *assert.CollectT) {
- _, err := client.InvokeActor(ctx, &rtv1.InvokeActorRequest{
+ register := func(client rtv1.DaprClient, id string) error {
+ _, err := client.RegisterActorTimer(ctx, &rtv1.RegisterActorTimerRequest{
ActorType: "abc",
- ActorId: strconv.Itoa(int(i.Add(1))),
- Method: "foo",
- })
- require.NoError(t, err)
- _, err = client.RegisterActorTimer(ctx, &rtv1.RegisterActorTimerRequest{
- ActorType: "abc",
- ActorId: strconv.Itoa(int(i.Load())),
+ ActorId: id,
Name: "foo",
DueTime: "0s",
Period: "1s",
Data: []byte("hello"),
})
- require.NoError(t, err)
+ return err
+ }
- assert.Positive(c, g.called1.Load())
- assert.Positive(c, g.called2.Load())
- }, time.Second*10, 1)
+ // Wait until both hosts agree on the placement table before requiring
+ // exactly one owner per actor.
+ assert.EventuallyWithT(t, func(c *assert.CollectT) {
+ err1 := register(client1, "probe")
+ err2 := register(client2, "probe")
+ assert.NotEqual(c, err1 == nil, err2 == nil)
+ }, time.Second*10, time.Millisecond*10)
+
+ owner1, owner2 := "", ""
+ for i := 0; owner1 == "" || owner2 == ""; i++ {
+ require.Less(t, i, 100, "actor IDs never hashed to both hosts")
+ id := strconv.Itoa(i)
+ err1 := register(client1, id)
+ err2 := register(client2, id)
+ require.False(t, err1 == nil && err2 == nil, "both hosts accepted the same timer registration")
+ require.False(t, err1 != nil && err2 != nil, "both hosts rejected the timer registration")
+ if err1 == nil {
+ require.Equal(t, grpccodes.PermissionDenied, status.Code(err2))
+ if owner1 == "" {
+ owner1 = id
+ }
+ } else {
+ require.Equal(t, grpccodes.PermissionDenied, status.Code(err1))
+ if owner2 == "" {
+ owner2 = id
+ }
+ }
+ }
assert.EventuallyWithT(t, func(c *assert.CollectT) {
- assert.GreaterOrEqual(c, g.timer1.Load(), int64(2))
- assert.GreaterOrEqual(c, g.timer2.Load(), int64(2))
+ assert.Positive(c, g.timer1.Load())
+ assert.Positive(c, g.timer2.Load())
}, time.Second*10, time.Millisecond*10)
+
+ _, err := client2.UnregisterActorTimer(ctx, &rtv1.UnregisterActorTimerRequest{
+ ActorType: "abc", ActorId: owner1, Name: "foo",
+ })
+ require.Equal(t, grpccodes.PermissionDenied, status.Code(err))
+ _, err = client1.UnregisterActorTimer(ctx, &rtv1.UnregisterActorTimerRequest{
+ ActorType: "abc", ActorId: owner1, Name: "foo",
+ })
+ require.NoError(t, err)
+ _, err = client2.UnregisterActorTimer(ctx, &rtv1.UnregisterActorTimerRequest{
+ ActorType: "abc", ActorId: owner2, Name: "foo",
+ })
+ require.NoError(t, err)
}
diff --git a/tests/integration/suite/actors/timers/remote/http.go b/tests/integration/suite/actors/timers/remote/http.go
index 117c28a6..013bc4b9 100644
--- a/tests/integration/suite/actors/timers/remote/http.go
+++ b/tests/integration/suite/actors/timers/remote/http.go
@@ -15,15 +15,18 @@ package remote
import (
"context"
+ "encoding/json"
"fmt"
"io"
nethttp "net/http"
+ "strconv"
"strings"
"sync/atomic"
"testing"
"time"
"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"
@@ -39,25 +42,15 @@ type http struct {
app1 *actors.Actors
app2 *actors.Actors
- called1, called2 atomic.Int64
- timer1, timer2 atomic.Int64
+ timer1, timer2 atomic.Int64
}
func (h *http) Setup(t *testing.T) []framework.Option {
h.app1 = actors.New(t,
actors.WithActorTypes("abc"),
- actors.WithHandler("/actors/abc/{id}", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
- assert.Regexp(t, "/actors/abc/.+", r.URL.Path)
- assert.Equal(t, nethttp.MethodDelete, r.Method)
- }),
- actors.WithHandler("/actors/abc/{id}/method/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
- assert.Equal(t, nethttp.MethodPut, r.Method)
- assert.Regexp(t, "/actors/abc/.+/method/foo", r.URL.Path)
- h.called1.Add(1)
- }),
+ actors.WithHandler("/actors/abc/{id}", func(nethttp.ResponseWriter, *nethttp.Request) {}),
actors.WithHandler("/actors/abc/{id}/method/timer/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
assert.Equal(t, nethttp.MethodPut, r.Method)
- assert.Regexp(t, "/actors/abc/.+/method/timer/foo", r.URL.Path)
b, err := io.ReadAll(r.Body)
assert.NoError(t, err)
assert.JSONEq(t, `{"data":"hello","callback":"","dueTime":"0s","period":"1s"}`, string(b))
@@ -68,18 +61,9 @@ func (h *http) Setup(t *testing.T) []framework.Option {
h.app2 = actors.New(t,
actors.WithPeerActor(h.app1),
actors.WithActorTypes("abc"),
- actors.WithHandler("/actors/abc/{id}", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
- assert.Regexp(t, "/actors/abc/.+", r.URL.Path)
- assert.Equal(t, nethttp.MethodDelete, r.Method)
- }),
- actors.WithHandler("/actors/abc/{id}/method/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
- assert.Equal(t, nethttp.MethodPut, r.Method)
- assert.Regexp(t, "/actors/abc/.+", r.URL.Path)
- h.called2.Add(1)
- }),
+ actors.WithHandler("/actors/abc/{id}", func(nethttp.ResponseWriter, *nethttp.Request) {}),
actors.WithHandler("/actors/abc/{id}/method/timer/foo", func(_ nethttp.ResponseWriter, r *nethttp.Request) {
assert.Equal(t, nethttp.MethodPut, r.Method)
- assert.Regexp(t, "/actors/abc/.+/method/timer/foo", r.URL.Path)
b, err := io.ReadAll(r.Body)
assert.NoError(t, err)
assert.JSONEq(t, `{"data":"hello","callback":"","dueTime":"0s","period":"1s"}`, string(b))
@@ -96,37 +80,74 @@ func (h *http) Run(t *testing.T, ctx context.Context) {
h.app1.WaitUntilRunning(t, ctx)
h.app2.WaitUntilRunning(t, ctx)
- client := client.HTTP(t)
+ httpClient := client.HTTP(t)
+ body := `{"dueTime":"0s","period":"1s","data":"hello"}`
+
+ do := func(t require.TestingT, method, addr, id string) int {
+ var reqBody io.Reader
+ if method == nethttp.MethodPost {
+ reqBody = strings.NewReader(body)
+ }
+ url := fmt.Sprintf("http://%s/v1.0/actors/abc/%s/timers/foo", addr, id)
+ req, err := nethttp.NewRequestWithContext(ctx, method, url, reqBody)
+ require.NoError(t, err)
+ resp, err := httpClient.Do(req)
+ require.NoError(t, err)
+ if resp.StatusCode == nethttp.StatusForbidden {
+ b, rerr := io.ReadAll(resp.Body)
+ require.NoError(t, rerr)
+ var apiErr struct {
+ ErrorCode string `json:"errorCode"`
+ }
+ require.NoError(t, json.Unmarshal(b, &apiErr))
+ assert.Equal(t, "ERR_ACTOR_TIMER_NOT_OWNED", apiErr.ErrorCode)
+ }
+ require.NoError(t, resp.Body.Close())
+ return resp.StatusCode
+ }
- body := `{
-"dueTime": "0s",
-"period": "1s",
-"data": "hello"
-}`
+ probe := func(c *assert.CollectT, addr string) int {
+ url := fmt.Sprintf("http://%s/v1.0/actors/abc/probe/timers/foo", addr)
+ req, err := nethttp.NewRequestWithContext(ctx, nethttp.MethodPost, url, strings.NewReader(body))
+ require.NoError(c, err)
+ resp, err := httpClient.Do(req)
+ require.NoError(c, err)
+ require.NoError(c, resp.Body.Close())
+ return resp.StatusCode
+ }
- var i atomic.Int64
+ // Wait until both hosts agree on the placement table before requiring
+ // exactly one owner per actor.
assert.EventuallyWithT(t, func(c *assert.CollectT) {
- url := fmt.Sprintf("http://%s/v1.0/actors/abc/%d/method/foo", h.app1.Daprd().HTTPAddress(), i.Add(1))
- req, err := nethttp.NewRequestWithContext(ctx, nethttp.MethodPost, url, nil)
- assert.NoError(t, err)
- resp, err := client.Do(req)
- assert.NoError(t, err)
- assert.Equal(t, nethttp.StatusOK, resp.StatusCode)
- assert.NoError(t, resp.Body.Close())
-
- url = fmt.Sprintf("http://%s/v1.0/actors/abc/%d/timers/foo", h.app1.Daprd().HTTPAddress(), i.Load())
- req, err = nethttp.NewRequestWithContext(ctx, nethttp.MethodPost, url, strings.NewReader(body))
- assert.NoError(t, err)
- resp, err = client.Do(req)
- assert.NoError(t, err)
- assert.Equal(t, nethttp.StatusNoContent, resp.StatusCode)
- assert.Positive(c, h.called1.Load())
- assert.Positive(c, h.called2.Load())
- assert.NoError(t, resp.Body.Close())
- }, time.Second*10, 1)
+ code1 := probe(c, h.app1.Daprd().HTTPAddress())
+ code2 := probe(c, h.app2.Daprd().HTTPAddress())
+ assert.NotEqual(c, code1 == nethttp.StatusNoContent, code2 == nethttp.StatusNoContent)
+ assert.ElementsMatch(c, []int{nethttp.StatusNoContent, nethttp.StatusForbidden}, []int{code1, code2})
+ }, time.Second*10, time.Millisecond*10)
+
+ owner1, owner2 := "", ""
+ for i := 0; owner1 == "" || owner2 == ""; i++ {
+ require.Less(t, i, 100, "actor IDs never hashed to both hosts")
+ id := strconv.Itoa(i)
+ code1 := do(t, nethttp.MethodPost, h.app1.Daprd().HTTPAddress(), id)
+ code2 := do(t, nethttp.MethodPost, h.app2.Daprd().HTTPAddress(), id)
+ codes := []int{code1, code2}
+ assert.ElementsMatch(t, []int{nethttp.StatusNoContent, nethttp.StatusForbidden}, codes)
+ if code1 == nethttp.StatusNoContent && owner1 == "" {
+ owner1 = id
+ }
+ if code2 == nethttp.StatusNoContent && owner2 == "" {
+ owner2 = id
+ }
+ }
assert.EventuallyWithT(t, func(c *assert.CollectT) {
- assert.GreaterOrEqual(c, h.timer1.Load(), int64(2))
- assert.GreaterOrEqual(c, h.timer2.Load(), int64(2))
+ assert.Positive(c, h.timer1.Load())
+ assert.Positive(c, h.timer2.Load())
}, time.Second*10, time.Millisecond*10)
+
+ assert.Equal(t, nethttp.StatusForbidden, do(t, nethttp.MethodDelete, h.app2.Daprd().HTTPAddress(), owner1))
+ assert.Equal(t, nethttp.StatusForbidden, do(t, nethttp.MethodDelete, h.app1.Daprd().HTTPAddress(), owner2))
+ assert.Equal(t, nethttp.StatusNoContent, do(t, nethttp.MethodDelete, h.app1.Daprd().HTTPAddress(), owner1))
+ assert.Equal(t, nethttp.StatusNoContent, do(t, nethttp.MethodDelete, h.app2.Daprd().HTTPAddress(), owner2))
}
.github/scripts/version-skew-test-patches/integration/release-1.18/dapr-sidecar-master/006-connpool-app-stream-limit.patch
+111
View File
@@ -0,0 +1,111 @@
diff --git a/tests/integration/suite/daprd/serviceinvocation/grpc/connpool.go b/tests/integration/suite/daprd/serviceinvocation/grpc/connpool.go
index dca27de0..3cb7ffea 100644
--- a/tests/integration/suite/daprd/serviceinvocation/grpc/connpool.go
+++ b/tests/integration/suite/daprd/serviceinvocation/grpc/connpool.go
@@ -18,8 +18,10 @@ import (
"sync"
"sync/atomic"
"testing"
+ "time"
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
"google.golang.org/grpc"
commonv1 "github.com/dapr/dapr/pkg/proto/common/v1"
@@ -35,38 +37,43 @@ func init() {
suite.Register(new(connpool))
}
-// connpool verifies that gRPC connection pooling allows more than 100
-// concurrent streams from daprd to the app via the InvokeService path.
-// The app server enforces MaxConcurrentStreams=100, so without connection
-// pooling only 100 simultaneous streams can exist on a single connection.
-// This test sends 150 concurrent InvokeService calls through daprd, each of
-// which creates a stream on a daprd->app connection. The app handler blocks
-// until all 150 are in-flight simultaneously, proving that the pool opened
-// additional connections to accommodate them.
+// connpool verifies pooled-connection semantics against a server that
+// advertises an HTTP/2 stream limit. The app enforces
+// MaxConcurrentStreams=100 while 150 concurrent InvokeService calls flow
+// through daprd. With the pool's shared-ref cap above the server limit,
+// excess calls queue at the transport behind the app's advertised limit
+// rather than fanning out onto extra connections to circumvent it: all
+// calls complete, and the app never observes more in-flight work than it
+// asked for. (The pre-raise pool opened a second connection at ref 101,
+// silently exceeding the app's declared concurrency.)
type connpool struct {
- daprd *procdaprd.Daprd
- allArrived chan struct{}
+ daprd *procdaprd.Daprd
+ maxInflight *atomic.Int32
}
-const numConcurrentPoolStreams = 150
+const (
+ numConcurrentPoolStreams = 150
+ appMaxConcurrentStreams = 100
+)
func (c *connpool) Setup(t *testing.T) []framework.Option {
- c.allArrived = make(chan struct{})
- var once sync.Once
var inflight atomic.Int32
+ var maxInflight atomic.Int32
onInvoke := func(ctx context.Context, in *commonv1.InvokeRequest) (*commonv1.InvokeResponse, error) {
n := inflight.Add(1)
defer inflight.Add(-1)
-
- if int(n) >= numConcurrentPoolStreams {
- once.Do(func() { close(c.allArrived) })
+ for {
+ seen := maxInflight.Load()
+ if n <= seen || maxInflight.CompareAndSwap(seen, n) {
+ break
+ }
}
- // Block until all concurrent requests have arrived, proving that
- // >100 streams are active simultaneously across pooled connections.
+ // Hold the stream briefly so the concurrent callers overlap and
+ // the transport-level queueing is actually exercised.
select {
- case <-c.allArrived:
+ case <-time.After(100 * time.Millisecond):
case <-ctx.Done():
return nil, ctx.Err()
}
@@ -74,13 +81,15 @@ func (c *connpool) Setup(t *testing.T) []framework.Option {
return new(commonv1.InvokeResponse), nil
}
+ c.maxInflight = &maxInflight
+
srv := app.New(t,
app.WithOnInvokeFn(onInvoke),
- // Enforce the 100-stream-per-connection limit so that the pool must
- // open a second connection to handle >100 concurrent streams.
+ // The app declares its concurrency appetite; the pool must respect
+ // it, not defeat it with extra connections.
app.WithGRPCOptions(procgrpc.WithServerOption(
func(*testing.T, context.Context) grpc.ServerOption {
- return grpc.MaxConcurrentStreams(100)
+ return grpc.MaxConcurrentStreams(appMaxConcurrentStreams)
},
)),
)
@@ -119,6 +128,12 @@ func (c *connpool) Run(t *testing.T, ctx context.Context) {
wg.Wait()
for i, err := range errs {
- assert.NoErrorf(t, err, "concurrent request %d failed", i)
+ require.NoErrorf(t, err, "concurrent request %d failed", i)
}
+
+ maxSeen := c.maxInflight.Load()
+ assert.LessOrEqual(t, maxSeen, int32(appMaxConcurrentStreams),
+ "the app's advertised stream limit must be respected, not circumvented by extra connections")
+ assert.GreaterOrEqual(t, maxSeen, int32(20),
+ "requests must genuinely overlap for this test to prove queueing")
}
.github/scripts/version-skew-test-patches/integration/release-1.18/dapr-sidecar-master/007-workflow-actor-residency.patch
+69
View File
@@ -0,0 +1,69 @@
diff --git a/tests/integration/suite/daprd/workflow/continueasnew/inboxoverload.go b/tests/integration/suite/daprd/workflow/continueasnew/inboxoverload.go
index 6ba5fd5c..2f785ea0 100644
--- a/tests/integration/suite/daprd/workflow/continueasnew/inboxoverload.go
+++ b/tests/integration/suite/daprd/workflow/continueasnew/inboxoverload.go
@@ -19,6 +19,7 @@ import (
"encoding/base64"
"fmt"
"strconv"
+ "strings"
"sync"
"sync/atomic"
"testing"
@@ -104,26 +105,33 @@ func (i *inboxoverload) Run(t *testing.T, ctx context.Context) {
actorType := "dapr.internal.default." + appID + ".workflow"
actorID := "inboxoverloadi"
- // Inject the deactivation reminder via the scheduler directly: the
- // daprd RegisterActorReminder API rejects "dapr.internal.*" actor
- // types because they are reserved for the workflow runtime.
_, err = i.workflow.Scheduler().Client(t, ctx).ScheduleJob(ctx,
i.workflow.Scheduler().JobNowActor("new-event-deactivate", "default", appID, actorType, actorID))
require.NoError(t, err)
require.EventuallyWithT(t, func(c *assert.CollectT) {
- meta := i.workflow.Dapr().GetMetadata(c, ctx)
- if !assert.NotNil(c, meta.ActorRuntime) {
- return
- }
- for _, a := range meta.ActorRuntime.ActiveActors {
- if a.Type == actorType {
- assert.Zero(c, a.Count, "workflow actor %q still has %d active instance(s)", actorType, a.Count)
- return
+ var count int
+ for _, key := range i.workflow.Scheduler().ListAllKeys(t, ctx, "dapr/jobs") {
+ if strings.Contains(key, "new-event-deactivate") {
+ count++
}
}
+ assert.Zero(c, count, "the injected empty-inbox reminder must be acked and deleted")
}, 10*time.Second, 50*time.Millisecond)
+ // The new residency contract: the live workflow remains active after
+ // the empty-inbox ack.
+ dmeta := i.workflow.Dapr().GetMetadata(t, ctx)
+ require.NotNil(t, dmeta.ActorRuntime)
+ found := false
+ for _, a := range dmeta.ActorRuntime.ActiveActors {
+ if a.Type == actorType {
+ found = true
+ require.Equal(t, 1, a.Count, "live workflow actor %q must stay resident after an empty-inbox ack", actorType)
+ }
+ }
+ require.True(t, found, "workflow actor type %q must be active; absence means it was deactivated", actorType)
+
db := i.workflow.DB().GetConnection(t)
tableName := i.workflow.DB().TableName()
writeUniquePayloadInbox(t, ctx, db, tableName, appID, actorType, actorID, totalEvents)
@@ -140,9 +148,9 @@ func (i *inboxoverload) Run(t *testing.T, ctx context.Context) {
}, time.Second*30, 10*time.Millisecond)
drainMode.Store(true)
- meta, err := client.WaitForWorkflowCompletion(ctx, id)
+ wmeta, err := client.WaitForWorkflowCompletion(ctx, id)
require.NoError(t, err)
- require.NotNil(t, meta.GetOutput())
+ require.NotNil(t, wmeta.GetOutput())
mu.Lock()
defer mu.Unlock()