diff --git a/internal/workflow/stages/wait_event/run.go b/internal/workflow/stages/wait_event/run.go index fd42208..6f90327 100644 --- a/internal/workflow/stages/wait_event/run.go +++ b/internal/workflow/stages/wait_event/run.go @@ -7,16 +7,18 @@ import ( func RunWaitEvent(ctx workflow.Context, waitEvent WaitEvent) error { channel := workflow.GetSignalChannel(ctx, internalWorkflow.EventSignalName) - return workflow.Await(ctx, func() bool { + // Drain the signal channel one signal at a time until the expected event + // arrives. Using a blocking Receive loop (rather than ReceiveAsync inside + // an Await predicate) guarantees no buffered signal is consumed and + // dropped: an Await predicate is only evaluated once per workflow-task + // wakeup, so two signals delivered in the same task would leave the second + // one buffered with nothing left to re-wake the coroutine, blocking forever. + for { var signal internalWorkflow.Event - ok := channel.ReceiveAsync(&signal) - if !ok { - return false + channel.Receive(ctx, &signal) + if signal.Name == waitEvent.Event { + return nil } - if signal.Name != waitEvent.Event { - workflow.GetLogger(ctx).Debug("receive unexpected event", "event", signal.Name) - return false - } - return true - }) + workflow.GetLogger(ctx).Debug("received unexpected event, still waiting", "event", signal.Name) + } } diff --git a/internal/workflow/stages/wait_event/wait_event_test.go b/internal/workflow/stages/wait_event/wait_event_test.go index 7eed168..610a867 100644 --- a/internal/workflow/stages/wait_event/wait_event_test.go +++ b/internal/workflow/stages/wait_event/wait_event_test.go @@ -49,5 +49,29 @@ func TestWaitEvent(t *testing.T) { }}, Name: "nominal", }, + { + Stage: WaitEvent{ + Event: "test", + }, + DelayedCallbacks: []stagestesting.DelayedCallback{{ + Fn: func(environment *testsuite.TestWorkflowEnvironment) func() { + return func() { + // Two signals delivered in the same workflow task: a + // non-matching one followed by the matching one. The + // stage must consume the first, keep the second, and + // complete (the previous ReceiveAsync-in-Await + // implementation would drop the buffered match and hang). + environment.SignalWorkflow(workflow.EventSignalName, workflow.Event{ + Name: "other", + }) + environment.SignalWorkflow(workflow.EventSignalName, workflow.Event{ + Name: "test", + }) + } + }, + Duration: 100 * time.Millisecond, + }}, + Name: "ignores non-matching event delivered in the same task", + }, }...) }