ci: speed up e2e (#10435)

* ci: speed up e2e

Signed-off-by: joshvanl <me@joshvanl.dev>

* tests: address review feedback on e2e speedup

- Note in dapr_tests.mk that DAPR_E2E_SERIAL_PACKAGES entries must be
  plain directory names, since they are joined into a grep -E
  alternation.
- Name the quiet window after the single-fire reminder check
  (singleFireQuietWindow) instead of an inline 5s sleep.

Signed-off-by: joshvanl <me@joshvanl.dev>

* ci: harden the e2e image cache against stale and silent failures

Address three review findings on the cache mechanism:

- Cache keys now include shared build inputs. Apps without their own
  Dockerfile are packaged with the shared tests/apps/Dockerfile, and if
  they also have no go.mod of their own they compile against the root
  module; neither input was part of the content hash, so changing them
  would silently reuse stale cached images. Both are now hashed for
  exactly the apps they affect: a root go.sum change re-keys hellodapr
  but not actorjava (own Dockerfile) or resiliencyapp (own go.mod).
- The cache-population workflow sets DAPR_CACHE_REQUIRED, which turns a
  failed cache push into a run failure instead of a silently incomplete
  cache. All other runs keep the best-effort behavior since they may
  legitimately lack write access.
- The precompiled build tool binary is only used when the caller opts
  in with USE_BUILD_TOOLS_BIN=true, which CI sets right after
  compile-build-tools. Locally `go run` remains the default so source
  edits are always reflected instead of silently running a stale
  binary.

Signed-off-by: joshvanl <me@joshvanl.dev>

* tests: make k6 unit test fake tolerant of extra job polls

Signed-off-by: joshvanl <me@joshvanl.dev>

---------

Signed-off-by: joshvanl <me@joshvanl.dev>
Co-authored-by: Cassie Coyle <cassie@diagrid.io>

.build-tools/cmd/zz-e2e-perf.go

@@ -210,22 +210,29 @@ func (c *cmdE2EPerf) buildAndPushCmd(cmd *cobra.Command, args []string) error {
 	// Try to copy the image between the cache and the target directly, without pulling first
 	// This will fail if the image is not in cache, and that's ok
 	if c.flags.CacheRegistry != "" {
-		fmt.Printf("Trying to copy image %s directly from cache %s\n", destImage, cachedImage)
-		for attempt := 1; attempt <= 3; attempt++ {
-			err = crane.Copy(cachedImage, destImage)
-			if err == nil {
-				// If there's no error, we're done
-				fmt.Println("Image copied from cache directly! You're all set.")
-				return nil
+		// Probe the cache once before retrying the copy: a missing or unreadable
+		// image fails identically on every attempt, so retries only add delay
+		_, err = crane.Digest(cachedImage)
+		if err != nil {
+			fmt.Printf("Image %s not found in cache (%v). Will build image.\n", cachedImage, err)
+		} else {
+			fmt.Printf("Trying to copy image %s directly from cache %s\n", destImage, cachedImage)
+			for attempt := 1; attempt <= 3; attempt++ {
+				err = crane.Copy(cachedImage, destImage)
+				if err == nil {
+					// If there's no error, we're done
+					fmt.Println("Image copied from cache directly! You're all set.")
+					return nil
+				}
+				fmt.Printf("crane.Copy attempt %d/3 failed: %v\n", attempt, err)
+				if attempt < 3 {
+					time.Sleep(time.Duration(attempt*2) * time.Second)
+				}
 			}
-			fmt.Printf("crane.Copy attempt %d/3 failed: %v\n", attempt, err)
-			if attempt < 3 {
-				time.Sleep(time.Duration(attempt*2) * time.Second)
-			}
-		}
 
-		// Copying the image failed, so we'll resort to build + push
-		fmt.Printf("Copying image directly from cache failed after 3 attempts: %v. Will build image.\n", err)
+			// Copying the image failed, so we'll resort to build + push
+			fmt.Printf("Copying image directly from cache failed after 3 attempts: %v. Will build image.\n", err)
+		}
 	} else {
 		fmt.Println("Cache registry not set: will not use cache")
 	}
@@ -356,7 +363,9 @@ func (c *cmdE2EPerf) buildDockerImage(cachedImage string) error {
 	}
 
 	// Push to the cache, if needed
-	if c.flags.CacheRegistry != "" {
+	// DAPR_CACHE_READONLY skips the push on runs that have no write access to
+	// the cache registry, such as pull requests
+	if c.flags.CacheRegistry != "" && os.Getenv("DAPR_CACHE_READONLY") != "true" {
 		fmt.Printf("Pushing image %s to cache…\n", cachedImage)
 
 		err = exec.Command("docker", "tag", destImage, cachedImage).Run()
@@ -369,8 +378,15 @@ func (c *cmdE2EPerf) buildDockerImage(cachedImage string) error {
 		e.Stdout = os.Stdout
 		e.Stderr = os.Stderr
 		err = e.Run()
-		// If there's an error, we probably didn't have permissions to push to the registry, so we can just ignore that
 		if err != nil {
+			// The dedicated cache-population workflow sets DAPR_CACHE_REQUIRED
+			// so a failed push fails the run instead of leaving the cache
+			// silently incomplete. Everywhere else the push is best-effort:
+			// the run probably lacks permissions on the registry.
+			if os.Getenv("DAPR_CACHE_REQUIRED") == "true" {
+				fmt.Println("Failed to push to the cache registry:", err)
+				return err
+			}
 			fmt.Println("Failed to push to the cache registry; ignored")
 		}
 	}
@@ -543,6 +559,33 @@ func (c *cmdE2EPerf) getHashDir() (string, error) {
 		}
 	}
 
+	// Also hash shared build inputs that live outside the app's folder, so
+	// that changing them invalidates the cached image. Apps without their own
+	// Dockerfile are compiled on the host and packaged with the shared
+	// Dockerfile, so that file is an input; if such an app also has no go.mod
+	// of its own it is built against the root module, so the root go.mod and
+	// go.sum are inputs too. Apps with their own Dockerfile handle all their
+	// build inputs inside their folder, which is already hashed.
+	shared := []string{}
+	if _, err := os.Stat(filepath.Join(basePath, c.flags.Dockerfile)); err != nil {
+		if c.cmdType == "perf" {
+			shared = append(shared, filepath.Join(c.getAppDir(), "..", "Dockerfile"))
+		} else {
+			shared = append(shared, filepath.Join(c.getAppDir(), c.flags.Dockerfile))
+		}
+		if _, err := os.Stat(filepath.Join(basePath, "go.mod")); err != nil {
+			rootDir := filepath.Join(c.flags.AppDir, "..", "..")
+			shared = append(shared, filepath.Join(rootDir, "go.mod"), filepath.Join(rootDir, "go.sum"))
+		}
+	}
+	for _, f := range shared {
+		checksum, err := hashEntryForFile(f, filepath.Join("_shared", filepath.Base(f)))
+		if err != nil {
+			continue
+		}
+		files = append(files, checksum)
+	}
+
 	// Sort files to have a consistent order, then compute the checksum of that slice (getting the first 10 chars only)
 	sort.Strings(files)
 	fileList := strings.Join(files, "\n")

.github/workflows/dapr-test.yml

@@ -608,7 +608,7 @@ jobs:
         with:
           #TODO: .json suffix can be removed from artifact name after test analytics scripts are updated
           name: ${{ matrix.target_os }}_${{ matrix.target_arch }}_e2e.json
-          path: ${{ env.TEST_OUTPUT_FILE_PREFIX }}_e2e.*
+          path: ${{ env.TEST_OUTPUT_FILE_PREFIX }}_e2e*
       - name: Update PR comment for success
         if: success() && env.PR_NUMBER != '' && env.SKIP_E2E != 'true'
         uses: artursouza/sticky-pull-request-comment@v2.2.0

.github/workflows/e2e-cache-populate.yaml

@@ -0,0 +1,81 @@
+#
+# 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.
+#
+
+# Populates the public GHCR cache of e2e test app images that kind-e2e.yaml
+# and version-skew.yaml read anonymously. Cache tags are content addressed
+# (hash of the app sources), so runs are incremental: apps whose sources are
+# unchanged are cache hits and are not rebuilt.
+#
+# Note: the ghcr.io/dapr/e2e-cache/* packages must be marked public (one-time,
+# in the org package settings) so that pull requests, including from forks,
+# can read them without credentials.
+name: e2e-cache-populate
+
+on:
+  push:
+    branches:
+    - master
+    - release-*
+    paths:
+    - "tests/apps/**"
+    - "tests/dapr_tests.mk"
+    - ".build-tools/**"
+    - "Makefile"
+    - ".github/workflows/e2e-cache-populate.yaml"
+  schedule:
+  - cron: "0 4 * * 1"
+  workflow_dispatch: {}
+
+permissions: {}
+
+jobs:
+  populate:
+    name: populate
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+    env:
+      REGISTRY_PORT: 5000
+      REGISTRY_NAME: local-registry
+      DAPR_REGISTRY: localhost:5000/dapr
+      DAPR_TAG: dev
+      DAPR_CACHE_REGISTRY: "ghcr.io/dapr/e2e-cache"
+      # A failed cache push fails this run: this workflow exists to publish
+      # the cache, so finishing green without publishing would be misleading.
+      DAPR_CACHE_REQUIRED: "true"
+    steps:
+    - name: Check out code
+      uses: actions/checkout@v6
+    - name: Set up Go
+      uses: actions/setup-go@v6
+      with:
+        go-version-file: 'go.mod'
+    - name: Login to GitHub Container Registry
+      run: |
+        echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
+    - name: Setup local registry
+      # The build tool pushes each app to a destination registry as well as to
+      # the cache; a throwaway local registry keeps that fast.
+      run: |
+        docker run -d --restart=always -p ${{ env.REGISTRY_PORT }}:5000 --name ${{ env.REGISTRY_NAME }} registry:2
+    - name: Free disk space
+      run: |
+        sudo rm -rf /usr/local/lib/android || true
+        sudo rm -rf /opt/ghc || true
+        sudo rm -rf /opt/hostedtoolcache/CodeQL || true
+        df -h
+    - name: Build and push test apps
+      run: |
+        make compile-build-tools
+        make -j 6 USE_BUILD_TOOLS_BIN=true build-push-e2e-app-all

.github/workflows/kind-e2e.yaml

@@ -59,8 +59,12 @@ jobs:
       # TODO: Make this auto-populated based on GitHub's releases.
       DAPR_TEST_N_MINUS_1_IMAGE: "ghcr.io/dapr/daprd:1.16.8"
       DAPR_TEST_N_MINUS_2_IMAGE: "ghcr.io/dapr/daprd:1.15.13"
-      # Container registry where to cache e2e test images
-      DAPR_CACHE_REGISTRY: "dapre2eacr.azurecr.io"
+      # Container registry where to cache e2e test images. The packages under
+      # this path are public, so pull requests (including from forks) can read
+      # the cache anonymously. Only the e2e-cache-populate workflow writes to
+      # it; this job runs with DAPR_CACHE_READONLY=true.
+      DAPR_CACHE_REGISTRY: "ghcr.io/dapr/e2e-cache"
+      DAPR_CACHE_READONLY: "true"
       PULL_POLICY: IfNotPresent
     strategy:
       fail-fast: false # Keep running if one leg fails.
@@ -189,9 +193,21 @@ jobs:
         make docker-build
         make docker-push-retry
 
-    - name: Build and push test apps
+    - name: Precompile e2e test binaries in the background
+      # Warms the Go build cache for the Run tests step while the app images
+      # and the cluster are being prepared. -exec=true in the target means the
+      # binaries are built but nothing executes.
       run: |
-        make build-push-e2e-app-all
+        nohup make build-e2e-tests > /tmp/build-e2e-tests.log 2>&1 &
+        echo $! > /tmp/build-e2e-tests.pid
+
+    - name: Build and push test apps
+      # Compile the build tool once instead of `go run` per app, and build the
+      # apps in parallel. On a warm cache each app is a direct registry-to-registry
+      # copy from the cache into the local KinD registry.
+      run: |
+        make compile-build-tools
+        make -j 6 USE_BUILD_TOOLS_BIN=true build-push-e2e-app-all
 
     - name: Setup Dapr
       run: |
@@ -227,6 +243,11 @@ jobs:
       env:
         DAPR_TEST_CONFIG_STORE: ${{ matrix.dapr-test-config-store }}
 
+    - name: Wait for e2e test binary precompile
+      run: |
+        while kill -0 "$(cat /tmp/build-e2e-tests.pid)" 2>/dev/null; do sleep 1; done
+        tail -5 /tmp/build-e2e-tests.log || true
+
     - name: Run tests
       run: |
         make test-e2e-all
@@ -268,4 +289,5 @@ jobs:
       with:
         #TODO: .json suffix can be removed from artifact name after test analytics scripts are updated
         name: ${{ matrix.k8s-version }}_${{ matrix.mode }}_test_e2e.json
-        path: ${{ env.TEST_OUTPUT_FILE_PREFIX }}_e2e.*
+        # Also matches the _e2e_serial.json/.xml reports from the serial pass
+        path: ${{ env.TEST_OUTPUT_FILE_PREFIX }}_e2e*

.github/workflows/version-skew.yaml

@@ -258,7 +258,10 @@ jobs:
       DAPR_REGISTRY: localhost:5000/dapr
       DAPR_TAG: dev
       DAPR_NAMESPACE: dapr-tests
-      DAPR_CACHE_REGISTRY: "dapre2eacr.azurecr.io"
+      # Public-read cache, see kind-e2e.yaml. Reads are anonymous and writes
+      # are skipped via DAPR_CACHE_READONLY.
+      DAPR_CACHE_REGISTRY: "ghcr.io/dapr/e2e-cache"
+      DAPR_CACHE_READONLY: "true"
       PULL_POLICY: Always
     strategy:
       fail-fast: false # Keep running if one leg fails.
@@ -428,7 +431,7 @@ jobs:
 
     - name: Build and push test apps
       run: |
-        cd latest-release && make build-push-e2e-app-all
+        cd latest-release && make compile-build-tools && make -j 6 USE_BUILD_TOOLS_BIN=true build-push-e2e-app-all
 
     - name: Setup Dapr - control-plane master
       if: matrix.mode == 'control-plane-master'
@@ -508,7 +511,7 @@ jobs:
       uses: actions/upload-artifact@v4
       with:
         name: ${{ matrix.mode }}_test_e2e.json
-        path: ${{ env.TEST_OUTPUT_FILE_PREFIX }}_e2e.*
+        path: ${{ env.TEST_OUTPUT_FILE_PREFIX }}_e2e*
     - name: Update PR comment for success
       if: ${{ success() }}
       uses: artursouza/sticky-pull-request-comment@v2.2.0

Makefile

@@ -105,7 +105,11 @@ ifeq ($(TARGET_OS_LOCAL),windows)
 else
 	BUILD_TOOLS_BIN ?= build-tools
 	BUILD_TOOLS ?= ./.build-tools/$(BUILD_TOOLS_BIN)
-	RUN_BUILD_TOOLS ?= cd .build-tools; GOOS=$(TARGET_OS_LOCAL) GOARCH=$(TARGET_ARCH_LOCAL) go run .
+	# Use the binary produced by compile-build-tools only when explicitly
+	# requested (CI sets USE_BUILD_TOOLS_BIN=true right after compiling);
+	# `go run` always reflects source changes so it stays the default for
+	# local use, at the cost of re-linking the tool per invocation
+	RUN_BUILD_TOOLS ?= cd .build-tools; GOOS=$(TARGET_OS_LOCAL) GOARCH=$(TARGET_ARCH_LOCAL) $(if $(filter true,$(USE_BUILD_TOOLS_BIN)),./$(BUILD_TOOLS_BIN),go run .)
 endif
 
 # Default docker container and e2e test target.

tests/apps/service_invocation_grpc_proxy_client/app.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"os"
 	"strings"
 	"time"
 
@@ -62,7 +63,13 @@ func run(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
 	defer cancel()
 
-	ctx = metadata.AppendToOutgoingContext(ctx, "dapr-app-id", "grpcproxyserver")
+	// The target app id is overridable so that test suites can deploy this app
+	// and the proxy server under names that do not collide with other suites
+	serverAppID := os.Getenv("PROXY_SERVER_APP_ID")
+	if serverAppID == "" {
+		serverAppID = "grpcproxyserver"
+	}
+	ctx = metadata.AppendToOutgoingContext(ctx, "dapr-app-id", serverAppID)
 
 	var name string
 	if request.Method == "maxsize" {

tests/config/kubernetes_allowlists_grpc_config.yaml

@@ -20,7 +20,7 @@ spec:
             action: deny
           - name: grpcToGrpcWithoutVerbTest
             action: allow
-      - appId: "grpcproxyclient"
+      - appId: "allowlists-grpcproxyclient"
         defaultAction: deny
         trustDomain: 'public'
         namespace: "dapr-tests"

tests/dapr_tests.mk

@@ -343,17 +343,40 @@ push-kind-perf-app-all: $(PUSH_KIND_PERF_APPS_TARGETS)
 .PHONY: test-deps
 test-deps:
 	# The desire here is to download this test dependency without polluting go.mod
-	command -v gotestsum || go install gotest.tools/gotestsum@latest
+	command -v gotestsum || go install gotest.tools/gotestsum@v1.13.0
+
+# Packages that must not run concurrently with the rest of the e2e suite:
+# - hotreloading creates and updates resources in the shared test namespace
+#   whose hot reload restarts every daprd sidecar in that namespace
+# - scheduler deletes a scheduler control plane pod, which actor reminders,
+#   jobs and workflows in other packages depend on
+# - job consumes pubsub topics shared with the pubsub packages and asserts on
+#   exact delivery counts
+# Names must be plain directory names: they are joined into a grep -E
+# alternation in test-e2e-all, so regex metacharacters would mis-match.
+DAPR_E2E_SERIAL_PACKAGES ?= hotreloading scheduler job
+
+# Compile the e2e test binaries without running them, so that a later
+# test-e2e-all gets a warm build cache. -exec=true builds and links each test
+# binary but replaces its execution with /bin/true, so no TestMain runs and
+# nothing is deployed.
+.PHONY: build-e2e-tests
+build-e2e-tests:
+	GOOS=$(TARGET_OS_LOCAL) go test -tags=e2e -exec=true -count=1 -p 4 ./tests/e2e/...
+
+E2E_TEST_ENV_VARS := DAPR_CONTAINER_LOG_PATH=$(DAPR_CONTAINER_LOG_PATH) DAPR_TEST_LOG_PATH=$(DAPR_TEST_LOG_PATH) GOOS=$(TARGET_OS_LOCAL) DAPR_TEST_NAMESPACE=$(DAPR_TEST_NAMESPACE) DAPR_TEST_TAG=$(DAPR_TEST_TAG) DAPR_TEST_REGISTRY=$(DAPR_TEST_REGISTRY) DAPR_TEST_MINIKUBE_IP=$(MINIKUBE_NODE_IP)
 
 # start all e2e tests
 test-e2e-all: check-e2e-env test-deps
-	# Note: we can set -p 2 to run two tests apps at a time, because today we do not share state between
-	# tests. In the future, if we add any tests that modify global state (such as dapr config), we'll
-	# have to be sure and run them after the main test suite, so as not to alter the state of a running
-	# test
-	# Note2: use env variable DAPR_E2E_TEST to pick one e2e test to run.
+	# The packages in DAPR_E2E_SERIAL_PACKAGES share cluster-wide state with the
+	# rest of the suite, so they run in a second, fully serial pass. Everything
+	# else runs at -p 3. Both passes always run; the target fails if either did.
+	# Note: use env variable DAPR_E2E_TEST to pick one e2e test to run.
      ifeq ($(DAPR_E2E_TEST),)
-	DAPR_CONTAINER_LOG_PATH=$(DAPR_CONTAINER_LOG_PATH) DAPR_TEST_LOG_PATH=$(DAPR_TEST_LOG_PATH) GOOS=$(TARGET_OS_LOCAL) DAPR_TEST_NAMESPACE=$(DAPR_TEST_NAMESPACE) DAPR_TEST_TAG=$(DAPR_TEST_TAG) DAPR_TEST_REGISTRY=$(DAPR_TEST_REGISTRY) DAPR_TEST_MINIKUBE_IP=$(MINIKUBE_NODE_IP) gotestsum --jsonfile $(TEST_OUTPUT_FILE_PREFIX)_e2e.json --junitfile $(TEST_OUTPUT_FILE_PREFIX)_e2e.xml --format standard-quiet -- -timeout 20m -p 2 -count=1 -v -tags=e2e ./tests/e2e/$(DAPR_E2E_TEST)/...
+	ret=0; \
+	$(E2E_TEST_ENV_VARS) gotestsum --jsonfile $(TEST_OUTPUT_FILE_PREFIX)_e2e.json --junitfile $(TEST_OUTPUT_FILE_PREFIX)_e2e.xml --format standard-quiet -- -timeout 20m -p 3 -count=1 -v -tags=e2e $$(go list -tags=e2e ./tests/e2e/... | grep -vE "/tests/e2e/($$(echo $(DAPR_E2E_SERIAL_PACKAGES) | tr ' ' '|'))$$") || ret=$$?; \
+	$(E2E_TEST_ENV_VARS) gotestsum --jsonfile $(TEST_OUTPUT_FILE_PREFIX)_e2e_serial.json --junitfile $(TEST_OUTPUT_FILE_PREFIX)_e2e_serial.xml --format standard-quiet -- -timeout 20m -p 1 -count=1 -v -tags=e2e $(addprefix ./tests/e2e/,$(DAPR_E2E_SERIAL_PACKAGES)) || ret=$$?; \
+	exit $$ret
      else
 	for app in $(DAPR_E2E_TEST); do \
 		DAPR_CONTAINER_LOG_PATH=$(DAPR_CONTAINER_LOG_PATH) DAPR_TEST_LOG_PATH=$(DAPR_TEST_LOG_PATH) GOOS=$(TARGET_OS_LOCAL) DAPR_TEST_NAMESPACE=$(DAPR_TEST_NAMESPACE) DAPR_TEST_TAG=$(DAPR_TEST_TAG) DAPR_TEST_REGISTRY=$(DAPR_TEST_REGISTRY) DAPR_TEST_MINIKUBE_IP=$(MINIKUBE_NODE_IP) gotestsum --jsonfile $(TEST_OUTPUT_FILE_PREFIX)_e2e.json --junitfile $(TEST_OUTPUT_FILE_PREFIX)_e2e.xml --format standard-quiet -- -timeout 20m -p 2 -count=1 -v -tags=e2e ./tests/e2e/$$app/...; \

tests/e2e/actor_activation/actor_activation_test.go

@@ -20,6 +20,7 @@ package activation
 import (
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"os"
 	"testing"
 	"time"
@@ -116,8 +117,12 @@ func TestActorActivation(t *testing.T) {
 	_, err := utils.HTTPGetNTimes(externalURL, numHealthChecks)
 	require.NoError(t, err)
 
-	// Wait until runtime finds the leader of placements.
-	time.Sleep(15 * time.Second)
+	// Wait until the actor runtime is ready (placement leader found) by
+	// polling an actor invocation with a throwaway actor ID
+	require.Eventually(t, func() bool {
+		_, code, errp := utils.HTTPPostWithStatus(fmt.Sprintf(actorInvokeURLFormat, externalURL, "warmup-probe"), []byte{})
+		return errp == nil && code == http.StatusOK
+	}, 60*time.Second, time.Second, "actor invocation was not ready in time")
 
 	t.Run("Actor deactivates due to timeout.", func(t *testing.T) {
 		actorID := "100"

tests/e2e/actor_features/actor_features_test.go

@@ -42,7 +42,7 @@ const (
 	timerName                             = "myTimer"                            // Timer name.
 	numHealthChecks                       = 60                                   // Number of get calls before starting tests.
 	secondsToCheckTimerAndReminderResult  = 20                                   // How much time to wait to make sure the result is in logs.
-	secondsToCheckGetMetadata             = 40                                   // How much time to wait to check metadata.
+	singleFireQuietWindow                 = 5 * time.Second                      // Quiet hold after a single-fire reminder so an erroneous second fire is caught.
 	secondsBetweenChecksForActorFailover  = 5                                    // How much time to wait to make sure the result is in logs.
 	minimumCallsForTimerAndReminderResult = 10                                   // How many calls to timer or reminder should be at minimum.
 	actorsToCheckRebalance                = 30                                   // How many actors to create in the rebalance check test.
@@ -54,7 +54,6 @@ const (
 	actorlogsURLFormat                    = "%s/test/logs"                       // URL to fetch logs from test app.
 	actorMetadataURLFormat                = "%s/test/metadata"                   // URL to fetch metadata from test app.
 	shutdownURLFormat                     = "%s/test/shutdown"                   // URL to shutdown sidecar and app.
-	actorInvokeRetriesAfterRestart        = 10                                   // Number of retried to invoke actor after restart.
 )
 
 // represents a response for the APIs in this app.
@@ -307,7 +306,15 @@ func TestActorFeatures(t *testing.T) {
 		}
 		require.NoError(t, err, "failed to set reminder")
 
-		time.Sleep(secondsToCheckTimerAndReminderResult * time.Second)
+		// The assertion is a lower bound, so poll the logs until the reminder
+		// has fired enough times instead of waiting a fixed window
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			logs, errl := httpGet(logsURL)
+			if !assert.NoError(c, errl, "failed to get logs") {
+				return
+			}
+			assert.GreaterOrEqual(c, countActorAction(logs, actorID, reminderName), minimumCallsForTimerAndReminderResult)
+		}, (secondsToCheckTimerAndReminderResult+10)*time.Second, time.Second, "reminder did not fire enough times")
 
 		// Reset reminder
 		res, err = httpDelete(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorID, "reminders", reminderName))
@@ -315,15 +322,6 @@ func TestActorFeatures(t *testing.T) {
 			log.Printf("failed to reset reminder. Error='%v' Response='%s'", err, string(res))
 		}
 		require.NoError(t, err, "failed to reset reminder")
-
-		res, err = httpGet(logsURL)
-		if err != nil {
-			log.Printf("failed to get logs. Error='%v' Response='%s'", err, string(res))
-		}
-		require.NoError(t, err, "failed to get logs")
-		count := countActorAction(res, actorID, reminderName)
-		require.True(t, count >= 1, "condition failed: %d not >= 1. Response='%s'", count, string(res))
-		require.True(t, count >= minimumCallsForTimerAndReminderResult, "condition failed: %d not >= %d. Response='%s'", count, minimumCallsForTimerAndReminderResult, string(res))
 	})
 
 	t.Run("Actor single fire reminder.", func(t *testing.T) {
@@ -349,7 +347,16 @@ func TestActorFeatures(t *testing.T) {
 		}
 		require.NoError(t, err, "failed to set reminder")
 
-		time.Sleep(secondsToCheckTimerAndReminderResult * time.Second)
+		// Wait for the single fire, then hold a short window so an erroneous
+		// second fire would be caught by the exact-count assertion below
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			logs, errl := httpGet(logsURL)
+			if !assert.NoError(c, errl, "failed to get logs") {
+				return
+			}
+			assert.GreaterOrEqual(c, countActorAction(logs, actorID, reminderName), 1)
+		}, secondsToCheckTimerAndReminderResult*time.Second, time.Second, "reminder did not fire")
+		time.Sleep(singleFireQuietWindow)
 
 		// Reset reminder
 		res, err = httpDelete(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorID, "reminders", reminderName))
@@ -524,21 +531,15 @@ func TestActorFeatures(t *testing.T) {
 
 		logsURL = fmt.Sprintf(actorlogsURLFormat, externalURL)
 
-		err = backoff.Retry(func() error {
-			time.Sleep(30 * time.Second)
+		// The app's log is reset by the restart, so this converges once the
+		// reminder has resumed and fired enough times on the new pod
+		require.Eventually(t, func() bool {
 			resp, errb := httpGet(logsURL)
 			if errb != nil {
-				return errb
+				return false
 			}
-
-			count := countActorAction(resp, actorID, reminderName)
-			if count < minimumCallsForTimerAndReminderResult {
-				return fmt.Errorf("Not enough reminder calls: %d vs %d", count, minimumCallsForTimerAndReminderResult)
-			}
-
-			return nil
-		}, backoff.WithMaxRetries(backoff.NewExponentialBackOff(), 10))
-		require.NoError(t, err)
+			return countActorAction(resp, actorID, reminderName) >= minimumCallsForTimerAndReminderResult
+		}, 300*time.Second, 2*time.Second, "not enough reminder calls after app restart")
 
 		// Reset reminder
 		res, err = httpDelete(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorID, "reminders", reminderName))
@@ -574,15 +575,17 @@ func TestActorFeatures(t *testing.T) {
 		}
 		require.NoError(t, err, "failed to set reminder")
 
-		time.Sleep(secondsToCheckTimerAndReminderResult * time.Second)
-
-		// get reminder
-		res, err = httpGet(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorID, "reminders", reminderName))
-		if err != nil {
-			log.Printf("failed to get reminder. Error='%v' Response='%s'", err, string(res))
-		}
-		require.NoError(t, err, "failed to get reminder")
-		require.Empty(t, res, "Reminder %s exist", reminderName)
+		// The app deletes its own reminder when it first fires; poll until the
+		// reminder is gone, then hold briefly so an erroneous extra fire would
+		// be caught by the count assertion below
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			reminder, errg := httpGet(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorID, "reminders", reminderName))
+			if !assert.NoError(c, errg, "failed to get reminder") {
+				return
+			}
+			assert.Empty(c, reminder, "Reminder %s exist", reminderName)
+		}, secondsToCheckTimerAndReminderResult*time.Second, time.Second, "reminder was not deleted by the actor")
+		time.Sleep(2 * time.Second)
 
 		res, err = httpGet(logsURL)
 		if err != nil {
@@ -623,19 +626,19 @@ func TestActorFeatures(t *testing.T) {
 		}
 		require.NoError(t, err, "failed to set timer")
 
-		time.Sleep(secondsToCheckTimerAndReminderResult * time.Second)
+		// The assertion is a lower bound, so poll the logs until the timer has
+		// fired enough times instead of waiting a fixed window
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			logs, errl := httpGet(logsURL)
+			if !assert.NoError(c, errl, "failed to get logs") {
+				return
+			}
+			assert.GreaterOrEqual(c, countActorAction(logs, actorID, timerName), minimumCallsForTimerAndReminderResult)
+		}, (secondsToCheckTimerAndReminderResult+10)*time.Second, time.Second, "timer did not fire enough times")
 
 		// Reset timer
 		_, err = httpDelete(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorID, "timers", timerName))
 		require.NoError(t, err)
