Add Dockerfile for all SDK e2e tests, fix flaky e2e tests

.github/workflows/e2e-image.yml

@@ -0,0 +1,83 @@
+name: e2e-image
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - "sdks/go/**"
+      - "sdks/python/**"
+      - "sdks/typescript/**"
+      - "sdks/ruby/**"
+
+jobs:
+  build-push-e2e-go:
+    name: hatchet-e2e-go
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - name: Login to GHCR
+        run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
+      - name: Build and push
+        run: |
+          DOCKER_BUILDKIT=1 docker build -f ./build/package/e2e-go.dockerfile \
+            -t ghcr.io/hatchet-dev/hatchet/hatchet-e2e-go:${{ github.sha }} \
+            -t ghcr.io/hatchet-dev/hatchet/hatchet-e2e-go:main \
+            --platform linux/amd64 \
+            .
+          docker push ghcr.io/hatchet-dev/hatchet/hatchet-e2e-go:${{ github.sha }}
+          docker push ghcr.io/hatchet-dev/hatchet/hatchet-e2e-go:main
+
+  build-push-e2e-python:
+    name: hatchet-e2e-python
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - name: Login to GHCR
+        run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
+      - name: Build and push
+        run: |
+          DOCKER_BUILDKIT=1 docker build -f ./build/package/e2e-python.dockerfile \
+            -t ghcr.io/hatchet-dev/hatchet/hatchet-e2e-python:${{ github.sha }} \
+            -t ghcr.io/hatchet-dev/hatchet/hatchet-e2e-python:main \
+            --platform linux/amd64 \
+            .
+          docker push ghcr.io/hatchet-dev/hatchet/hatchet-e2e-python:${{ github.sha }}
+          docker push ghcr.io/hatchet-dev/hatchet/hatchet-e2e-python:main
+
+  build-push-e2e-typescript:
+    name: hatchet-e2e-typescript
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - name: Login to GHCR
+        run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
+      - name: Build and push
+        run: |
+          DOCKER_BUILDKIT=1 docker build -f ./build/package/e2e-typescript.dockerfile \
+            -t ghcr.io/hatchet-dev/hatchet/hatchet-e2e-typescript:${{ github.sha }} \
+            -t ghcr.io/hatchet-dev/hatchet/hatchet-e2e-typescript:main \
+            --platform linux/amd64 \
+            .
+          docker push ghcr.io/hatchet-dev/hatchet/hatchet-e2e-typescript:${{ github.sha }}
+          docker push ghcr.io/hatchet-dev/hatchet/hatchet-e2e-typescript:main
+
+  build-push-e2e-ruby:
+    name: hatchet-e2e-ruby
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - name: Login to GHCR
+        run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
+      - name: Build and push
+        run: |
+          DOCKER_BUILDKIT=1 docker build -f ./build/package/e2e-ruby.dockerfile \
+            -t ghcr.io/hatchet-dev/hatchet/hatchet-e2e-ruby:${{ github.sha }} \
+            -t ghcr.io/hatchet-dev/hatchet/hatchet-e2e-ruby:main \
+            --platform linux/amd64 \
+            .
+          docker push ghcr.io/hatchet-dev/hatchet/hatchet-e2e-ruby:${{ github.sha }}
+          docker push ghcr.io/hatchet-dev/hatchet/hatchet-e2e-ruby:main

.github/workflows/sdk-python.yml

@@ -147,7 +147,7 @@ jobs:
       - name: Run pytest
         run: |
           echo "Using HATCHET_CLIENT_NAMESPACE: $HATCHET_CLIENT_NAMESPACE"
-
+          export HATCHET_CLIENT_SERVER_URL=http://localhost:8080
           poetry run pytest -s -vvv --maxfail=5 --capture=no --retries 3 --retry-delay 2 -n 8
 
       - name: Test with wheel

build/package/e2e-go.dockerfile

@@ -0,0 +1,31 @@
+# Base Go environment
+# -------------------
+FROM golang:1.25-alpine as base
+WORKDIR /hatchet
+
+COPY go.mod go.sum ./
+
+RUN go mod download
+
+COPY /pkg ./pkg
+COPY /internal ./internal
+COPY /api ./api
+COPY /sdks/go ./sdks/go
+
+# Go build environment
+# --------------------
+FROM base AS build-go
+
+RUN go test -c -tags e2e -v -o ./bin/e2e-test ./sdks/go/e2e/
+
+# Deployment environment
+# ----------------------
+FROM alpine AS deployment
+
+WORKDIR /hatchet
+
+RUN apk update && apk add --no-cache ca-certificates tzdata
+
+COPY --from=build-go /hatchet/bin/e2e-test /hatchet/
+
+CMD ["/hatchet/e2e-test", "-test.v", "-test.timeout=10m"]

