diff --git a/boot/components/runtime/lua/all.go b/boot/components/runtime/lua/all.go index 19546bc0f..69721c003 100644 --- a/boot/components/runtime/lua/all.go +++ b/boot/components/runtime/lua/all.go @@ -33,6 +33,7 @@ func All() []boot.Component { Payload(), Queue(), Registry(), + Replayer(), Security(), SQL(), Store(), diff --git a/boot/components/runtime/lua/constants.go b/boot/components/runtime/lua/constants.go index ec9d64a8f..e81d8903e 100644 --- a/boot/components/runtime/lua/constants.go +++ b/boot/components/runtime/lua/constants.go @@ -27,6 +27,7 @@ const ( MetricsName = "lua.metrics" JSONName = "lua.json" PayloadName = "lua.payload" + ReplayerName = "lua.replayer" RegistryName = "lua.registry" SecurityName = "lua.security" SQLName = "lua.sql" diff --git a/boot/components/runtime/lua/replayer.go b/boot/components/runtime/lua/replayer.go new file mode 100644 index 000000000..a78479da1 --- /dev/null +++ b/boot/components/runtime/lua/replayer.go @@ -0,0 +1,31 @@ +// SPDX-License-Identifier: MPL-2.0 + +package lua + +import ( + "context" + + "github.com/wippyai/runtime/api/boot" + "github.com/wippyai/runtime/runtime/lua/modules/replayer" +) + +func Replayer() boot.Component { + return boot.New(boot.P{ + Name: ReplayerName, + DependsOn: []boot.Name{EngineName}, + Load: func(ctx context.Context) (context.Context, error) { + cm := GetCodeManager(ctx) + if cm == nil { + return ctx, nil + } + + if err := AddModules(ctx, cm, + replayer.Module, + ); err != nil { + return ctx, err + } + + return ctx, nil + }, + }) +} diff --git a/runtime/lua/modules/replayer/module.go b/runtime/lua/modules/replayer/module.go new file mode 100644 index 000000000..55a7f6578 --- /dev/null +++ b/runtime/lua/modules/replayer/module.go @@ -0,0 +1,102 @@ +// SPDX-License-Identifier: MPL-2.0 + +// Package replayer exposes Temporal's workflow history replayer to Lua. +package replayer + +import ( + lua "github.com/wippyai/go-lua" + "github.com/wippyai/runtime/api/registry" + luaapi "github.com/wippyai/runtime/api/runtime/lua" + temporalapi "github.com/wippyai/runtime/api/service/temporal" + tpropagator "github.com/wippyai/runtime/service/temporal/propagator" + tworkflow "github.com/wippyai/runtime/service/temporal/workflow" + "go.temporal.io/sdk/worker" + sdkworkflow "go.temporal.io/sdk/workflow" +) + +// Module is the replayer Lua module (ClassIO: usable from tests, not workflow bodies). +var Module = &luaapi.ModuleDef{ + Name: "replayer", + Description: "Temporal workflow history replay for determinism tests", + Class: []string{luaapi.ClassIO, luaapi.ClassNondeterministic}, + Build: buildModule, + Types: ModuleTypes, +} + +func buildModule() (*lua.LTable, []luaapi.YieldType) { + mod := lua.CreateTable(0, 1) + mod.RawSetString("replay_json_file", lua.LGoFunc(replayJSONFile)) + mod.Immutable = true + return mod, nil +} + +func invalidError(l *lua.LState, msg string) int { + err := lua.NewLuaError(l, msg).WithKind(lua.Invalid).WithRetryable(false) + l.Push(lua.LNil) + l.Push(err) + return 2 +} + +func internalError(l *lua.LState, goErr error, context string) int { + err := lua.WrapErrorWithLua(l, goErr, context).WithKind(lua.Internal).WithRetryable(false) + l.Push(lua.LNil) + l.Push(err) + return 2 +} + +// replayJSONFile(workflow_id, history_json_path[, workflow_type_name]) -> (true) | (nil, err); +// nil err == deterministic replay. workflow_type_name overrides the registered type name +// when the workflow uses a custom meta name (defaults to workflow_id). +func replayJSONFile(l *lua.LState) int { + workflowID := l.CheckString(1) + path := l.CheckString(2) + if workflowID == "" { + return invalidError(l, "workflow id required") + } + if path == "" { + return invalidError(l, "history file path required") + } + + regID := registry.ParseID(workflowID) + if regID.NS == "" || regID.Name == "" { + return invalidError(l, "workflow id must be in 'namespace:name' form") + } + + typeName := workflowID + if tn := l.OptString(3, ""); tn != "" { + typeName = tn + } + + ctx := l.Context() + + dcReg := temporalapi.GetDataConverterRegistry(ctx) + if dcReg == nil { + return invalidError(l, "temporal data converter registry not available in context") + } + dc := dcReg.Build() + + opts := worker.WorkflowReplayerOptions{ + DataConverter: dc, + ContextPropagators: []sdkworkflow.ContextPropagator{tpropagator.New(dc)}, + } + if wreg := temporalapi.GetWorkerInterceptorRegistry(ctx); wreg != nil { + opts.Interceptors = wreg.GetAll() + } + + r, err := worker.NewWorkflowReplayerWithOptions(opts) + if err != nil { + return internalError(l, err, "create workflow replayer") + } + + // Register the runtime's dynamic definition (SDK accepts it as a WorkflowDefinitionFactory). + factory := (&tworkflow.DefinitionFactory{ID: regID}).WithContext(ctx) + r.RegisterWorkflowWithOptions(factory, sdkworkflow.RegisterOptions{Name: typeName}) + + if err := r.ReplayWorkflowHistoryFromJSONFile(nil, path); err != nil { + return internalError(l, err, "replay") + } + + l.Push(lua.LBool(true)) + l.Push(lua.LNil) + return 2 +} diff --git a/runtime/lua/modules/replayer/module_test.go b/runtime/lua/modules/replayer/module_test.go new file mode 100644 index 000000000..aa50ff1dc --- /dev/null +++ b/runtime/lua/modules/replayer/module_test.go @@ -0,0 +1,65 @@ +// SPDX-License-Identifier: MPL-2.0 + +package replayer + +import ( + "testing" + + lua "github.com/wippyai/go-lua" +) + +func newBound(t *testing.T) *lua.LState { + t.Helper() + l := lua.NewState() + tbl, _ := Module.Build() + l.SetGlobal(Module.Name, tbl) + return l +} + +func TestBind(t *testing.T) { + l := newBound(t) + defer l.Close() + + mod := l.GetGlobal("replayer") + if mod.Type() != lua.LTTable { + t.Fatal("replayer module not registered") + } + if mod.(*lua.LTable).RawGetString("replay_json_file").Type() != lua.LTFunction { + t.Error("replay_json_file function not registered") + } +} + +// Argument validation runs before any context/Temporal access, so this needs no runtime. +func TestReplayBadArgs(t *testing.T) { + l := newBound(t) + defer l.Close() + + err := l.DoString(` + local _, err = replayer.replay_json_file("", "history.json") + if err == nil then error("expected error for empty workflow id") end + + _, err = replayer.replay_json_file("no-colon-here", "history.json") + if err == nil then error("expected error for non ns:name id") end + + _, err = replayer.replay_json_file("app:wf", "") + if err == nil then error("expected error for empty path") end + `) + if err != nil { + t.Errorf("bad-args test failed: %v", err) + } +} + +// Well-formed args with no Temporal in context must fail gracefully, not panic. +func TestReplayNoTemporalContext(t *testing.T) { + l := newBound(t) + defer l.Close() + + err := l.DoString(` + local ok, err = replayer.replay_json_file("app.letter:scheduled_delivery_workflow", "history.json") + if ok ~= nil then error("expected nil result without temporal context") end + if err == nil then error("expected error without temporal context") end + `) + if err != nil { + t.Errorf("no-temporal-context test failed: %v", err) + } +} diff --git a/runtime/lua/modules/replayer/types.go b/runtime/lua/modules/replayer/types.go new file mode 100644 index 000000000..647b3a35f --- /dev/null +++ b/runtime/lua/modules/replayer/types.go @@ -0,0 +1,23 @@ +// SPDX-License-Identifier: MPL-2.0 + +package replayer + +import ( + "github.com/wippyai/go-lua/types/io" + "github.com/wippyai/go-lua/types/typ" +) + +// ModuleTypes returns the type manifest for the replayer module. +func ModuleTypes() *io.Manifest { + m := io.NewManifest("replayer") + + moduleType := typ.NewInterface("replayer", []typ.Method{ + {Name: "replay_json_file", Type: typ.Func(). + Param("workflow_id", typ.Any). + Param("history_json_path", typ.Any). + Returns(typ.Any, typ.NewOptional(typ.LuaError)).Build()}, + }) + + m.SetExport(moduleType) + return m +} diff --git a/service/temporal/workflow/replay_integration_test.go b/service/temporal/workflow/replay_integration_test.go new file mode 100644 index 000000000..99807187a --- /dev/null +++ b/service/temporal/workflow/replay_integration_test.go @@ -0,0 +1,68 @@ +// SPDX-License-Identifier: MPL-2.0 + +package workflow_test + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + "github.com/wippyai/runtime/api/registry" + "github.com/wippyai/runtime/service/temporal/dataconverter" + "github.com/wippyai/runtime/service/temporal/workflow" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + "go.temporal.io/api/temporalproto" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/worker" + sdkworkflow "go.temporal.io/sdk/workflow" +) + +// TestReplay_Integration runs a Lua workflow, then replays its history against the +// same DefinitionFactory. A nil error proves deterministic replay. +func TestReplay_Integration(t *testing.T) { + wfID := registry.NewID("test.workflow", "replay-hello") + f := newWorkflowTestFixture(t, workflowTestOpts{ + workflowID: wfID, + source: helloWorkflowSource, + taskQueue: "test-replay-queue", + }) + defer f.cleanup() + + run := f.startWorkflow(wfID.String(), map[string]any{"name": "Replay"}) + var result map[string]any + require.NoError(t, run.Get(f.ctx, &result)) + + hist := fetchHistory(t, f.ctx, f.temporalClient, run.GetID(), run.GetRunID()) + require.NotEmpty(t, hist.Events) + + dc := dataconverter.NewDataConverter(newTestTranscoder()) + replayer, err := worker.NewWorkflowReplayerWithOptions(worker.WorkflowReplayerOptions{DataConverter: dc}) + require.NoError(t, err) + + factory := (&workflow.DefinitionFactory{ID: wfID}).WithContext(f.ctx) + replayer.RegisterWorkflowWithOptions(factory, sdkworkflow.RegisterOptions{Name: wfID.String()}) + + require.NoError(t, replayer.ReplayWorkflowHistory(nil, hist)) + + // Cover the module's exact path: history written to a JSON file, replayed with a nil logger. + data, err := temporalproto.CustomJSONMarshalOptions{}.Marshal(hist) + require.NoError(t, err) + histFile := filepath.Join(t.TempDir(), "history.json") + require.NoError(t, os.WriteFile(histFile, data, 0o600)) + require.NoError(t, replayer.ReplayWorkflowHistoryFromJSONFile(nil, histFile)) +} + +func fetchHistory(t *testing.T, ctx context.Context, c client.Client, wfID, runID string) *historypb.History { + t.Helper() + it := c.GetWorkflowHistory(ctx, wfID, runID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT) + var events []*historypb.HistoryEvent + for it.HasNext() { + ev, err := it.Next() + require.NoError(t, err) + events = append(events, ev) + } + return &historypb.History{Events: events} +}