-
-		res, err = httpGet(logsURL)
-		if err != nil {
-			log.Printf("failed to get logs. Error='%v' Response='%s'", err, string(res))
-		}
-		require.NoError(t, err, "failed to get logs")
-		require.True(t, countActorAction(res, actorID, timerName) >= 1)
-		require.True(t, countActorAction(res, actorID, timerName) >= minimumCallsForTimerAndReminderResult)
 	})
 
 	t.Run("Actor reset timer.", func(t *testing.T) {
@@ -801,20 +804,14 @@ func TestActorFeatures(t *testing.T) {
 
 		logsURL = fmt.Sprintf(actorlogsURLFormat, externalURL)
 
+		// Poll until the actors are redistributed and the actor activates on
+		// the replacement pod
 		newHostname := []byte{}
-		for i := 0; i <= actorInvokeRetriesAfterRestart; i++ {
-			// wait until actors are redistributed.
-			time.Sleep(30 * time.Second)
-
-			newHostname, err = httpPost(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorID, "method", "hostname"), []byte{})
-			if i == actorInvokeRetriesAfterRestart {
-				require.NoError(t, err, "error getting hostname")
-			}
-
-			if err == nil {
-				break
-			}
-		}
+		require.Eventually(t, func() bool {
+			var errh error
+			newHostname, errh = httpPost(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorID, "method", "hostname"), []byte{})
+			return errh == nil
+		}, 330*time.Second, 2*time.Second, "error getting hostname after restart")
 
 		require.NotEqual(t, string(firstHostname), string(newHostname))
 		close(quit)
@@ -836,25 +833,31 @@ func TestActorFeatures(t *testing.T) {
 
 		tr.Platform.Scale(appName, appScaleToCheckRebalance)
 
-		// wait until actors are redistributed.
-		time.Sleep(30 * time.Second)
-
-		anyActorMoved := false
-		for index := 0; index < actorsToCheckRebalance; index++ {
-			hostname, err := httpPost(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorIDBase+strconv.Itoa(index), "method", "hostname"), []byte{})
-			require.NoError(t, err, "error getting hostname for actor %d", index)
-
-			if hostnameForActor[index] != string(hostname) {
-				anyActorMoved = true
+		// Poll until the actors are redistributed and at least one reports a
+		// different hostname
+		require.Eventually(t, func() bool {
+			for index := 0; index < actorsToCheckRebalance; index++ {
+				hostname, errh := httpPost(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorIDBase+strconv.Itoa(index), "method", "hostname"), []byte{})
+				if errh != nil {
+					return false
+				}
+				if hostnameForActor[index] != string(hostname) {
+					return true
+				}
 			}
-		}
-
-		require.True(t, anyActorMoved, "no actor moved")
+			return false
+		}, 120*time.Second, 2*time.Second, "no actor moved")
 	})
 
 	t.Run("Get actor metadata", func(t *testing.T) {
 		tr.Platform.Scale(appName, appScaleToCheckMetadata)
-		time.Sleep(secondsToCheckGetMetadata * time.Second)
+		// Wait for the scale-down to complete, then a short settle for the
+		// placement table dissemination
+		require.Eventually(t, func() bool {
+			endpoints, errp := tr.Platform.GetAppPodEndpoints(appName)
+			return errp == nil && len(endpoints) == appScaleToCheckMetadata
+		}, 90*time.Second, 2*time.Second, "app did not scale down")
+		time.Sleep(10 * time.Second)
 
 		res, err = httpGet(fmt.Sprintf(actorMetadataURLFormat, externalURL))
 		log.Printf("Got metadata: Error='%v' Response='%s'", err, string(res))

tests/e2e/actor_reminder/actor_reminder_test.go

@@ -34,21 +34,20 @@ import (
 )
 
 const (
-	appName                      = "actorreminder"                  // App name in Dapr.
-	actorIDRestartTemplate       = "actor-reminder-restart-test-%d" // Template for Actor ID
-	restartReminderName          = "RestartTestReminder"            // Reminder name
-	actorIDGetTemplate           = "actor-reminder-get-test-%d"     // Template for Actor ID
-	reminderNameForGet           = "GetTestReminder"                // Reminder name for getting tests
-	numIterations                = 7                                // Number of times each test should run.
-	numHealthChecks              = 60                               // Number of get calls before starting tests.
-	numActorsPerThread           = 10                               // Number of get calls before starting tests.
-	secondsToCheckReminderResult = 20                               // How much time to wait to make sure the result is in logs.
-	actorName                    = "testactorreminder"              // Actor name
-	actorNameMis                 = "testactorremindermiss"          // Actor name
-	actorInvokeURLFormat         = "%s/test/%s/%s/%s/%s"            // URL to invoke a Dapr's actor method in test app.
-	actorlogsURLFormat           = "%s/test/logs"                   // URL to fetch logs from test app.
-	shutdownURLFormat            = "%s/test/shutdown"               // URL to shutdown sidecar and app.
-	misconfiguredAppName         = "actor-reminder-no-state-store"  // Actor-reminder app without a state store (should fail to start)
+	appName                = "actorreminder"                  // App name in Dapr.
+	actorIDRestartTemplate = "actor-reminder-restart-test-%d" // Template for Actor ID
+	restartReminderName    = "RestartTestReminder"            // Reminder name
+	actorIDGetTemplate     = "actor-reminder-get-test-%d"     // Template for Actor ID
+	reminderNameForGet     = "GetTestReminder"                // Reminder name for getting tests
+	numIterations          = 7                                // Number of times each test should run.
+	numHealthChecks        = 60                               // Number of get calls before starting tests.
+	numActorsPerThread     = 10                               // Number of get calls before starting tests.
+	actorName              = "testactorreminder"              // Actor name
+	actorNameMis           = "testactorremindermiss"          // Actor name
+	actorInvokeURLFormat   = "%s/test/%s/%s/%s/%s"            // URL to invoke a Dapr's actor method in test app.
+	actorlogsURLFormat     = "%s/test/logs"                   // URL to fetch logs from test app.
+	shutdownURLFormat      = "%s/test/shutdown"               // URL to shutdown sidecar and app.
+	misconfiguredAppName   = "actor-reminder-no-state-store"  // Actor-reminder app without a state store (should fail to start)
 )
 
 // represents a response for the APIs in this app.
@@ -199,8 +198,19 @@ func testActorReminder(t *testing.T, appName, actorName string) {
 					require.NoError(t, errInternal)
 				}
 
-				t.Logf("Sleeping for %d seconds ...", secondsToCheckReminderResult)
-				time.Sleep(secondsToCheckReminderResult * time.Second)
+				// The check below is a lower bound (at least one trigger per
+				// actor), so poll the logs until every reminder in this
+				// iteration has fired instead of waiting a fixed window
+				require.EventuallyWithT(t, func(c *assert.CollectT) {
+					resp, errGet := utils.HTTPGet(logsURL)
+					if !assert.NoError(c, errGet) {
+						return
+					}
+					for i := 0; i < numActorsPerThread; i++ {
+						actorID := fmt.Sprintf(actorIDRestartTemplate, i+(1000*iteration))
+						assert.GreaterOrEqual(c, countActorAction(resp, actorID, restartReminderName), 1)
+					}
+				}, 60*time.Second, time.Second, "not all reminders triggered")
 
 				for i := 0; i < numActorsPerThread; i++ {
 					_, err := utils.HTTPGetNTimes(externalURL, numHealthChecks)
@@ -239,7 +249,9 @@ func testActorReminder(t *testing.T, appName, actorName string) {
 					return err
 				}
 
-				time.Sleep(secondsToCheckReminderResult * time.Second)
+				// Negative window: with a 1s reminder period, 10 quiet seconds
+				// between the two snapshots shows the unregistration took hold
+				time.Sleep(10 * time.Second)
 
 				resp2, err := utils.HTTPGet(logsURL)
 				if err != nil {
@@ -275,8 +287,10 @@ func testActorReminder(t *testing.T, appName, actorName string) {
 		_, err = utils.HTTPPost(fmt.Sprintf(shutdownURLFormat, externalURL), []byte(""))
 		require.NoError(t, err)
 
-		t.Logf("Sleeping for %d seconds to see if reminders will trigger ...", secondsToCheckReminderResult)
-		time.Sleep(secondsToCheckReminderResult * time.Second)
+		// Negative window: with a 1s reminder period, 10 quiet seconds on top
+		// of the restart itself is enough for a lingering reminder to show up
+		t.Log("Sleeping to see if reminders will trigger ...")
+		time.Sleep(10 * time.Second)
 
 		// Re-establish port forwarding after app restart, likely connection would be lost to pod.
 		// replace externalUrl and Logs URL since the new port will be assigned
@@ -330,9 +344,6 @@ func testActorReminder(t *testing.T, appName, actorName string) {
 					_, errInternal = utils.HTTPPost(fmt.Sprintf(actorInvokeURLFormat, externalURL, actorName, actorID, "reminders", reminderNameForGet), reminderBody)
 					require.NoError(t, errInternal)
 				}
-
-				t.Logf("Sleeping for %d seconds ...", secondsToCheckReminderResult)
-				time.Sleep(secondsToCheckReminderResult * time.Second)
 			}(iteration)
 		}
 		wg.Wait()

tests/e2e/actor_state/actor_state_test.go

@@ -80,8 +80,12 @@ func TestActorState(t *testing.T) {
 	_, err := utils.HTTPGetNTimes(externalURL, numHealthChecks)
 	require.NoError(t, err)
 
-	// Wait until runtime finds the leader of placements.
-	time.Sleep(15 * time.Second)
+	// Wait until the actor runtime is ready (placement leader found) by
+	// polling an actor state read with a throwaway actor ID
+	require.Eventually(t, func() bool {
+		_, code, errp := utils.HTTPGetWithStatus(fmt.Sprintf("%s/httpMyActorType/warmup-probe-myActorID", initActorURL))
+		return errp == nil && code == http.StatusOK
+	}, 60*time.Second, time.Second, "actor runtime was not ready in time")
 
 	t.Run("http", func(t *testing.T) {
 		t.Run("getting state which does not exist should error", func(t *testing.T) {

tests/e2e/allowlists/allowlists_service_invocation_test.go

@@ -77,17 +77,23 @@ func TestMain(m *testing.M) {
 			AppProtocol:    "grpc",
 		},
 		{
-			AppName:        "grpcproxyclient",
+			// The service_invocation suite deploys the same proxy apps under the
+			// unprefixed names; distinct names keep the two suites isolated when
+			// test packages run concurrently
+			AppName:        "allowlists-grpcproxyclient",
 			DaprEnabled:    true,
 			ImageName:      "e2e-service_invocation_grpc_proxy_client",
 			Replicas:       1,
 			AppProtocol:    "grpc",
 			IngressEnabled: true,
 			MetricsEnabled: true,
+			AppEnv: map[string]string{
+				"PROXY_SERVER_APP_ID": "allowlists-grpcproxyserver",
+			},
 		},
 		{
 			Config:         "allowlistsgrpcappconfig",
-			AppName:        "grpcproxyserver",
+			AppName:        "allowlists-grpcproxyserver",
 			DaprEnabled:    true,
 			ImageName:      "e2e-service_invocation_grpc_proxy_server",
 			Replicas:       1,
@@ -170,7 +176,7 @@ var allowListsForServiceInvocationForProxyTests = []struct {
 	{
 		"Test allow with callee side grpc without http verb by grpc proxy",
 		"",
-		"grpcproxyclient",
+		"allowlists-grpcproxyclient",
 		"",
 		"success",
 		"grpc",
@@ -179,7 +185,7 @@ var allowListsForServiceInvocationForProxyTests = []struct {
 }
 
 func TestServiceInvocationWithAllowListsForGrpcProxy(t *testing.T) {
-	externalURL := tr.Platform.AcquireAppExternalURL("grpcproxyclient")
+	externalURL := tr.Platform.AcquireAppExternalURL("allowlists-grpcproxyclient")
 	require.NotEmpty(t, externalURL, "external URL must not be empty!")
 	var err error
 	// This initial probe makes the test wait a little bit longer when needed,

tests/e2e/bindings/bindings_test.go

@@ -232,42 +232,39 @@ func testBindingsForApp(app struct {
 		// act for http
 		httpPostWithAssert(t, fmt.Sprintf("%s/tests/send", outputExternalURL), body, http.StatusOK)
 
-		// This delay allows all the messages to reach corresponding input bindings.
-		time.Sleep(bindingPropagationDelay * time.Second)
-
-		// assert for HTTP
-		resp := httpPostWithAssert(t, fmt.Sprintf("%s/tests/get_received_topics", inputExternalURL), nil, http.StatusOK)
-
+		// Poll until all the messages have reached the corresponding input
+		// bindings; the sets converge to the asserted state
 		var decodedResponse receivedTopicsResponse
-		err = json.Unmarshal(resp, &decodedResponse)
-		require.NoError(t, err)
-
-		// Only the first message fails, all other messages are successfully consumed.
-		// nine messages succeed.
-		require.Equal(t, testMessages[1:], decodedResponse.ReceivedMessages)
-		// one message fails.
-		require.Equal(t, testMessages[0], decodedResponse.FailedMessage)
-		// routed binding will receive all messages
-		require.Equal(t, testMessages[0:], decodedResponse.RoutedMessages)
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			resp := httpPostWithAssert(t, fmt.Sprintf("%s/tests/get_received_topics", inputExternalURL), nil, http.StatusOK)
+			if !assert.NoError(c, json.Unmarshal(resp, &decodedResponse)) {
+				return
+			}
+			// Only the first message fails, all other messages are successfully consumed.
+			// nine messages succeed.
+			assert.Equal(c, testMessages[1:], decodedResponse.ReceivedMessages)
+			// one message fails.
+			assert.Equal(c, testMessages[0], decodedResponse.FailedMessage)
+			// routed binding will receive all messages
+			assert.Equal(c, testMessages[0:], decodedResponse.RoutedMessages)
+		}, bindingPropagationDelay*2*time.Second, time.Second, "messages did not reach the input bindings")
 
 		// act for gRPC
 		httpPostWithAssert(t, fmt.Sprintf("%s/tests/sendGRPC", outputExternalURL), body, http.StatusOK)
 
-		// This delay allows all the messages to reach corresponding input bindings.
-		time.Sleep(bindingPropagationDelay * time.Second)
-
-		// assert for gRPC
-		resp = httpPostWithAssert(t, fmt.Sprintf("%s/tests/get_received_topics_grpc", outputExternalURL), nil, http.StatusOK)
-
-		// assert for gRPC
-		err = json.Unmarshal(resp, &decodedResponse)
-		require.NoError(t, err)
-
-		// Only the first message fails, all other messages are successfully consumed.
-		// nine messages succeed.
-		require.Equal(t, testMessages[1:], decodedResponse.ReceivedMessages)
-		// one message fails.
-		require.Equal(t, testMessages[0], decodedResponse.FailedMessage)
+		// Poll until all the messages have reached the corresponding input
+		// bindings; the sets converge to the asserted state
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			resp := httpPostWithAssert(t, fmt.Sprintf("%s/tests/get_received_topics_grpc", outputExternalURL), nil, http.StatusOK)
+			if !assert.NoError(c, json.Unmarshal(resp, &decodedResponse)) {
+				return
+			}
+			// Only the first message fails, all other messages are successfully consumed.
+			// nine messages succeed.
+			assert.Equal(c, testMessages[1:], decodedResponse.ReceivedMessages)
+			// one message fails.
+			assert.Equal(c, testMessages[0], decodedResponse.FailedMessage)
+		}, bindingPropagationDelay*2*time.Second, time.Second, "messages did not reach the input bindings")
 	}
 }
 

tests/e2e/bulk-pubsub/bulk_pubsub_test.go

@@ -51,7 +51,8 @@ const (
 	numberOfMessagesToPublish = 60
 	publishRateLimitRPS       = 25
 
-	receiveMessageRetries = 5
+	// Validation polls every 2s up to this many attempts (~50s budget)
+	receiveMessageRetries = 25
 
 	publisherAppName      = "pubsub-publisher-bulk-subscribe"
 	bulkSubscriberAppName = "pubsub-bulk-subscriber"
@@ -219,7 +220,7 @@ func testPublishBulkSubscribeSuccessfully(t *testing.T, publisherExternalURL, su
 	log.Printf("Test publish bulk subscribe success flow\n")
 	sentMessages := testPublishForBulkSubscribe(t, publisherExternalURL, protocol)
 
-	time.Sleep(5 * time.Second)
+	time.Sleep(2 * time.Second)
 	validateMessagesReceivedWhenSomeTopicsBulkSubscribed(t, publisherExternalURL, bulkSubscriberAppName, protocol, false, sentMessages)
 	return subscriberExternalURL
 }
@@ -301,7 +302,7 @@ func validateMessagesReceivedWhenSomeTopicsBulkSubscribed(t *testing.T, publishe
 		log.Printf("(reqID=%s) Attempt %d complete; took %s", request.ReqID, retryCount, utils.FormatDuration(time.Now().Sub(start)))
 		if err != nil {
 			log.Printf("(reqID=%s) Error in response: %v", request.ReqID, err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -309,7 +310,7 @@ func validateMessagesReceivedWhenSomeTopicsBulkSubscribed(t *testing.T, publishe
 		if err != nil {
 			err = fmt.Errorf("(reqID=%s) failed to unmarshal JSON. Error: %v. Raw data: %s", request.ReqID, err, string(resp))
 			log.Printf("Error in response: %v", err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -326,7 +327,7 @@ func validateMessagesReceivedWhenSomeTopicsBulkSubscribed(t *testing.T, publishe
 			len(appResp.ReceivedByTopicRawBulkSub) != len(sentMessages.ReceivedByTopicRawBulkSub) ||
 			len(appResp.ReceivedByTopicCEBulkSub) != len(sentMessages.ReceivedByTopicCEBulkSub) {
 			log.Printf("Differing lengths in received vs. sent messages, retrying.")
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 		} else {
 			break
 		}

tests/e2e/pubsub/pubsub_test.go

@@ -50,7 +50,10 @@ const (
 	numberOfMessagesToPublish = 40
 	publishRateLimitRPS       = 25
 
-	receiveMessageRetries = 5
+	// Validation polls every 2s up to this many attempts (~50s budget); a
+	// short interval lets positive validations return as soon as the
+	// expected messages have all been recorded.
+	receiveMessageRetries = 25
 
 	metadataPrefix             = "metadata."
 	publisherAppName           = "pubsub-publisher"
@@ -331,7 +334,9 @@ func testDropToDeadLetter(t *testing.T, publisherExternalURL, subscriberExternal
 	require.NoError(t, err)
 	offset += numberOfMessagesToPublish + 1
 
-	time.Sleep(5 * time.Second)
+	// Brief settle for the empty-topic assertions; the positive counts below
+	// converge via the validation's own polling
+	time.Sleep(2 * time.Second)
 	received := receivedMessagesResponse{
 		ReceivedByTopicA:          sentTopicNormal,
 		ReceivedByTopicB:          []string{},
@@ -365,11 +370,13 @@ func testResiliencyExhaustion(t *testing.T, publisherExternalURL, subscriberExte
 	_ = sentMessages
 
 	// After exhaustion, messages should be ACK'd and dropped. The wait is
-	// comfortably longer than the pubsubRetry budget (maxRetries=5); this
-	// assertion only requires that the messages are not delivered to the
-	// app, which holds whether they are still retrying or already dropped.
+	// comfortably longer than the pubsubRetry budget (maxRetries=5 at 1s,
+	// plus a 1-3s broker redelivery cycle per attempt); this assertion only
+	// requires that the messages are not delivered to the app, which holds
+	// whether they are still retrying or already dropped. Any stragglers
+	// beyond the wait are flushed by the next scenario's backlog drain.
 	log.Printf("Waiting for resiliency policy to exhaust retries...")
-	time.Sleep(65 * time.Second)
+	time.Sleep(20 * time.Second)
 
 	log.Printf("Validating messages were dropped after retry exhaustion...")
 	validateMessagesReceivedBySubscriber(t, publisherExternalURL, subscriberAppName, protocol, false, receivedMessagesResponse{
@@ -409,7 +416,7 @@ func testBulkPublishSuccessfully(t *testing.T, publisherExternalURL, subscriberE
 	log.Printf("Test bulkPublish and normal subscribe success flow")
 	sentMessages := testPublishBulk(t, publisherExternalURL, protocol)
 
-	time.Sleep(10 * time.Second)
+	time.Sleep(2 * time.Second)
 	validateBulkMessagesReceivedBySubscriber(t, publisherExternalURL, subscriberAppName, protocol, sentMessages, podEndpoints)
 	return subscriberExternalURL
 }
@@ -426,7 +433,7 @@ func testPublishSubscribeSuccessfully(t *testing.T, publisherExternalURL, subscr
 	log.Print("Test publish subscribe success flow")
 	sentMessages := testPublish(t, publisherExternalURL, protocol)
 
-	time.Sleep(5 * time.Second)
+	time.Sleep(2 * time.Second)
 	validateMessagesReceivedBySubscriber(t, publisherExternalURL, subscriberAppName, protocol, false, sentMessages, podEndpoints)
 	return subscriberExternalURL
 }
@@ -475,7 +482,7 @@ func testValidateRedeliveryOrEmptyJSON(t *testing.T, publisherExternalURL, subsc
 
 	if subscriberResponse == "empty-json" {
 		// on empty-json response case immediately validate the received messages
-		time.Sleep(10 * time.Second)
+		time.Sleep(2 * time.Second)
 		validateMessagesReceivedBySubscriber(t, publisherExternalURL, subscriberAppName, protocol, false, sentMessages, podEndpoints)
 
 		callInitialize(t, subscriberAppName, publisherExternalURL, protocol)
@@ -506,7 +513,7 @@ func testValidateRedeliveryOrEmptyJSON(t *testing.T, publisherExternalURL, subsc
 			got, err := subscriberReceivedDeadLetterCount(publisherExternalURL, subscriberAppName, protocol, podEndpoints)
 			assert.NoError(c, err, "error calling subscriber to get dead letter count")
 			assert.Equal(c, len(sentMessages.ReceivedByTopicDeadLetter), got)
-		}, 300*time.Second, 5*time.Second,
+		}, 300*time.Second, time.Second,
 			"subscriber did not receive all dead letter messages within timeout")
 
 		// Now flip to success so the non-dead-letter topics (a/b/c/raw)
@@ -527,8 +534,12 @@ func testValidateRedeliveryOrEmptyJSON(t *testing.T, publisherExternalURL, subsc
 	setDesiredResponse(t, subscriberAppName, "success", publisherExternalURL, protocol)
 
 	if subscriberResponse == "empty-json" {
+		// Negative assertion: nothing may be redelivered after the empty-json
+		// acknowledgements. The window must outlast the redelivery horizon,
+		// which is bounded by the pubsubRetry budget (maxRetries=5 at 1s) plus
+		// a 1-3s broker redelivery cycle per attempt.
 		log.Printf("Validating no redelivered messages for 'empty-json' subscriber...")
-		time.Sleep(30 * time.Second)
+		time.Sleep(15 * time.Second)
 		validateMessagesReceivedBySubscriber(t, publisherExternalURL, subscriberAppName, protocol, false, receivedMessagesResponse{
 			// empty string slices
 			ReceivedByTopicA:    []string{},
@@ -538,9 +549,9 @@ func testValidateRedeliveryOrEmptyJSON(t *testing.T, publisherExternalURL, subsc
 			ReceivedByTopicDead: []string{},
 		}, podEndpoints)
 	} else {
-		// validate redelivery of messages (retry / invalid-status cases)
+		// validate redelivery of messages (retry / invalid-status cases); the
+		// validation polls until the redelivered counts converge
 		log.Printf("Validating redelivered messages...")
-		time.Sleep(30 * time.Second)
 		validateMessagesReceivedBySubscriber(t, publisherExternalURL, subscriberAppName, protocol, false, sentMessages, podEndpoints)
 	}
 
@@ -648,10 +659,13 @@ func drainSubscriberBacklog(t *testing.T, publisherExternalURL, subscriberAppNam
 			stable = 0
 			last = got
 		}
-		// Three consecutive quiet polls (~15s) with no new arrivals means
-		// the broker has no more in-flight redeliveries to hand us.
-		return stable >= 3
-	}, 180*time.Second, 5*time.Second, "subscriber backlog did not drain before scenario start")
+		// Eight consecutive quiet polls (~8s) with no new arrivals means the
+		// broker has no more in-flight redeliveries to hand us: with the
+		// subscriber in success mode the gap between redelivery waves is
+		// bounded by processingTimeout + idleCheckFrequency (1s each), so a
+		// quiet window several times that is a drained backlog.
+		return stable >= 8
+	}, 180*time.Second, time.Second, "subscriber backlog did not drain before scenario start")
 }
 
 func validateBulkMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL string, subscriberApp string, protocol string, sentMessages receivedMessagesResponse, podEndpoints []string) {
@@ -677,7 +691,7 @@ func validateBulkMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL
 		log.Printf("(reqID=%s) Attempt %d complete; took %s", request.ReqID, retryCount, utils.FormatDuration(time.Now().Sub(start)))
 		if err != nil {
 			log.Printf("(reqID=%s) Error in response: %v", request.ReqID, err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -685,7 +699,7 @@ func validateBulkMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL
 		if err != nil {
 			err = fmt.Errorf("(reqID=%s) failed to unmarshal JSON. Error: %v. Raw data: %s", request.ReqID, err, string(resp))
 			log.Printf("Error in response: %v", err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -703,7 +717,7 @@ func validateBulkMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL
 			len(appResp.ReceivedByTopicCEBulk) != len(sentMessages.ReceivedByTopicCEBulk) ||
 			len(appResp.ReceivedByTopicDefBulk) != len(sentMessages.ReceivedByTopicDefBulk) {
 			log.Printf("Differing lengths in received vs. sent messages, retrying.")
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 		} else {
 			break
 		}
@@ -749,7 +763,7 @@ func validateMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL str
 		log.Printf("(reqID=%s) Attempt %d complete; took %s", request.ReqID, retryCount, utils.FormatDuration(time.Now().Sub(start)))
 		if err != nil {
 			log.Printf("(reqID=%s) Error in response: %v", request.ReqID, err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -757,7 +771,7 @@ func validateMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL str
 		if err != nil {
 			err = fmt.Errorf("(reqID=%s) failed to unmarshal JSON. Error: %v. Raw data: %s", request.ReqID, err, string(resp))
 			log.Printf("Error in response: %v", err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -776,7 +790,7 @@ func validateMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL str
 			len(appResp.ReceivedByTopicRaw) != len(sentMessages.ReceivedByTopicRaw) ||
 			(validateDeadLetter && len(appResp.ReceivedByTopicDeadLetter) != len(sentMessages.ReceivedByTopicDeadLetter)) {
 			log.Printf("Differing lengths in received vs. sent messages, retrying.")
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 		} else {
 			break
 		}

tests/e2e/pubsub_grpc/pubsub_grpc_test.go

@@ -49,7 +49,10 @@ const (
 	randomOffsetMax           = 49
 	numberOfMessagesToPublish = 60
 	publishRateLimitRPS       = 25
-	receiveMessageRetries     = 5
+	// Validation polls every 2s up to this many attempts (~50s budget); a
+	// short interval lets positive validations return as soon as the
+	// expected messages have all been recorded.
+	receiveMessageRetries = 25
 
 	metadataPrefix    = "metadata."
 	publisherAppName  = "pubsub-publisher-grpc"
@@ -370,7 +373,7 @@ func waitForSubscriberReady(t *testing.T, subscriberExternalURL, protocol string
 			conn.Close()
 			log.Printf("gRPC subscriber at %s is ready", subscriberExternalURL)
 			return true
-		}, 50*time.Second, 5*time.Second, "gRPC subscriber not ready after retries")
+		}, 50*time.Second, time.Second, "gRPC subscriber not ready after retries")
 	}
 }
 
@@ -385,7 +388,7 @@ func testBulkPublishSuccessfully(t *testing.T, publisherExternalURL, subscriberE
 	log.Printf("Test bulkPublish and normal subscribe success flow")
 	sentMessages := testPublishBulk(t, publisherExternalURL, protocol)
 
-	time.Sleep(10 * time.Second)
+	time.Sleep(2 * time.Second)
 	validateBulkMessagesReceivedBySubscriber(t, publisherExternalURL, subscriberAppName, protocol, sentMessages)
 	return subscriberExternalURL
 }
@@ -412,7 +415,7 @@ func testPublishBulkSubscribeSuccessfully(t *testing.T, publisherExternalURL, su
 	log.Printf("Test publish bulk subscribe success flow\n")
 	sentMessages := testPublishForBulkSubscribe(t, publisherExternalURL, protocol)
 
-	time.Sleep(5 * time.Second)
+	time.Sleep(2 * time.Second)
 	validateMessagesReceivedWhenSomeTopicsBulkSubscribed(t, publisherExternalURL, subscriberAppName, protocol, sentMessages)
 	return subscriberExternalURL
 }
@@ -439,7 +442,9 @@ func testPublishWithoutTopic(t *testing.T, publisherExternalURL, subscriberExter
 	return subscriberExternalURL
 }
 
-// testResiliencyExhaustion: Wait for resiliency policy to exhaust (60 retries @ 1s = 60s + buffer)
+// testResiliencyExhaustion verifies that messages are not delivered to the
+// app while the subscriber keeps erroring: they are retried per the
+// pubsubRetry policy (maxRetries=5 at 1s) and then dropped.
 func testResiliencyExhaustion(t *testing.T, publisherExternalURL, subscriberExternalURL, subscriberResponse, subscriberAppName, protocol string) string {
 	var err error
 	var code int
@@ -459,7 +464,7 @@ func testResiliencyExhaustion(t *testing.T, publisherExternalURL, subscriberExte
 	var lastRetryError error
 	for retryCount := 0; retryCount < receiveMessageRetries; retryCount++ {
 		if retryCount > 0 {
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 		}
 		lastRetryError = nil
 		_, code, err = utils.HTTPPostWithStatus(publisherExternalURL+"/tests/callSubscriberMethod", reqBytes)
@@ -478,9 +483,13 @@ func testResiliencyExhaustion(t *testing.T, publisherExternalURL, subscriberExte
 	sentMessages := testPublish(t, publisherExternalURL, protocol)
 	_ = sentMessages
 
-	// After exhaustion, messages should be ACK'd and dropped
-	log.Printf("Waiting 65 seconds for resiliency policy to exhaust retries (maxRetries=60)...")
-	time.Sleep(65 * time.Second)
+	// After exhaustion, messages should be ACK'd and dropped. The wait is
+	// comfortably longer than the pubsubRetry budget (maxRetries=5 at 1s,
+	// plus a 1-3s broker redelivery cycle per attempt); this assertion only
+	// requires that the messages are not delivered to the app, which holds
+	// whether they are still retrying or already dropped.
+	log.Printf("Waiting for resiliency policy to exhaust retries...")
+	time.Sleep(20 * time.Second)
 
 	log.Printf("Validating messages were dropped after retry exhaustion...")
 	validateMessagesReceivedBySubscriber(t, publisherExternalURL, subscriberAppName, protocol, receivedMessagesResponse{
@@ -517,7 +526,7 @@ func testValidateRedeliveryOrEmptyJSON(t *testing.T, publisherExternalURL, subsc
 	var lastRetryError error
 	for retryCount := 0; retryCount < receiveMessageRetries; retryCount++ {
 		if retryCount > 0 {
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 		}
 		lastRetryError = nil
 		_, code, err = utils.HTTPPostWithStatus(publisherExternalURL+"/tests/callSubscriberMethod", reqBytes)
@@ -578,7 +587,7 @@ func setDesiredResponse(t *testing.T, subscriberAppName, subscriberResponse, pub
 	var lastRetryError error
 	for retryCount := 0; retryCount < receiveMessageRetries; retryCount++ {
 		if retryCount > 0 {
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 		}
 		lastRetryError = nil
 		_, code, err := utils.HTTPPostWithStatus(publisherExternalURL+"/tests/callSubscriberMethod", reqBytes)
@@ -618,7 +627,7 @@ func validateBulkMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL
 		log.Printf("(reqID=%s) Attempt %d complete; took %s", request.ReqID, retryCount, utils.FormatDuration(time.Now().Sub(start)))
 		if err != nil {
 			log.Printf("(reqID=%s) Error in response: %v", request.ReqID, err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -626,7 +635,7 @@ func validateBulkMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL
 		if err != nil {
 			err = fmt.Errorf("(reqID=%s) failed to unmarshal JSON. Error: %v. Raw data: %s", request.ReqID, err, string(resp))
 			log.Printf("Error in response: %v", err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -644,7 +653,7 @@ func validateBulkMessagesReceivedBySubscriber(t *testing.T, publisherExternalURL
 			len(appResp.ReceivedByTopicCEBulk) != len(sentMessages.ReceivedByTopicCEBulk) ||
 			len(appResp.ReceivedByTopicDefBulk) != len(sentMessages.ReceivedByTopicDefBulk) {
 			log.Printf("Differing lengths in received vs. sent messages, retrying.")
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 		} else {
 			break
 		}
@@ -691,7 +700,7 @@ func validateMessagesReceivedBySubscriber(
 		log.Printf("(reqID=%s) Attempt %d complete; took %s", request.ReqID, retryCount, utils.FormatDuration(time.Now().Sub(start)))
 		if err != nil {
 			log.Printf("(reqID=%s) Error in response: %v", request.ReqID, err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -699,7 +708,7 @@ func validateMessagesReceivedBySubscriber(
 		if err != nil {
 			err = fmt.Errorf("(reqID=%s) failed to unmarshal JSON. Error: %v. Raw data: %s", request.ReqID, err, string(resp))
 			log.Printf("Error in response: %v", err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -716,7 +725,7 @@ func validateMessagesReceivedBySubscriber(
 			len(appResp.ReceivedByTopicC) != len(sentMessages.ReceivedByTopicC) ||
 			len(appResp.ReceivedByTopicRaw) != len(sentMessages.ReceivedByTopicRaw) {
 			log.Printf("Differing lengths in received vs. sent messages, retrying.")
-			time.Sleep(5 * time.Second)
+			time.Sleep(2 * time.Second)
 		} else {
 			break
 		}
@@ -763,7 +772,7 @@ func validateMessagesReceivedWhenSomeTopicsBulkSubscribed(
 		log.Printf("(reqID=%s) Attempt %d complete; took %s", request.ReqID, retryCount, utils.FormatDuration(time.Now().Sub(start)))
 		if err != nil {
 			log.Printf("(reqID=%s) Error in response: %v", request.ReqID, err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -771,7 +780,7 @@ func validateMessagesReceivedWhenSomeTopicsBulkSubscribed(
 		if err != nil {
 			err = fmt.Errorf("(reqID=%s) failed to unmarshal JSON. Error: %v. Raw data: %s", request.ReqID, err, string(resp))
 			log.Printf("Error in response: %v", err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -788,7 +797,7 @@ func validateMessagesReceivedWhenSomeTopicsBulkSubscribed(
 			len(appResp.ReceivedByTopicRawBulkSub) != len(sentMessages.ReceivedByTopicRawBulkSub) ||
 			len(appResp.ReceivedByTopicCEBulkSub) != len(sentMessages.ReceivedByTopicCEBulkSub) {
 			log.Printf("Differing lengths in received vs. sent messages, retrying.")
-			time.Sleep(5 * time.Second)
+			time.Sleep(2 * time.Second)
 		} else {
 			break
 		}

tests/e2e/pubsub_routing/pubsub_routing_test.go

@@ -47,7 +47,8 @@ const (
 	numberOfMessagesToPublish = 60
 	publishRateLimitRPS       = 25
 
-	receiveMessageRetries = 5
+	// Validation polls every 2s up to this many attempts (~50s budget)
+	receiveMessageRetries = 25
 
 	publisherAppName  = "pubsub-publisher-routing"
 	subscriberAppName = "pubsub-subscriber-routing"
@@ -173,7 +174,7 @@ func testPublishSubscribeRouting(t *testing.T, publisherExternalURL, subscriberE
 	callInitialize(t, publisherExternalURL, protocol)
 	sentMessages := testPublishRouting(t, publisherExternalURL, protocol)
 
-	time.Sleep(5 * time.Second)
+	time.Sleep(2 * time.Second)
 	validateMessagesRouted(t, publisherExternalURL, subscriberAppName, protocol, sentMessages)
 	return subscriberExternalURL
 }
@@ -238,7 +239,7 @@ func validateMessagesRouted(t *testing.T, publisherExternalURL string, subscribe
 		log.Printf("(reqID=%s) Attempt %d complete; took %s", request.ReqID, retryCount, utils.FormatDuration(time.Now().Sub(start)))
 		if err != nil {
 			log.Printf("(reqID=%s) Error in response: %v", request.ReqID, err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -246,7 +247,7 @@ func validateMessagesRouted(t *testing.T, publisherExternalURL string, subscribe
 		if err != nil {
 			err = fmt.Errorf("(reqID=%s) failed to unmarshal JSON. Error: %v. Raw data: %s", request.ReqID, err, string(resp))
 			log.Printf("Error in response: %v", err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -267,7 +268,7 @@ func validateMessagesRouted(t *testing.T, publisherExternalURL string, subscribe
 			len(appResp.RouteE) != len(sentMessages.RouteE) ||
 			len(appResp.RouteF) != len(sentMessages.RouteF) {
 			log.Printf("Differing lengths in received vs. sent messages, retrying.")
-			time.Sleep(5 * time.Second)
+			time.Sleep(2 * time.Second)
 		} else {
 			break
 		}

tests/e2e/pubsub_routing_grpc/pubsub_routing_grpc_test.go

@@ -50,7 +50,8 @@ const (
 	numberOfMessagesToPublish = 60
 	publishRateLimitRPS       = 25
 
-	receiveMessageRetries = 5
+	// Validation polls every 2s up to this many attempts (~50s budget)
+	receiveMessageRetries = 25
 
 	publisherAppName  = "pubsub-publisher-routing-grpc"
 	subscriberAppName = "pubsub-subscriber-routing-grpc"
@@ -176,7 +177,7 @@ func testPublishSubscribeRouting(t *testing.T, publisherExternalURL, subscriberE
 	callInitialize(t, publisherExternalURL, protocol)
 	sentMessages := testPublishRouting(t, publisherExternalURL, protocol)
 
-	time.Sleep(5 * time.Second)
+	time.Sleep(2 * time.Second)
 	validateMessagesRouted(t, publisherExternalURL, subscriberAppName, protocol, sentMessages)
 	return subscriberExternalURL
 }
@@ -241,7 +242,7 @@ func validateMessagesRouted(t *testing.T, publisherExternalURL string, subscribe
 		log.Printf("(reqID=%s) Attempt %d complete; took %s", request.ReqID, retryCount, utils.FormatDuration(time.Now().Sub(start)))
 		if err != nil {
 			log.Printf("(reqID=%s) Error in response: %v", request.ReqID, err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -249,7 +250,7 @@ func validateMessagesRouted(t *testing.T, publisherExternalURL string, subscribe
 		if err != nil {
 			err = fmt.Errorf("(reqID=%s) failed to unmarshal JSON. Error: %v. Raw data: %s", request.ReqID, err, string(resp))
 			log.Printf("Error in response: %v", err)
-			time.Sleep(10 * time.Second)
+			time.Sleep(2 * time.Second)
 			continue
 		}
 
@@ -270,7 +271,7 @@ func validateMessagesRouted(t *testing.T, publisherExternalURL string, subscribe
 			len(appResp.RouteE) != len(sentMessages.RouteE) ||
 			len(appResp.RouteF) != len(sentMessages.RouteF) {
 			log.Printf("Differing lengths in received vs. sent messages, retrying.")
-			time.Sleep(5 * time.Second)
+			time.Sleep(2 * time.Second)
 		} else {
 			break
 		}

tests/e2e/workflows/workflow_test.go

@@ -319,13 +319,13 @@ func monitorTest(url string, instanceID string) func(t *testing.T) {
 		resp, err = utils.HTTPPost(fmt.Sprintf("%s/RaiseWorkflowEvent/dapr/%s/PayByCard/1", url, instanceID), nil)
 		require.NoError(t, err, "failure raising event on workflow")
 
-		time.Sleep(15 * time.Second)
-
+		// Positive convergence: poll straight to Completed instead of a fixed
+		// settle followed by a short poll
 		require.EventuallyWithT(t, func(t *assert.CollectT) {
 			resp, err = utils.HTTPGet(getString)
 			assert.NoError(t, err, "failure getting info on workflow")
 			assert.Equalf(t, "Completed", string(resp), "expected workflow to be Completed, actual workflow state is: %s", string(resp))
-		}, 5*time.Second, 100*time.Millisecond)
+		}, 30*time.Second, 500*time.Millisecond)
 
 		require.EventuallyWithT(t, func(t *assert.CollectT) {
 			resp, err = utils.HTTPGet(getMonitorString)

tests/runner/loadtest/k6_test.go

@@ -401,7 +401,12 @@ func TestK6(t *testing.T) {
 		called := 0
 		jobs.jobsResultF = func() *batchv1.JobList {
 			called++
-			if called == 2 {
+			// The first call comes from Dispose waiting for deletion and must
+			// see no jobs; every later call comes from waitForCompletion and
+			// must see the succeeded job. Returning the success only on one
+			// exact call count would turn any extra poll into a hang until
+			// the test binary timeout instead of a failure.
+			if called >= 2 {
 				return &batchv1.JobList{
 					Items: []batchv1.Job{{
 						Status: batchv1.JobStatus{