build/package/e2e-python.dockerfile

@@ -0,0 +1,13 @@
+# Base Python environment
+# -----------------------
+FROM python:3.13-slim AS deployment
+
+WORKDIR /hatchet/sdks/python
+
+RUN pip install --no-cache-dir poetry==2.3.0
+
+COPY sdks/python/ .
+
+RUN poetry install --no-interaction --all-extras
+
+CMD ["poetry", "run", "pytest", "-s", "-vvv", "-x", "--capture=no"]

build/package/e2e-ruby-entrypoint.sh

@@ -0,0 +1,44 @@
+#!/bin/sh
+
+WORKER_PID=""
+FINAL_EXIT=0
+
+cleanup() {
+  if [ -n "$WORKER_PID" ]; then
+    kill -9 "$WORKER_PID" 2>/dev/null || true
+    wait "$WORKER_PID" 2>/dev/null || true
+  fi
+}
+trap cleanup EXIT INT TERM
+
+echo "=== Running integration tests ==="
+cd /hatchet/sdks/ruby/src
+timeout 300 bundle exec rspec spec/integration/ --format documentation --tag integration
+INTEGRATION_EXIT=$?
+[ $INTEGRATION_EXIT -ne 0 ] && FINAL_EXIT=$INTEGRATION_EXIT
+
+echo "=== Starting example worker ==="
+cd /hatchet/sdks/ruby/examples
+HATCHET_CLIENT_WORKER_HEALTHCHECK_ENABLED=true \
+HATCHET_CLIENT_WORKER_HEALTHCHECK_PORT=8001 \
+bundle exec ruby worker.rb &
+WORKER_PID=$!
+
+for i in $(seq 1 60); do
+    if curl -sf http://localhost:8001/health > /dev/null 2>&1; then
+        echo "Worker healthy after ${i}s"
+        break
+    fi
+    if [ "$i" -eq 60 ]; then
+        echo "Worker failed to start within 60s"
+        exit 1
+    fi
+    sleep 1
+done
+
+echo "=== Running e2e tests ==="
+timeout 1200 bundle exec rspec -f d --fail-fast
+E2E_EXIT=$?
+[ $E2E_EXIT -ne 0 ] && FINAL_EXIT=$E2E_EXIT
+
+exit $FINAL_EXIT

build/package/e2e-ruby.dockerfile

