refactor: enhance WAITFOR EVENT handling with result propagation, improve iteration and recovery logic, and add timeout-aware event filtering

Author: ziflex

pkg/compiler/internal/wait_event.go

@@ -34,7 +34,7 @@ func (c *WaitCompiler) compileEvent(ctx fql.IWaitForEventExpressionContext) byte
 	end := c.ctx.Program.Emitter.NewLabel()
 
 	c.ctx.Program.Emitter.MarkLabel(start)
-	c.emitWaitEventIteration(ctx, state, streamReg, bytecode.NoopOperand, start, end)
+	c.emitWaitEventIteration(ctx, state, streamReg, resultReg, bytecode.NoopOperand, start, end)
 
 	c.ctx.Program.Emitter.MarkLabel(end)
 	c.emitWaitEventCleanup(state, streamReg)
@@ -65,7 +65,7 @@ func (c *WaitCompiler) compileEventWithTimeoutRecovery(
 	cleanup := c.ctx.Program.Emitter.NewLabel()
 
 	c.ctx.Program.Emitter.MarkLabel(start)
-	c.emitWaitEventIteration(ctx, state, streamReg, timeoutStateReg, start, iterationDone)
+	c.emitWaitEventIteration(ctx, state, streamReg, resultReg, timeoutStateReg, start, iterationDone)
 
 	c.ctx.Program.Emitter.EmitJump(cleanup)
 	c.ctx.Program.Emitter.MarkLabel(iterationDone)
@@ -112,7 +112,7 @@ func (c *WaitCompiler) emitWaitEventStreamSetup(state waitEventCompileState, str
 func (c *WaitCompiler) emitWaitEventIteration(
 	ctx fql.IWaitForEventExpressionContext,
 	state waitEventCompileState,
-	streamReg, timeoutStateReg bytecode.Operand,
+	streamReg, resultReg, timeoutStateReg bytecode.Operand,
 	restartLabel, doneLabel core.Label,
 ) {
 	c.ctx.Program.Emitter.WithSpan(state.span, func() {
@@ -124,16 +124,26 @@ func (c *WaitCompiler) emitWaitEventIteration(
 		c.ctx.Program.Emitter.EmitIterNext(streamReg, doneLabel)
 	})
 
-	if filter := ctx.EventFilterClause(); filter != nil {
-		eventValReg, _ := c.ctx.Function.Symbols.DeclareLocal(core.PseudoVariable, core.TypeUnknown)
-
+	filter := ctx.EventFilterClause()
+	if filter == nil {
 		c.ctx.Program.Emitter.WithSpan(state.span, func() {
-			c.ctx.Program.Emitter.EmitAB(bytecode.OpIterValue, eventValReg, streamReg)
+			c.ctx.Program.Emitter.EmitIterValue(resultReg, streamReg)
 		})
-
-		cond := c.exprs.CompileWithImplicitCurrent(filter.Expression())
-		c.ctx.Program.Emitter.EmitJumpIfFalse(cond, restartLabel)
+		return
 	}
+
+	eventValReg, _ := c.ctx.Function.Symbols.DeclareLocal(core.PseudoVariable, core.TypeUnknown)
+
+	c.ctx.Program.Emitter.WithSpan(state.span, func() {
+		c.ctx.Program.Emitter.EmitIterValue(eventValReg, streamReg)
+	})
+
+	cond := c.exprs.CompileWithImplicitCurrent(filter.Expression())
+	c.ctx.Program.Emitter.EmitJumpIfFalse(cond, restartLabel)
+
+	c.ctx.Program.Emitter.WithSpan(state.span, func() {
+		c.ctx.Program.Emitter.EmitMove(resultReg, eventValReg)
+	})
 }
 
 func (c *WaitCompiler) emitWaitEventCleanup(state waitEventCompileState, streamReg bytecode.Operand) {

pkg/compiler/internal/wait_recovery.go

@@ -152,7 +152,7 @@ func (c *WaitCompiler) buildProtectedEventRecovery(
 	routeRecovery := c.ctx.Program.Emitter.NewLabel("waitfor", "event", "recover")
 
 	c.ctx.Program.Emitter.MarkLabel(start)
-	c.emitWaitEventIteration(ctx, state, streamReg, timeoutStateReg, start, iterationDone)
+	c.emitWaitEventIteration(ctx, state, streamReg, resultReg, timeoutStateReg, start, iterationDone)
 
 	c.ctx.Program.Emitter.EmitJump(cleanup)
 	c.ctx.Program.Emitter.MarkLabel(iterationDone)

test/integration/vm/vm_waitfor_event_test.go

@@ -42,6 +42,13 @@ RETURN status`, "error", "WAITFOR EVENT should choose ON ERROR for runtime failu
 LET status = (WAITFOR EVENT "test" IN obj TIMEOUT 1ms) ON TIMEOUT RETURN "timeout" ON ERROR RETURN "error"
 
 RETURN status`, "error", "Grouped WAITFOR EVENT should choose ON ERROR for runtime failures"),
+		S(`LET obs = @obs
+
+LET evt = WAITFOR EVENT "test" IN obs
+
+RETURN evt.type`, "match", "WAITFOR EVENT should return the received event value").Env(vm.WithParams(map[string]runtime.Value{
+			"obs": matchFirst,
+		})),
 		Fn(`LET obs = @obs
 WAITFOR EVENT "test" IN obs WHEN .type == "match"
 RETURN 1`, ObservableReturnOneAndReads(matchFirst, 1)).Env(vm.WithParams(map[string]runtime.Value{
@@ -54,6 +61,27 @@ RETURN 1`, ObservableReturnOneAndReads(matchSecond, 2)).Env(vm.WithParams(map[st
 		})),
 		S(`LET obs = @obs
 
+LET evt = WAITFOR EVENT "test" IN obs WHEN .type == "match"
+
+RETURN evt.type`, "match", "WAITFOR EVENT filter should return the matched event value").Env(vm.WithParams(map[string]runtime.Value{
+			"obs": matchSecond,
+		})),
+		S(`LET obs = @obs
+
+LET evt = WAITFOR EVENT "test" IN obs TIMEOUT 1ms ON TIMEOUT RETURN NONE
+
+RETURN evt.type`, "match", "WAITFOR EVENT timeout-aware success should return the event value").Env(vm.WithParams(map[string]runtime.Value{
+			"obs": matchFirst,
+		})),
+		S(`LET obs = @obs
+
+LET evt = WAITFOR EVENT "test" IN obs ON ERROR RETURN NONE
+
+RETURN evt.type`, "match", "WAITFOR EVENT protected recovery success should return the event value").Env(vm.WithParams(map[string]runtime.Value{
+			"obs": matchFirst,
+		})),
+		S(`LET obs = @obs
+
 LET status = WAITFOR EVENT "test" IN obs TIMEOUT 1ms ON TIMEOUT RETURN "timeout" ON ERROR RETURN "error"
 
 RETURN status`, "timeout", "WAITFOR EVENT should choose ON TIMEOUT when the stream times out").Env(vm.WithParams(map[string]runtime.Value{