workflows/wasm/host: classify epoch interrupts as timeouts

pkg/workflows/wasm/host/epoch_timeout_test.go

@@ -0,0 +1,85 @@
+package host
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/bytecodealliance/wasmtime-go/v47"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/workflows/host/mocks"
+	sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk"
+)
+
+func TestModuleEpochDeadline(t *testing.T) {
+	// Manual ticks reproduce an interrupt before the wall-clock deadline without
+	// depending on scheduler timing. This test must not run in parallel.
+	engine := newEngine(logger.Test(t))
+	globalEngineMu.Lock()
+	previous := globalEngine
+	globalEngine = engine
+	globalEngineMu.Unlock()
+	t.Cleanup(func() {
+		globalEngineMu.Lock()
+		globalEngine = previous
+		globalEngineMu.Unlock()
+		engine.Close()
+	})
+	for range 17 {
+		engine.IncrementEpoch()
+	}
+
+	for _, tc := range []struct {
+		name    string
+		timeout time.Duration
+		ticks   int
+		body    string
+		want    string
+	}{
+		{"fractional budget before deadline", 999 * time.Millisecond, 9, "", "success"},
+		{"fractional budget at deadline", 999 * time.Millisecond, 10, "", "timeout"},
+		{"whole ticks before deadline", time.Second, 9, "", "success"},
+		{"whole ticks at deadline", time.Second, 10, "", "timeout"},
+		{"sub-tick budget before deadline", 50 * time.Millisecond, 0, "", "success"},
+		{"sub-tick budget at deadline", 50 * time.Millisecond, 1, "", "timeout"},
+		{"unrelated trap", 50 * time.Millisecond, 0, "unreachable", "unreachable"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			raw, err := wasmtime.Wat2Wasm(`(module
+  (import "env" "version_v2" (func))
+  (memory (export "memory") 1)
+  (func (export "_start") ` + tc.body + `))`)
+			require.NoError(t, err)
+			mod, err := NewModule(t.Context(), &ModuleConfig{
+				Logger: logger.Test(t), IsUncompressed: true, Timeout: &tc.timeout,
+			}, raw)
+			require.NoError(t, err)
+			defer mod.Close()
+			helper := mocks.NewMockExecutionHelper(t)
+			helper.EXPECT().GetWorkflowExecutionID().RunAndReturn(func() string {
+				// The store's relative deadline is armed before this callback.
+				for range tc.ticks {
+					engine.IncrementEpoch()
+				}
+				return "id"
+			})
+			result, err := mod.Execute(t.Context(), &sdkpb.ExecuteRequest{}, helper)
+			require.Nil(t, result)
+			switch tc.want {
+			case "success":
+				require.NoError(t, err)
+			case "timeout":
+				require.ErrorIs(t, err, context.DeadlineExceeded)
+			case "unreachable":
+				var trap *wasmtime.Trap
+				require.ErrorAs(t, err, &trap)
+				require.NotNil(t, trap.Code())
+				require.Equal(t, wasmtime.UnreachableCodeReached, *trap.Code())
+			default:
+				t.Fatalf("unknown expected outcome %q", tc.want)
+			}
+		})
+	}
+}

pkg/workflows/wasm/host/module.go

@@ -626,6 +626,9 @@ func runWasm[I, O proto.Message](
 	)
 
 	deadline := maxTimeout / eng.engineTickInterval
+	if maxTimeout%eng.engineTickInterval > 0 {
+		deadline++
+	}
 	store.SetEpochDeadline(uint64(deadline))
 
 	h := fnv.New64a()
@@ -653,9 +656,7 @@ func runWasm[I, O proto.Message](
 		return o, fmt.Errorf("error linking wasm: %w", err)
 	}
 
-	startTime := time.Now()
 	_, err = callStart(m, instance, store)
-	executionDuration := time.Since(startTime)
 
 	// A guest that ran to completion exits with WASI status wasiExitSuccess
 	// after delivering its response via the send_response import.
@@ -666,11 +667,14 @@ func runWasm[I, O proto.Message](
 		return exec.response, nil
 	}
 
-	// If an error has occurred and the deadline has been reached or exceeded, return a deadline exceeded error.
-	// Note - there is no other reliable signal on the error that can be used to infer it is due to epoch deadline
-	// being reached, so if an error is returned after the deadline it is assumed it is due to that and return
-	// context.DeadlineExceeded.
-	if err != nil && ((executionDuration >= maxTimeout-engineTickInterval) || ctx.Err() != nil) { // As start could be called just before epoch update 1 tick interval is deducted to account for this
+	// Epoch interrupts can precede the wall-clock deadline because the ticker is shared.
+	var trap *wasmtime.Trap
+	epochExpired := false
+	if errors.As(err, &trap) {
+		code := trap.Code()
+		epochExpired = code != nil && *code == wasmtime.Interrupt
+	}
+	if err != nil && (epochExpired || ctxWithTimeout.Err() != nil) {
 		m.cfg.Logger.Errorw("start function returned error after deadline reached, returning deadline exceeded error", "errFromStartFunction", err)
 		return o, context.DeadlineExceeded
 	}