@@ -0,0 +1,18 @@
+# Base Ruby environment
+# ---------------------
+FROM ruby:3.2 AS deployment
+
+WORKDIR /hatchet/sdks/ruby
+
+RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/*
+
+COPY sdks/ruby/src/ src/
+RUN cd src && bundle lock --add-platform x86_64-linux && bundle install
+
+COPY sdks/ruby/examples/ examples/
+RUN cd examples && bundle lock --add-platform x86_64-linux && bundle install
+
+COPY build/package/e2e-ruby-entrypoint.sh /entrypoint.sh
+RUN chmod +x /entrypoint.sh
+
+CMD ["/entrypoint.sh"]

build/package/e2e-typescript.dockerfile

@@ -0,0 +1,13 @@
+# Base Node environment
+# ---------------------
+FROM node:20-alpine AS deployment
+
+WORKDIR /hatchet/sdks/typescript
+
+RUN corepack enable && corepack prepare pnpm@10.16.1 --activate
+
+COPY sdks/typescript/ .
+
+RUN pnpm install --frozen-lockfile
+
+CMD ["pnpm", "run", "test:e2e"]

examples/python/durable_eviction/worker.py

@@ -149,7 +149,7 @@ async def capacity_evictable_sleep(
 )
 async def non_evictable_sleep(input: EmptyModel, ctx: DurableContext) -> dict[str, Any]:
     """Has eviction disabled -- should never be evicted."""
-    await ctx.aio_sleep_for(timedelta(seconds=10))
+    await ctx.aio_sleep_for(timedelta(seconds=30))
     return {"status": "completed"}
 
 

pkg/worker/context.go

@@ -20,6 +20,7 @@ import (
 	"github.com/hatchet-dev/hatchet/pkg/client"
 	"github.com/hatchet-dev/hatchet/pkg/client/create"
 	"github.com/hatchet-dev/hatchet/pkg/client/types"
+	clientconfig "github.com/hatchet-dev/hatchet/pkg/config/client"
 	"github.com/hatchet-dev/hatchet/pkg/worker/condition"
 )
 
@@ -1061,6 +1062,8 @@ func (d *durableHatchetContext) SleepFor(duration time.Duration) (*SingleWaitRes
 
 // WaitForEvent implements the DurableHatchetContext.WaitForEvent method.
 func (d *durableHatchetContext) WaitForEvent(eventKey, expression string) (*SingleWaitResult, error) {
+	namespace := d.c.Namespace()
+	eventKey = clientconfig.ApplyNamespace(eventKey, &namespace)
 	wr, err := d.waitFor(condition.UserEventCondition(eventKey, expression), "wait_for_event", eventKey)
 
 	if err != nil {

sdks/go/e2e/durable_test.go

@@ -44,7 +44,9 @@ func TestDurableWorkflow(t *testing.T) {
 
 	id := uniqueID()
 
-	time.Sleep(time.Duration(sleepTime+10) * time.Second)
+	// Wait for the run to start, then let the internal SleepFor(sleepTime) finish before pushing the event.
+	pollUntilRunStatus(t, ctx, sharedClient, ref.RunId, string(rest.V1TaskStatusRUNNING))
+	time.Sleep(time.Duration(sleepTime+3) * time.Second)
 
 	err = sharedClient.Events().Push(ctx, eventKey, AwaitedEvent{ID: id})
 	require.NoError(t, err)
@@ -68,15 +70,21 @@ func TestDurableSleepCancelReplay(t *testing.T) {
 	ref, err := testWaitForSleepTwice.RunNoWait(ctx, EmptyInput{})
 	require.NoError(t, err)
 
-	time.Sleep(time.Duration(sleepTime/2) * time.Second)
+	pollUntilRunStatus(t, ctx, sharedClient, ref.RunId, string(rest.V1TaskStatusRUNNING))
 
 	_, err = sharedClient.Runs().Cancel(ctx, rest.V1CancelTaskRequest{
 		ExternalIds: toUUIDs(ref.RunId),
 	})
 	require.NoError(t, err)
 
-	// Wait for cancellation
-	time.Sleep(2 * time.Second)
+	// Wait for cancellation to propagate before replaying.
+	pollUntil(t, ctx, func() (bool, error) {
+		status, err := sharedClient.Runs().GetStatus(ctx, ref.RunId)
+		if err != nil {
+			return false, err
+		}
+		return *status == rest.V1TaskStatusCANCELLED, nil
+	})
 
 	replayStart := time.Now()
 	_, err = sharedClient.Runs().Replay(ctx, rest.V1ReplayTaskRequest{
@@ -92,8 +100,8 @@ func TestDurableSleepCancelReplay(t *testing.T) {
 	err = result.TaskOutput("wait-for-sleep-twice").Into(&output)
 	require.NoError(t, err)
 
-	assert.Less(t, output["runtime"], float64(sleepTime))
-	assert.LessOrEqual(t, replayElapsed, float64(sleepTime))
+	assert.Less(t, output["runtime"], float64(sleepTime)+timingTolerance)
+	assert.LessOrEqual(t, replayElapsed, float64(sleepTime)+timingTolerance)
 }
 
 func TestDurableChildSpawn(t *testing.T) {
@@ -122,6 +130,7 @@ func TestDurableChildBulkSpawn(t *testing.T) {
 	require.NoError(t, err)
 	outputs, ok := m["child_outputs"].([]any)
 	require.True(t, ok, "expected child_outputs to be an array")
+
 	assert.GreaterOrEqual(t, len(outputs), n-1)
 	assert.LessOrEqual(t, len(outputs), n)
 
@@ -145,7 +154,9 @@ func TestDurableSleepEventSpawnReplay(t *testing.T) {
 	ref, err := testDurableSleepEventSpawn.RunNoWait(ctx, EmptyInput{})
 	require.NoError(t, err)
 
-	time.Sleep(time.Duration(sleepTime+5) * time.Second)
+	// Wait for the run to start, then let the internal SleepFor(sleepTime) finish before pushing the event.
+	pollUntilRunStatus(t, ctx, sharedClient, ref.RunId, string(rest.V1TaskStatusRUNNING))
+	time.Sleep(time.Duration(sleepTime+3) * time.Second)
 	err = sharedClient.Events().Push(ctx, eventKey, map[string]string{"test": "test"})
 	require.NoError(t, err)
 
@@ -173,7 +184,7 @@ func TestDurableSleepEventSpawnReplay(t *testing.T) {
 	replayChild, ok := rm["child_output"].(map[string]any)
 	require.True(t, ok)
 	assert.Equal(t, "hello from child 1", replayChild["message"])
-	assert.Less(t, replayElapsed, float64(sleepTime))
+	assert.Less(t, replayElapsed, float64(sleepTime)+timingTolerance)
 }
 
 func TestDurableCompletedReplay(t *testing.T) {
@@ -207,8 +218,8 @@ func TestDurableCompletedReplay(t *testing.T) {
 	var replayOutput map[string]float64
 	err = replayResult.TaskOutput("wait-for-sleep-twice").Into(&replayOutput)
 	require.NoError(t, err)
-	assert.Less(t, replayOutput["runtime"], float64(sleepTime))
-	assert.Less(t, elapsed, float64(sleepTime))
+	assert.Less(t, replayOutput["runtime"], float64(sleepTime)+timingTolerance)
+	assert.Less(t, elapsed, float64(sleepTime)+timingTolerance)
 }
 
 func TestDurableSpawnDAG(t *testing.T) {
@@ -306,7 +317,7 @@ func TestDurableReplayReset(t *testing.T) {
 			durations := []float64{resetOutput.Sleep1Duration, resetOutput.Sleep2Duration, resetOutput.Sleep3Duration}
 			for i, d := range durations {
 				if int64(i+1) < nodeID {
-					assert.Less(t, d, float64(replayResetSleepTime))
+					assert.Less(t, d, float64(replayResetSleepTime)+timingTolerance)
 				} else {
 					assert.GreaterOrEqual(t, d, float64(replayResetSleepTime))
 				}
@@ -369,7 +380,7 @@ func TestDurableBranchingOffBranch(t *testing.T) {
 	err = resetResult2.TaskOutput("durable-replay-reset").Into(&resetOutput2)
 	require.NoError(t, err)
 
-	assert.Less(t, resetOutput2.Sleep1Duration, float64(replayResetSleepTime))
+	assert.Less(t, resetOutput2.Sleep1Duration, float64(replayResetSleepTime)+timingTolerance)
 	assert.GreaterOrEqual(t, resetOutput2.Sleep2Duration, float64(replayResetSleepTime))
 	assert.GreaterOrEqual(t, resetOutput2.Sleep3Duration, float64(replayResetSleepTime))
 	assert.GreaterOrEqual(t, resetElapsed2, float64(2*replayResetSleepTime))
@@ -407,7 +418,7 @@ func TestDurableMemoizationViaReplay(t *testing.T) {
 	require.NoError(t, err)
 
 	assert.GreaterOrEqual(t, duration1, float64(sleepTime))
-	assert.Less(t, duration2, 1.0)
+	assert.Less(t, duration2, 1.1)
 	assert.Equal(t, output1.Message, output2.Message)
 }
 

sdks/go/e2e/e2e_test.go

@@ -16,9 +16,9 @@ import (
 )
 
 const (
-	defaultTimeout = 5 * time.Minute
-	pollInterval   = 200 * time.Millisecond
-	maxPolls       = 150
+	defaultTimeout  = 5 * time.Minute
+	pollInterval    = 200 * time.Millisecond
+	timingTolerance = 1.0 // seconds of slack for scheduling/measurement overhead in timing assertions
 )
 
 var (
@@ -62,13 +62,13 @@ func uniqueID() string {
 	return uuid.New().String()
 }
 
-// pollUntil polls fn every pollInterval until it returns true or maxPolls is reached.
+// pollUntil polls fn every pollInterval until it returns true or the context is done.
 func pollUntil(t *testing.T, ctx context.Context, fn func() (bool, error)) {
 	t.Helper()
-	for i := 0; i < maxPolls; i++ {
+	for {
 		done, err := fn()
 		if err != nil {
-			t.Logf("poll error (attempt %d): %v", i, err)
+			t.Logf("poll error: %v", err)
 		}
 		if done {
 			return
@@ -79,7 +79,6 @@ func pollUntil(t *testing.T, ctx context.Context, fn func() (bool, error)) {
 		case <-time.After(pollInterval):
 		}
 	}
-	t.Fatalf("polling timed out after %d attempts", maxPolls)
 }
 
 // pollUntilRunStatus polls run details until any task reaches the given status.