diff --git a/.golangci.yml b/.golangci.yml index cdbeddac..bb977a28 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -7,7 +7,6 @@ linters: enable: - bidichk - bodyclose - - errorlint - goprintffuncname - govet - importas @@ -22,7 +21,6 @@ linters: - testifylint - tparallel - unconvert - - usetesting - wastedassign - whitespace - unused @@ -30,8 +28,6 @@ linters: disable: - errname - errcheck - - errorlint - - usetesting settings: staticcheck: checks: diff --git a/internal/workflowstate/workflowstate.go b/internal/workflowstate/workflowstate.go index 6311e5db..c310e26f 100644 --- a/internal/workflowstate/workflowstate.go +++ b/internal/workflowstate/workflowstate.go @@ -65,8 +65,9 @@ type WfState struct { logger *slog.Logger tracer trace.Tracer - clock clock.Clock - time time.Time + clock clock.Clock + time time.Time + historyLength int64 } func NewWorkflowState(instance *core.WorkflowInstance, logger *slog.Logger, tracer trace.Tracer, clock clock.Clock) *WfState { @@ -174,3 +175,11 @@ func (wf *WfState) Logger() *slog.Logger { func (wf *WfState) Tracer() trace.Tracer { return wf.tracer } + +func (wf *WfState) SetHistoryLength(length int64) { + wf.historyLength = length +} + +func (wf *WfState) HistoryLength() int64 { + return wf.historyLength +} diff --git a/samples/workflow-info/README.md b/samples/workflow-info/README.md new file mode 100644 index 00000000..28f617e5 --- /dev/null +++ b/samples/workflow-info/README.md @@ -0,0 +1,57 @@ +# Workflow Info Sample + +This sample demonstrates how to access workflow information during workflow execution, specifically the history length. + +## What it demonstrates + +- Using `workflow.InstanceExecutionDetails(ctx)` to access workflow metadata +- Tracking how the workflow history grows as events are added +- Accessing the `HistoryLength` field of `WorkflowInstanceExecutionDetails` + +## Running the sample + +```bash +go run . +``` + +## Expected Output + +You should see log messages showing the history length increasing as the workflow executes: + +``` +Workflow started historyLength=2 +Activity executed +After activity execution historyLength=5 +Activity executed +After second activity historyLength=8 +Workflow completed successfully! +``` + +## How it works + +The `WorkflowInstanceExecutionDetails` struct contains information about the current workflow execution. Currently it provides: + +- `HistoryLength`: The number of events in the workflow history at the current point in execution + +The history length includes all events that have been added to the workflow's event history, including: +- WorkflowExecutionStarted +- WorkflowTaskStarted +- ActivityScheduled +- ActivityCompleted +- TimerScheduled +- TimerFired +- And other workflow events + +This can be useful for: +- Monitoring workflow complexity +- Making decisions based on how far the workflow has progressed +- Implementing custom limits or checkpointing logic +- Debugging and understanding workflow execution + +## Future extensions + +The `WorkflowInstanceExecutionDetails` struct is designed to be extensible. Future additions might include: +- Execution duration +- Number of activities executed +- Number of retries +- Custom metadata diff --git a/samples/workflow-info/main.go b/samples/workflow-info/main.go new file mode 100644 index 00000000..a9ac22a2 --- /dev/null +++ b/samples/workflow-info/main.go @@ -0,0 +1,90 @@ +package main + +import ( + "context" + "fmt" + "log" + "time" + + "github.com/cschleiden/go-workflows/client" + "github.com/cschleiden/go-workflows/worker" + "github.com/cschleiden/go-workflows/workflow" + + "github.com/cschleiden/go-workflows/backend/sqlite" +) + +// Workflow demonstrates accessing workflow info including history length +func Workflow(ctx workflow.Context) error { + // Get workflow info at the start + info := workflow.InstanceExecutionDetails(ctx) + logger := workflow.Logger(ctx) + logger.Info("Workflow started", "historyLength", info.HistoryLength) + + // Execute an activity + _, err := workflow.ExecuteActivity[any](ctx, workflow.DefaultActivityOptions, Activity).Get(ctx) + if err != nil { + return err + } + + // Check history length again after activity + info = workflow.InstanceExecutionDetails(ctx) + logger.Info("After activity execution", "historyLength", info.HistoryLength) + + // Execute another activity + _, err = workflow.ExecuteActivity[any](ctx, workflow.DefaultActivityOptions, Activity).Get(ctx) + if err != nil { + return err + } + + // Check history length again + info = workflow.InstanceExecutionDetails(ctx) + logger.Info("After second activity", "historyLength", info.HistoryLength) + + return nil +} + +func Activity(ctx context.Context) error { + log.Println("Activity executed") + return nil +} + +func main() { + ctx := context.Background() + + // Create in-memory SQLite backend + b := sqlite.NewInMemoryBackend() + + // Create worker + w := worker.New(b, nil) + + // Register workflow and activity + w.RegisterWorkflow(Workflow) + w.RegisterActivity(Activity) + + // Start worker + if err := w.Start(ctx); err != nil { + panic(err) + } + + // Create client + c := client.New(b) + + // Create workflow instance + wfi, err := c.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{ + InstanceID: "workflow-info-demo", + }, Workflow) + if err != nil { + panic(err) + } + + fmt.Println("Created workflow instance:", wfi.InstanceID) + + // Wait for result (10 second timeout) + err = c.WaitForWorkflowInstance(ctx, wfi, 10*time.Second) + if err != nil { + panic(err) + } + + fmt.Println("Workflow completed successfully!") + fmt.Println("Check the logs above to see how the history length increased as the workflow executed.") +} diff --git a/tester/tester_workflowinfo_test.go b/tester/tester_workflowinfo_test.go new file mode 100644 index 00000000..7d2c9581 --- /dev/null +++ b/tester/tester_workflowinfo_test.go @@ -0,0 +1,95 @@ +package tester + +import ( + "context" + "testing" + + "github.com/cschleiden/go-workflows/workflow" + "github.com/stretchr/testify/require" +) + +func Test_InstanceExecutionDetails_HistoryLength(t *testing.T) { + var capturedLength int64 + + workflowWithInfo := func(ctx workflow.Context) error { + info := workflow.InstanceExecutionDetails(ctx) + capturedLength = info.HistoryLength + return nil + } + + tester := NewWorkflowTester[any](workflowWithInfo) + tester.Execute(context.Background()) + + require.True(t, tester.WorkflowFinished()) + require.Positive(t, capturedLength, "History length should be greater than 0") +} + +func Test_InstanceExecutionDetails_HistoryLength_WithActivity(t *testing.T) { + var lengthBeforeActivity, lengthAfterActivity int64 + + workflowWithActivity := func(ctx workflow.Context) (int, error) { + info := workflow.InstanceExecutionDetails(ctx) + lengthBeforeActivity = info.HistoryLength + + r, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, activity1).Get(ctx) + if err != nil { + return 0, err + } + + info = workflow.InstanceExecutionDetails(ctx) + lengthAfterActivity = info.HistoryLength + + return r, nil + } + + tester := NewWorkflowTester[int](workflowWithActivity) + tester.Registry().RegisterActivity(activity1) + + tester.Execute(context.Background()) + + require.True(t, tester.WorkflowFinished()) + require.Positive(t, lengthBeforeActivity) + require.Greater(t, lengthAfterActivity, lengthBeforeActivity, "History length should increase after activity execution") + + r, err := tester.WorkflowResult() + require.NoError(t, err) + require.Equal(t, 23, r) +} + +func Test_InstanceExecutionDetails_HistoryLength_MultipleSteps(t *testing.T) { + var lengths []int64 + + workflowMultipleSteps := func(ctx workflow.Context) error { + info := workflow.InstanceExecutionDetails(ctx) + lengths = append(lengths, info.HistoryLength) + + workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, activity1).Get(ctx) + + info = workflow.InstanceExecutionDetails(ctx) + lengths = append(lengths, info.HistoryLength) + + workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, activity1).Get(ctx) + + info = workflow.InstanceExecutionDetails(ctx) + lengths = append(lengths, info.HistoryLength) + + return nil + } + + tester := NewWorkflowTester[any](workflowMultipleSteps) + tester.Registry().RegisterActivity(activity1) + + tester.Execute(context.Background()) + + require.True(t, tester.WorkflowFinished()) + + // The tester replays the workflow, so we'll see the lengths multiple times + // We just need to verify that the final three captures show increasing values + require.GreaterOrEqual(t, len(lengths), 3, "Should have at least 3 length captures") + + // Get the last 3 values (from the final execution) + finalLengths := lengths[len(lengths)-3:] + require.Positive(t, finalLengths[0]) + require.Greater(t, finalLengths[1], finalLengths[0], "History should grow after first activity") + require.Greater(t, finalLengths[2], finalLengths[1], "History should grow after second activity") +} diff --git a/workflow/executor/executor.go b/workflow/executor/executor.go index 37838ff4..c4232c33 100644 --- a/workflow/executor/executor.go +++ b/workflow/executor/executor.go @@ -209,6 +209,8 @@ func (e *executor) ExecuteTask(ctx context.Context, t *backend.WorkflowTask) (*E executedEvents[i].SequenceID = e.nextSequenceID() } + e.workflowState.SetHistoryLength(e.lastSequenceID) + logger.Debug("Finished workflow task", log.ExecutedEventsKey, len(executedEvents), log.TaskLastSequenceIDKey, e.lastSequenceID, @@ -269,10 +271,13 @@ func (e *executor) replayHistory(h []*history.Event) error { return errors.New("history has older events than current state") } + // Note: lastSequenceID is updated below after successful event execution. + // For consistent history length reporting (e.g., for workflow code), we intentionally set historyLength here before executing the event. + e.workflowState.SetHistoryLength(e.lastSequenceID + 1) + if err := e.executeEvent(event); err != nil { return err } - e.lastSequenceID = event.SequenceID } @@ -283,6 +288,9 @@ func (e *executor) executeNewEvents(newEvents []*history.Event) ([]*history.Even e.workflowState.SetReplaying(false) for i, event := range newEvents { + // Update history length BEFORE executing the event to reflect the event about to be added + e.workflowState.SetHistoryLength(e.lastSequenceID + int64(i) + 1) + if err := e.executeEvent(event); err != nil { return newEvents[:i], err } diff --git a/workflow/executor/historyinfo_test.go b/workflow/executor/historyinfo_test.go new file mode 100644 index 00000000..82764fb4 --- /dev/null +++ b/workflow/executor/historyinfo_test.go @@ -0,0 +1,365 @@ +package executor + +import ( + "context" + "testing" + "time" + + "github.com/cschleiden/go-workflows/backend" + "github.com/cschleiden/go-workflows/backend/converter" + "github.com/cschleiden/go-workflows/backend/history" + "github.com/cschleiden/go-workflows/backend/metadata" + "github.com/cschleiden/go-workflows/backend/payload" + "github.com/cschleiden/go-workflows/core" + "github.com/cschleiden/go-workflows/internal/fn" + "github.com/cschleiden/go-workflows/internal/sync" + "github.com/cschleiden/go-workflows/registry" + wf "github.com/cschleiden/go-workflows/workflow" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +func Test_InstanceExecutionDetails_SimpleWorkflow(t *testing.T) { + r := registry.New() + i := core.NewWorkflowInstance(uuid.NewString(), "") + hp := &testHistoryProvider{} + e, err := newExecutor(r, i, hp) + require.NoError(t, err) + defer e.Close() + + var capturedLength int64 + simpleWorkflow := func(ctx sync.Context) error { + info := wf.InstanceExecutionDetails(ctx) + capturedLength = info.HistoryLength + return nil + } + + r.RegisterWorkflow(simpleWorkflow) + + task := startWorkflowTask(i.InstanceID, simpleWorkflow) + result, err := e.ExecuteTask(context.Background(), task) + require.NoError(t, err) + + // The workflow code runs when WorkflowExecutionStarted is processed (event index 1) + // At that point: WorkflowTaskStarted (index 0, will be seq 1) and WorkflowExecutionStarted (index 1, will be seq 2) + // So history length should be 2 + require.Equal(t, int64(2), capturedLength) + require.Equal(t, int64(3), e.lastSequenceID) + require.Len(t, result.Executed, 3) // WorkflowTaskStarted, WorkflowExecutionStarted, WorkflowExecutionFinished +} + +func Test_InstanceExecutionDetails_WithActivity(t *testing.T) { + r := registry.New() + i := core.NewWorkflowInstance(uuid.NewString(), "") + hp := &testHistoryProvider{} + e, err := newExecutor(r, i, hp) + require.NoError(t, err) + defer e.Close() + + var lengthBeforeActivity, lengthAfterActivity int64 + + workflowWithActivity := func(ctx sync.Context) error { + info := wf.InstanceExecutionDetails(ctx) + lengthBeforeActivity = info.HistoryLength + + wf.ExecuteActivity[int](ctx, wf.DefaultActivityOptions, activity1, 42).Get(ctx) + + info = wf.InstanceExecutionDetails(ctx) + lengthAfterActivity = info.HistoryLength + + return nil + } + + r.RegisterWorkflow(workflowWithActivity) + r.RegisterActivity(activity1) + + inputs, _ := converter.DefaultConverter.To(42) + result, _ := converter.DefaultConverter.To(42) + + task := &backend.WorkflowTask{ + ID: "taskID", + WorkflowInstance: core.NewWorkflowInstance(i.InstanceID, "executionID"), + Metadata: &metadata.WorkflowMetadata{}, + NewEvents: []*history.Event{ + history.NewPendingEvent( + time.Now(), + history.EventType_WorkflowExecutionStarted, + &history.ExecutionStartedAttributes{ + Name: fn.Name(workflowWithActivity), + Inputs: []payload.Payload{}, + }, + ), + }, + } + + // Execute first task - schedules activity + task1Result, err := e.ExecuteTask(context.Background(), task) + require.NoError(t, err) + require.Equal(t, int64(2), lengthBeforeActivity) + + // Setup history for replay + hp.history = []*history.Event{ + history.NewHistoryEvent( + 1, + time.Now(), + history.EventType_WorkflowExecutionStarted, + &history.ExecutionStartedAttributes{ + Name: fn.Name(workflowWithActivity), + Inputs: []payload.Payload{}, + }, + ), + history.NewHistoryEvent( + 2, + time.Now(), + history.EventType_WorkflowTaskStarted, + &history.WorkflowTaskStartedAttributes{}, + ), + history.NewHistoryEvent( + 3, + time.Now(), + history.EventType_ActivityScheduled, + &history.ActivityScheduledAttributes{ + Name: "activity1", + Inputs: []payload.Payload{inputs}, + }, + history.ScheduleEventID(1), + ), + } + + // Execute second task with activity completion + task2 := &backend.WorkflowTask{ + ID: "taskID2", + WorkflowInstance: core.NewWorkflowInstance(i.InstanceID, "executionID"), + Metadata: &metadata.WorkflowMetadata{}, + NewEvents: []*history.Event{ + history.NewPendingEvent( + time.Now(), + history.EventType_ActivityCompleted, + &history.ActivityCompletedAttributes{ + Result: result, + }, + history.ScheduleEventID(1), + ), + }, + LastSequenceID: task1Result.Executed[len(task1Result.Executed)-1].SequenceID, + } + + _, err = e.ExecuteTask(context.Background(), task2) + require.NoError(t, err) + + // History should have grown: WorkflowExecutionStarted, WorkflowTaskStarted, ActivityScheduled, WorkflowTaskStarted, ActivityCompleted + require.Greater(t, lengthAfterActivity, lengthBeforeActivity) + require.Equal(t, int64(5), lengthAfterActivity) +} + +func Test_InstanceExecutionDetails_DuringReplay(t *testing.T) { + r := registry.New() + i := core.NewWorkflowInstance(uuid.NewString(), "") + hp := &testHistoryProvider{} + e, err := newExecutor(r, i, hp) + require.NoError(t, err) + defer e.Close() + + var lengths []int64 + + workflowWithActivity := func(ctx sync.Context) error { + // Capture length at start + info := wf.InstanceExecutionDetails(ctx) + lengths = append(lengths, info.HistoryLength) + + wf.ExecuteActivity[int](ctx, wf.DefaultActivityOptions, activity1, 42).Get(ctx) + + // Capture length after activity + info = wf.InstanceExecutionDetails(ctx) + lengths = append(lengths, info.HistoryLength) + + return nil + } + + r.RegisterWorkflow(workflowWithActivity) + r.RegisterActivity(activity1) + + inputs, _ := converter.DefaultConverter.To(42) + result, _ := converter.DefaultConverter.To(42) + + // Setup complete history for replay + hp.history = []*history.Event{ + history.NewHistoryEvent( + 1, + time.Now(), + history.EventType_WorkflowExecutionStarted, + &history.ExecutionStartedAttributes{ + Name: fn.Name(workflowWithActivity), + Inputs: []payload.Payload{}, + }, + ), + history.NewHistoryEvent( + 2, + time.Now(), + history.EventType_WorkflowTaskStarted, + &history.WorkflowTaskStartedAttributes{}, + ), + history.NewHistoryEvent( + 3, + time.Now(), + history.EventType_ActivityScheduled, + &history.ActivityScheduledAttributes{ + Name: "activity1", + Inputs: []payload.Payload{inputs}, + }, + history.ScheduleEventID(1), + ), + history.NewHistoryEvent( + 4, + time.Now(), + history.EventType_WorkflowTaskStarted, + &history.WorkflowTaskStartedAttributes{}, + ), + history.NewHistoryEvent( + 5, + time.Now(), + history.EventType_ActivityCompleted, + &history.ActivityCompletedAttributes{ + Result: result, + }, + history.ScheduleEventID(1), + ), + } + + task := &backend.WorkflowTask{ + ID: "taskID", + WorkflowInstance: core.NewWorkflowInstance(i.InstanceID, "executionID"), + Metadata: &metadata.WorkflowMetadata{}, + LastSequenceID: 5, + } + + _, err = e.ExecuteTask(context.Background(), task) + require.NoError(t, err) + + // During replay, the workflow replays up to sequenceID 5 + // First time workflow code runs: after replaying event 1 (WorkflowExecutionStarted), history length = 1 + // Second time: after replaying event 5 (ActivityCompleted), then new WorkflowTaskStarted is added + require.Len(t, lengths, 2) + require.Equal(t, int64(1), lengths[0]) // After replaying WorkflowExecutionStarted + require.Equal(t, int64(5), lengths[1]) // After replaying all events including ActivityCompleted +} + +func Test_InstanceExecutionDetails_Incremental(t *testing.T) { + r := registry.New() + i := core.NewWorkflowInstance(uuid.NewString(), "") + hp := &testHistoryProvider{} + e, err := newExecutor(r, i, hp) + require.NoError(t, err) + defer e.Close() + + var lengths []int64 + + workflowMultipleSteps := func(ctx sync.Context) error { + info := wf.InstanceExecutionDetails(ctx) + lengths = append(lengths, info.HistoryLength) + + wf.Sleep(ctx, time.Millisecond) + + info = wf.InstanceExecutionDetails(ctx) + lengths = append(lengths, info.HistoryLength) + + wf.Sleep(ctx, time.Millisecond) + + info = wf.InstanceExecutionDetails(ctx) + lengths = append(lengths, info.HistoryLength) + + return nil + } + + r.RegisterWorkflow(workflowMultipleSteps) + + // First execution - start workflow, schedule first timer + task1 := startWorkflowTask(i.InstanceID, workflowMultipleSteps) + result1, err := e.ExecuteTask(context.Background(), task1) + require.NoError(t, err) + + // Second execution - first timer fires, schedule second timer + hp.history = append(hp.history, result1.Executed...) + task2 := continueTask(i.InstanceID, []*history.Event{ + history.NewPendingEvent(time.Now(), history.EventType_TimerFired, &history.TimerFiredAttributes{ + ScheduledAt: time.Now(), + At: time.Now().Add(time.Millisecond), + }, history.ScheduleEventID(1)), + }, result1.Executed[len(result1.Executed)-1].SequenceID) + result2, err := e.ExecuteTask(context.Background(), task2) + require.NoError(t, err) + + // Third execution - second timer fires, complete workflow + hp.history = append(hp.history, result2.Executed...) + task3 := continueTask(i.InstanceID, []*history.Event{ + history.NewPendingEvent(time.Now(), history.EventType_TimerFired, &history.TimerFiredAttributes{ + ScheduledAt: time.Now(), + At: time.Now().Add(time.Millisecond), + }, history.ScheduleEventID(2)), + }, result2.Executed[len(result2.Executed)-1].SequenceID) + _, err = e.ExecuteTask(context.Background(), task3) + require.NoError(t, err) + + // Verify history length increases at each step + require.Len(t, lengths, 3) + require.Equal(t, int64(2), lengths[0]) // Initial: WorkflowTaskStarted(1) + WorkflowExecutionStarted(2) + // After first timer fires: previous 3 events + WorkflowTaskStarted(4) + TimerFired(5) = 5, + // but we see it when WorkflowTaskStarted is being processed, so we see 4 + upcoming event = 5 + require.Greater(t, lengths[1], lengths[0], "History should grow after first timer") + require.Greater(t, lengths[2], lengths[1], "History should grow after second timer") +} + +func Test_InstanceExecutionDetails_WithSubworkflow(t *testing.T) { + r := registry.New() + i := core.NewWorkflowInstance(uuid.NewString(), "") + hp := &testHistoryProvider{} + e, err := newExecutor(r, i, hp) + require.NoError(t, err) + defer e.Close() + + var lengthBeforeSub, lengthAfterSub int64 + + subworkflow := func(ctx wf.Context) error { + return nil + } + + workflow := func(ctx wf.Context) error { + info := wf.InstanceExecutionDetails(ctx) + lengthBeforeSub = info.HistoryLength + + wf.CreateSubWorkflowInstance[any](ctx, wf.SubWorkflowOptions{ + InstanceID: "subworkflow", + }, subworkflow).Get(ctx) + + info = wf.InstanceExecutionDetails(ctx) + lengthAfterSub = info.HistoryLength + + return nil + } + + r.RegisterWorkflow(workflow) + r.RegisterWorkflow(subworkflow) + + task := startWorkflowTask(i.InstanceID, workflow) + + result, err := e.ExecuteTask(context.Background(), task) + require.NoError(t, err) + + // Should have scheduled the subworkflow + require.Equal(t, int64(2), lengthBeforeSub) + require.Greater(t, e.lastSequenceID, lengthBeforeSub) + + // Complete the subworkflow + hp.history = append(hp.history, result.Executed...) + swr, _ := converter.DefaultConverter.To(nil) + task2 := continueTask(i.InstanceID, []*history.Event{ + history.NewPendingEvent(time.Now(), history.EventType_SubWorkflowCompleted, &history.SubWorkflowCompletedAttributes{ + Result: swr, + }, history.ScheduleEventID(1)), + }, result.Executed[len(result.Executed)-1].SequenceID) + + _, err = e.ExecuteTask(context.Background(), task2) + require.NoError(t, err) + + require.Greater(t, lengthAfterSub, lengthBeforeSub) +} diff --git a/workflow/instance.go b/workflow/instance.go index 14b3994c..7070e734 100644 --- a/workflow/instance.go +++ b/workflow/instance.go @@ -9,3 +9,19 @@ func WorkflowInstance(ctx Context) *Instance { wfState := workflowstate.WorkflowState(ctx) return wfState.Instance() } + +// WorkflowInstanceExecutionDetails contains information about the current workflow execution. +type WorkflowInstanceExecutionDetails struct { + // HistoryLength is the number of events in the workflow history at the current point in execution. + // This value increases as the workflow executes and generates new events. + HistoryLength int64 +} + +// InstanceExecutionDetails returns information about the current workflow execution. +func InstanceExecutionDetails(ctx Context) WorkflowInstanceExecutionDetails { + wfState := workflowstate.WorkflowState(ctx) + + return WorkflowInstanceExecutionDetails{ + HistoryLength: wfState.HistoryLength(), + } +}