test: tolerate transient actor resume routing

Author: EItanya

internal/e2e/suites/demo/demo_test.go

@@ -20,6 +20,8 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"regexp"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -330,11 +332,7 @@ func pauseActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *
 	}
 	waitForActorStatus(ctx, t, clients, actorID, ateapipb.Actor_STATUS_RUNNING)
 
-	resp, err := callActor(t, demoAtespace, actorID)
-	if err != nil {
-		t.Fatalf("failed to call actor: %v", err)
-	}
-
+	resp := callActorUntilCountAtLeast(t, demoAtespace, actorID, 1)
 	if isMicroVMEnvironment() {
 		validateCounterResponse(t, resp, "after creation", 1, -1)
 	} else {
@@ -359,10 +357,7 @@ func pauseActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *
 	}
 	waitForActorStatus(ctx, t, clients, actorID, ateapipb.Actor_STATUS_RUNNING)
 
-	resp, err = callActor(t, demoAtespace, actorID)
-	if err != nil {
-		t.Fatalf("failed to call actor again: %v", err)
-	}
+	resp = callActorUntilCountAtLeast(t, demoAtespace, actorID, 2)
 	if isMicroVMEnvironment() {
 		validateCounterResponse(t, resp, "after pause", 2, -1)
 	} else {
@@ -418,10 +413,7 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
 	}
 	waitForActorStatus(ctx, t, clients, actorID, ateapipb.Actor_STATUS_RUNNING)
 
-	resp, err := callActor(t, demoAtespace, actorID)
-	if err != nil {
-		t.Fatalf("failed to call actor: %v", err)
-	}
+	resp := callActorUntilCountAtLeast(t, demoAtespace, actorID, 1)
 	if isMicroVMEnvironment() {
 		validateCounterResponse(t, resp, "after creation", 1, -1)
 	} else {
@@ -446,10 +438,7 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
 	}
 	waitForActorStatus(ctx, t, clients, actorID, ateapipb.Actor_STATUS_RUNNING)
 
-	resp, err = callActor(t, demoAtespace, actorID)
-	if err != nil {
-		t.Fatalf("failed to call actor again: %v", err)
-	}
+	resp = callActorUntilCountAtLeast(t, demoAtespace, actorID, 2)
 	if isMicroVMEnvironment() {
 		validateCounterResponse(t, resp, "after suspend", 2, -1)
 	} else {
@@ -619,6 +608,51 @@ func waitForActorStatus(ctx context.Context, t *testing.T, clients *e2e.Clients,
 	t.Fatalf("timed out waiting for actor %q to reach status %v", actorID, expectedStatus)
 }
 
+var preservedCountRe = regexp.MustCompile(`preserved memory count: ([0-9]+)`)
+
+func callActorUntilCountAtLeast(t *testing.T, atespace, actorID string, minCount int) string {
+	t.Helper()
+
+	var lastErr error
+	var lastResp string
+	deadline := time.Now().Add(20 * time.Second)
+	for time.Now().Before(deadline) {
+		resp, err := callActor(t, atespace, actorID)
+		if err != nil {
+			lastErr = err
+		} else {
+			lastResp = resp
+			count, err := preservedCount(resp)
+			if err != nil {
+				lastErr = err
+			} else if count >= minCount {
+				return resp
+			} else {
+				lastErr = fmt.Errorf("expected preserved memory count >= %d, got %d in response: %s", minCount, count, resp)
+			}
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+
+	if lastResp != "" {
+		t.Fatalf("timed out calling actor %q; last response: %s; last error: %v", actorID, lastResp, lastErr)
+	}
+	t.Fatalf("timed out calling actor %q; last error: %v", actorID, lastErr)
+	return ""
+}
+
+func preservedCount(resp string) (int, error) {
+	matches := preservedCountRe.FindStringSubmatch(resp)
+	if matches == nil {
+		return 0, fmt.Errorf("response does not include preserved memory count: %s", resp)
+	}
+	count, err := strconv.Atoi(matches[1])
+	if err != nil {
+		return 0, fmt.Errorf("parse preserved memory count %q: %w", matches[1], err)
+	}
+	return count, nil
+}
+
 func callActor(t *testing.T, atespace, actorID string) (string, error) {
 	t.Helper()