fix(wait_event): drain signals with a blocking Receive loop (EN-1223)

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)
+	}
 }

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",
+		},
 	}...)
 }