This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new d1ac8dc3b7 fix(amber): pass unstamped boundary states through LoopEnd
instead of consuming them (#6913)
d1ac8dc3b7 is described below
commit d1ac8dc3b76d3248bebc5c289ece6aa8c84f0766
Author: Xinyuan Lin <[email protected]>
AuthorDate: Sun Aug 9 21:52:52 2026 -0700
fix(amber): pass unstamped boundary states through LoopEnd instead of
consuming them (#6913)
### What changes were proposed in this PR?
Follow-up to #6661, addressing [this review
comment](https://github.com/apache/texera/pull/6661#discussion_r3648708075):
a loop-body operator that emits its own boundary state
(`produce_state_on_start/finish` — a public API on both engine sides)
sends it with the "no loop" envelope (counter `0`, `loop_start_id ""`).
The LoopEnd matching branch treated **every** counter-0 frame as the
loop's own boundary state:
```
LoopStart ──(0, LS-id)──▶ stateful body op ──▶ LoopEnd
│ produce_state_on_finish
└──(0, "")────────▶ LoopEnd ← arrives AFTER
the loop state
```
Consuming the unstamped state (1) clobbers the captured back-jump id
with `""` → `no loop-back state URI configured for LoopStart ''`, and
(2) hands `run_update` a State with no `table` payload → `KeyError`.
Either way the loop breaks. The reviewer reasoned this for a stateful
JVM operator; it is language-independent (a Python UDF hits the
identical path — no JVM built-in currently overrides
`produceStateOnFinish`, so the Python UDF is also the practical repro).
**Fix (consumer-side).** A real loop state is always stamped — the
matching LoopStart stamps its own id on every iteration's output state —
so the LoopEnd runtime now keys on the stamp:
| Frame at LoopEnd (counter 0) | before | after |
|---|---|---|
| stamped (`loop_start_id` set) | consume + capture id | unchanged |
| unstamped (`""`) | consume → id clobber / `KeyError` | forward
downstream unchanged, skip the operator (default pass-through
semantics), captured id untouched |
**Keeping the loud failure.** Forwarding unstamped frames also swallows
the symptom of the bug class #6660/#6661 fixed: a hop that blanks the
envelope makes the loop's *own* state arrive unstamped, and forwarding
it leaves `_loop_table` `None` → `condition()` returns `False` → the
loop stops after one iteration and reports **success with a wrong row
count**. So a Loop End that forwarded an unstamped state and never took
a stamped one now raises:
```
Loop End received a loop-boundary state with no LoopStart stamp and never
received its own (stamped) loop state: the loop envelope was lost upstream,
so this loop would silently stop after one iteration
```
Three properties of where that check lives:
- **`_process_end_channel`, not `complete()`** — `complete()` runs after
`port_completed` has gone out for the input port and every output port,
and region completion is port-based, so a raise there would be reported
only once the coordinator already considers the region done.
- **Order-independent** — the body operator's state may arrive before or
after the loop's; `EndChannel` is `PORT_ALIGNMENT`, so the port is
drained by then.
- **Reads nothing out of the `State`** — deliberately not keyed on the
reserved `table` key, which #6971 removes from the loop state (and which
an ordinary body UDF may legitimately emit), so the guard cannot rot
green.
It is narrow on purpose — *forwarded an unstamped state* **and** *never
took a stamped one*. A Loop End completing without any matching state is
legal (`LoopEndOperator.eval_condition`'s `_loop_table` guard), so only
positive evidence of an unstamped boundary state counts.
Also documents why a Loop **Start** must do the opposite — MERGE an
unstamped counter-0 state rather than forward it: the back-edge writes
the next iteration's variables to the Loop Start's own input-port state
URI with that same "no loop" envelope (`State.to_tuple(0)`), so a Loop
Start cannot tell its own state from an upstream operator's, while a
Loop End can. The key-collision hazard that follows from the merge is
filed as #7248.
### Any related issues, documentation, discussions?
Follow-up to #6661 (review discussion r3648708075). Related engine
context: #6660. Follow-up filed: #7248.
### How was this PR tested?
- **Unit** (`test_main_loop.py`), all verified red before the
corresponding change:
- an unstamped counter-0 frame at a LoopEnd is forwarded with its
envelope unchanged, the operator is not invoked, the captured back-jump
id is not clobbered, and it does not count as taking the loop's own
state;
- a LoopEnd that only ever saw an unstamped state reports the error from
`_process_end_channel` and sends **no** `port_completed`, so the region
is held;
- the legitimate shape (body-operator state *plus* the loop's own
stamped state) stays silent in **both** arrival orders;
- an unstamped counter-0 frame at a Loop **Start** is merged, not
forwarded.
- **E2E** (`LoopIntegrationSpec`, CI-only): `TextInput → LoopStart →
stateful Python UDF → LoopEnd` where the UDF emits boundary state via
`produce_state_on_finish` — it crashes without this fix and completes
exactly 3 iterations with it.
- Full pyamber suite, `scalafmtCheckAll` + `scalafixAll --check` + full
test-compile + ruff format/check pass locally (Java 17).
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---------
Co-authored-by: Yicong Huang
<[email protected]>
---
amber/src/main/python/core/runnables/main_loop.py | 90 ++++++++++-
.../worker/promisehandlers/EndChannelHandler.scala | 3 +
.../promisehandlers/StartChannelHandler.scala | 3 +
.../amber/engine/e2e/LoopIntegrationSpec.scala | 51 ++++++
.../test/python/core/runnables/test_main_loop.py | 179 +++++++++++++++++++++
5 files changed, 323 insertions(+), 3 deletions(-)
diff --git a/amber/src/main/python/core/runnables/main_loop.py
b/amber/src/main/python/core/runnables/main_loop.py
index fe44b36044..048bc0a289 100644
--- a/amber/src/main/python/core/runnables/main_loop.py
+++ b/amber/src/main/python/core/runnables/main_loop.py
@@ -93,6 +93,11 @@ class MainLoop(StoppableQueueBlockingRunnable):
# same iteration's state arrives once per branch. Workers are recreated
# on each region re-execution, so this instance flag is per iteration.
self._loop_state_consumed: bool = False
+ # Whether this LoopEnd forwarded an UNstamped counter-0 state instead
+ # of consuming it. Paired with _loop_start_id by
+ # _check_loop_state_arrived: forwarding one and never capturing a stamp
+ # means the loop's own state reached here with its stamp lost.
+ self._forwarded_unstamped_state: bool = False
self.context = Context(worker_id, input_queue)
self._async_rpc_server = AsyncRPCServer(output_queue,
context=self.context)
@@ -127,6 +132,48 @@ class MainLoop(StoppableQueueBlockingRunnable):
writer.put_one(executor.state.to_tuple(0))
writer.close()
+ def _check_loop_state_arrived(self) -> None:
+ # Keep the LOUD failure for a real loop state that lost its stamp
+ # upstream. Forwarding unstamped states (above) is right for a body
+ # operator's own boundary state, but it also swallows the symptom of
+ # the bug class #6660/#6661 fixed: a hop that blanks the envelope makes
+ # the loop's own state arrive unstamped, and forwarding it leaves
+ # _loop_table None, so condition() returns False and the loop stops
+ # after one iteration -- a WRONG RESULT reported as success.
+ #
+ # The two cases are told apart once the input port is done rather than
+ # on arrival: a LoopEnd that forwarded an unstamped state and never
+ # took a stamped one cannot have been looking at a body operator's
+ # boundary state -- its own state never arrived. Deciding at EndChannel
+ # is order-independent (the body operator's state may arrive before or
+ # after the loop's; EndChannel is PORT_ALIGNMENT, so every channel on
+ # the port has been drained) and reads nothing out of the State, so it
+ # keeps working once the input table stops riding inside it (#6971).
+ #
+ # Called from _process_end_channel, NOT complete(): complete() runs
+ # after port_completed has gone out for the input port and every output
+ # port, and region completion is port-based, so a raise there would be
+ # reported only once the coordinator already considers the region done.
+ #
+ # The check is deliberately narrow -- "forwarded an unstamped state AND
+ # never took a stamped one", not "never took a stamped one". A LoopEnd
+ # completing without any matching state is legal (see
+ # LoopEndOperator.eval_condition's _loop_table guard); only positive
+ # evidence that a boundary state arrived unstamped is a lost envelope.
+ #
+ # "Took a stamped one" is read off _loop_start_id, which only the
+ # stamped branch writes and nothing clears -- NOT off the
+ # _loop_state_consumed fan-in dedup flag, whose lifetime is owned by
+ # the dedup and may end before this runs. Keying on the durable field
+ # keeps this guard correct wherever it is called from.
+ if self._forwarded_unstamped_state and not self._loop_start_id:
+ raise RuntimeError(
+ "Loop End received a loop-boundary state with no LoopStart "
+ "stamp and never received its own (stamped) loop state: the "
+ "loop envelope was lost upstream, so this loop would silently "
+ "stop after one iteration"
+ )
+
def complete(self) -> None:
"""
Complete the DataProcessor, marking state to COMPLETED, and notify the
@@ -379,10 +426,42 @@ class MainLoop(StoppableQueueBlockingRunnable):
self._check_and_process_control()
return
+ # A LoopStart handles only the STAMPED case above. An UNstamped
+ # counter-0 state at a LoopStart takes neither branch: it falls all the
+ # way through to process_input_state at the bottom, and the operator's
+ # process_state MERGES its keys into the loop variables. That is the
+ # opposite of what the LoopEnd branch below does with the identical
+ # frame, and the asymmetry is forced, not an oversight. The back-edge
+ # writes the next iteration's variables to the LoopStart's own
+ # input-port state URI with that very same "no loop" envelope
+ # (_jump_to_loop_start -> State.to_tuple(0)), so an unstamped counter-0
+ # frame at a LoopStart is indistinguishable from -- and normally IS --
+ # the loop's own state. A LoopEnd may forward instead of consuming
+ # because its inbound loop state is always stamped by the matching
+ # LoopStart; a LoopStart has no such signal. Consequence of the merge:
+ # an upstream or body operator emitting a key that collides with a loop
+ # variable overwrites it, and one emitting `table` trips
+ # _reserved_name_error in produce_state_on_finish (see #7248).
+
if isinstance(executor, LoopEndOperator):
- # Matching LoopEnd (in_counter == 0): it will consume this state
- # and jump back. Remember which LoopStart to jump to (it rides
- # the envelope) for complete()/_jump_to_loop_start.
+ if not frame.loop_start_id:
+ # An UNstamped counter-0 state at a LoopEnd is not the loop's
+ # own boundary state -- it was produced by a loop-body
+ # operator's produce_state_on_start/finish (a public API on
+ # both engine sides), which emits with the "no loop" envelope.
+ # A real loop state is always stamped: the matching LoopStart
+ # stamps its own id on every iteration's output state.
+ # Forward it downstream unchanged, skipping the operator, like
+ # any default pass-through: consuming it would clobber the
+ # captured back-jump id with "" and hand run_update a State
+ # with no `table` payload.
+ self._forwarded_unstamped_state = True
+ self._emit_and_save_state(state, in_counter,
frame.loop_start_id)
+ self._check_and_process_control()
+ return
+ # Matching LoopEnd (in_counter == 0, stamped): it will consume this
+ # state and jump back. Remember which LoopStart to jump to (it
+ # rides the envelope) for complete()/_jump_to_loop_start.
#
# With a branching loop body, each branch's reader replays the same
# iteration's state, so this fires once per inbound link. Consume
@@ -412,6 +491,11 @@ class MainLoop(StoppableQueueBlockingRunnable):
def _process_end_channel(self) -> None:
self.process_input_state()
+ try:
+ self._check_loop_state_arrived()
+ except Exception as err:
+ self.context.report_exception(err)
+ self._check_exception()
if self.context.exception_manager.has_exception():
# A state-emission error was reported on the main loop thread (see
# _emit_and_save_state). Hold the region: skip port_completed and
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala
index 7794342690..f1b37d7d32 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala
@@ -43,6 +43,9 @@ trait EndChannelHandler {
try {
val outputState = dp.executor.produceStateOnFinish(portId.id)
if (outputState.isDefined) {
+ // Operator-ORIGINATED boundary state, so no LoopStart stamp
+ // (loopCounter = 0, loopStartId = ""); see
+ // `main_loop._process_state_frame` for how a Loop End treats it.
dp.outputManager.emitState(outputState.get)
}
dp.outputManager.outputIterator.setTupleOutput(
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartChannelHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartChannelHandler.scala
index 01cbb858bd..84de874aa6 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartChannelHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartChannelHandler.scala
@@ -42,6 +42,9 @@ trait StartChannelHandler {
try {
val outputState = dp.executor.produceStateOnStart(portId.id)
if (outputState.isDefined) {
+ // Operator-ORIGINATED boundary state, so no LoopStart stamp
+ // (loopCounter = 0, loopStartId = ""); see
+ // `main_loop._process_state_frame` for how a Loop End treats it.
dp.outputManager.emitState(outputState.get)
}
} catch safely {
diff --git
a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
index 858e10e20e..9da249f996 100644
---
a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
+++
b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
@@ -45,6 +45,7 @@ import org.apache.texera.amber.operator.limit.LimitOpDesc
import org.apache.texera.amber.operator.loop.{LoopEndOpDesc, LoopStartOpDesc}
import org.apache.texera.amber.operator.sleep.SleepOpDesc
import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.operator.udf.python.PythonUDFOpDescV2
import org.apache.texera.amber.tags.IntegrationTest
import org.apache.texera.common.compiler.model.LogicalLink
import org.scalatest.flatspec.AnyFlatSpecLike
@@ -195,6 +196,29 @@ class LoopIntegrationSpec
// any loop logic runs. Production workers launch with a real classpath,
// so this is a harness limitation, not an engine one.
+ /** A pass-through Python UDF that ALSO emits its own boundary state via the
+ * public `produce_state_on_finish` API -- the state arrives downstream
+ * with the "no loop" envelope (counter 0, no LoopStart stamp).
+ */
+ private def statefulPythonUDF(): PythonUDFOpDescV2 = {
+ val op = new PythonUDFOpDescV2()
+ op.code = """from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ |
+ | @overrides
+ | def process_tuple(self, tuple_: Tuple, port: int) ->
Iterator[Optional[TupleLike]]:
+ | yield tuple_
+ |
+ | @overrides
+ | def produce_state_on_finish(self, port: int) ->
Optional[State]:
+ | return State({"note": "from-body-op"})
+ |""".stripMargin
+ op.workers = 1
+ op.retainInputColumns = true
+ op
+ }
+
private def link(from: LogicalOp, to: LogicalOp): LogicalLink =
LogicalLink(from.operatorIdentifier, PortIdentity(),
to.operatorIdentifier, PortIdentity())
@@ -342,4 +366,31 @@ class LoopIntegrationSpec
)
}
+ it should "run a loop whose body operator emits its own boundary state" in {
+ // TextInput -> LoopStart -> stateful Python UDF -> LoopEnd.
+ //
+ // The UDF emits boundary state via produce_state_on_finish (a public API
+ // on both engine sides), which reaches the LoopEnd with the "no loop"
+ // envelope (counter 0, no LoopStart stamp) AFTER the forwarded loop
+ // state. The LoopEnd must pass that unstamped state through instead of
+ // consuming it: consuming would clobber the captured back-jump id with ""
+ // ("no loop-back state URI configured for LoopStart ''") and hand
+ // run_update a State with no `table` payload (KeyError). Regression test
+ // for #discussion_r3648708075 on #6661.
+ val src = textInput("1\n2\n3")
+ val start = loopStart("i = 0", "table.iloc[i]")
+ val mid = statefulPythonUDF()
+ val end = loopEnd("i += 1", "i < len(table)")
+ val materialized = runAndGetMaterializedRowCounts(
+ List(src, start, mid, end),
+ List(link(src, start), link(start, mid), link(mid, end))
+ )
+ val endRows = materialized.getOrElse(end.operatorIdentifier, -1L)
+ assert(
+ endRows == 3,
+ s"LoopEnd must accumulate all 3 iterations with a state-emitting " +
+ s"operator in the loop body: expected 3, got $endRows (all:
$materialized)"
+ )
+ }
+
}
diff --git a/amber/src/test/python/core/runnables/test_main_loop.py
b/amber/src/test/python/core/runnables/test_main_loop.py
index 73f3957559..1da2d19691 100644
--- a/amber/src/test/python/core/runnables/test_main_loop.py
+++ b/amber/src/test/python/core/runnables/test_main_loop.py
@@ -2416,6 +2416,44 @@ class TestMainLoop:
assert emitted_id == "outer-loop"
assert reset_calls == [], "a LoopStart never resets output storage"
+ def test_loopstart_merges_unstamped_state_instead_of_forwarding_it(
+ self, main_loop, monkeypatch
+ ):
+ # The deliberate asymmetry with the LoopEnd branch: an UNstamped
+ # counter-0 frame at a LoopStart is MERGED into the loop variables,
+ # not forwarded. It has to be -- the back-edge writes the next
+ # iteration's variables to this LoopStart's own input-port state URI
+ # with the identical "no loop" envelope (State.to_tuple(0)), so a
+ # LoopStart cannot tell the loop's own state from an upstream/body
+ # operator's boundary state. A LoopEnd can, because its inbound loop
+ # state is always stamped.
+ class StubLoopStart(LoopStartOperator):
+ def process_table(self, table, port):
+ yield
+
+ executor = StubLoopStart()
+ main_loop.context.executor_manager.executor = executor
+ emitted, switched, reset_calls = self._capture_state_emit(
+ main_loop, monkeypatch
+ )
+ monkeypatch.setattr(
+ main_loop.context.state_processing_manager,
+ "get_output_state",
+ lambda: None,
+ )
+
+ main_loop._process_state_frame(
+ StateFrame(State({"seed": 7}), loop_counter=0, loop_start_id="")
+ )
+
+ assert emitted == [], "an unstamped state at a LoopStart is not
forwarded"
+ assert switched == [True], "it reaches the operator, which merges it"
+ assert reset_calls == []
+ # It is handed to the operator as the current input state, so
+ # LoopStartOperator.process_state merges it into self.state.
+ passed = main_loop.context.state_processing_manager.current_input_state
+ assert passed == State({"seed": 7})
+
def test_loopend_passthrough_decrements_resets_output_and_skips_operator(
self, main_loop, monkeypatch
):
@@ -2495,6 +2533,147 @@ class TestMainLoop:
assert "loop_start_id" not in passed_to_operator
assert "loop_counter" not in passed_to_operator
+ def test_loopend_forwards_unstamped_state_without_consuming(
+ self, main_loop, monkeypatch
+ ):
+ # A loop-body operator that emits its own boundary state
+ # (produce_state_on_start/finish -- a public API on both engine sides)
+ # sends it with the "no loop" envelope (counter 0, id ""). That state
+ # is NOT the loop's own boundary state: consuming it would clobber the
+ # captured back-jump id with "" and hand run_update a State with no
+ # `table` payload (KeyError). A real loop state is always stamped --
+ # the matching LoopStart stamps its own id on every iteration's output
+ # -- so an UNstamped counter-0 frame at a LoopEnd must be forwarded
+ # downstream unchanged, skipping the operator, like any default
+ # pass-through.
+ main_loop.context.executor_manager.executor = _FalseLoopEnd()
+ emitted, switched, reset_calls = self._capture_state_emit(
+ main_loop, monkeypatch
+ )
+ # The loop's own state was already consumed and its id captured.
+ main_loop._loop_start_id = "loop-start-1"
+
+ main_loop._process_state_frame(
+ StateFrame(
+ State({"note": "from-body-op"}),
+ loop_counter=0,
+ loop_start_id="",
+ )
+ )
+
+ assert switched == [], "unstamped state must not invoke the operator"
+ assert reset_calls == [], "unstamped state must not reset output"
+ assert emitted == [(State({"note": "from-body-op"}), 0, "")], (
+ "unstamped state must forward downstream with its envelope "
+ f"unchanged; emitted: {emitted}"
+ )
+ assert main_loop._loop_start_id == "loop-start-1", (
+ "an unstamped state must not clobber the captured back-jump id"
+ )
+ # Forwarding an unstamped state must NOT count as taking the loop's
+ # own state -- that is what the completion guard keys on.
+ assert main_loop._loop_state_consumed is False
+
+ @pytest.mark.timeout(2)
+ def test_end_channel_holds_the_region_when_only_unstamped_states_arrived(
+ self, main_loop, monkeypatch
+ ):
+ # Forwarding unstamped states is right for a body operator's own
+ # boundary state, but it must not swallow the symptom of a LOST stamp:
+ # a hop that blanks the envelope (the bug class #6660/#6661 fixed)
+ # makes the loop's OWN state arrive unstamped, and forwarding it leaves
+ # _loop_table None -> condition() False -> one iteration reported as
+ # success. A Loop End that forwarded an unstamped state and never took
+ # a stamped one must fail loudly instead -- and it must do so from
+ # _process_end_channel, BEFORE port_completed goes out, because region
+ # completion is port-based: reported from complete() the error would
+ # arrive after the coordinator already considers the region done.
+ executor = _FalseLoopEnd()
+ main_loop.context.executor_manager.executor = executor
+ self._capture_state_emit(main_loop, monkeypatch)
+
+ completed = []
+ port_completed_calls = []
+ console_msgs = []
+ monkeypatch.setattr(main_loop, "process_input_tuple", lambda: None)
+ monkeypatch.setattr(main_loop, "complete", lambda:
completed.append(True))
+ monkeypatch.setattr(
+ main_loop, "_send_console_message", lambda msg:
console_msgs.append(msg)
+ )
+ monkeypatch.setattr(
+ main_loop.context.pause_manager,
+ "pause",
+ lambda pause_type, change_state=True: None,
+ )
+
+ class _Coordinator:
+ def port_completed(self, request):
+ port_completed_calls.append(request)
+
+ monkeypatch.setattr(
+ main_loop._async_rpc_client, "coordinator_stub", lambda:
_Coordinator()
+ )
+
+ # The loop's own state arrives with its stamp lost upstream.
+ main_loop._process_state_frame(
+ StateFrame(State({"i": 0}), loop_counter=0, loop_start_id="")
+ )
+ assert main_loop._forwarded_unstamped_state is True
+ assert main_loop._loop_state_consumed is False
+
+ monkeypatch.setattr(main_loop, "process_input_state", lambda *a, **k:
None)
+ main_loop._process_end_channel()
+
+ assert main_loop.context.exception_manager.has_exception()
+ error_msgs = [m for m in console_msgs if m.msg_type ==
ConsoleMessageType.ERROR]
+ assert len(error_msgs) == 1
+ assert "loop envelope was lost upstream" in error_msgs[0].title
+ assert port_completed_calls == [], "no port may be reported complete"
+ assert completed == [], "the worker must not complete"
+
+ def test_complete_accepts_unstamped_state_alongside_the_loop_state(
+ self, main_loop, monkeypatch
+ ):
+ # The legitimate shape this PR enables: a body operator's boundary
+ # state is forwarded AND the loop's own stamped state is consumed. The
+ # completion guard must stay silent, in either arrival order.
+ for body_state_first in (True, False):
+ executor = _FalseLoopEnd()
+ main_loop.context.executor_manager.executor = executor
+ main_loop._forwarded_unstamped_state = False
+ main_loop._loop_state_consumed = False
+ main_loop._loop_start_id = ""
+ self._capture_state_emit(main_loop, monkeypatch)
+ monkeypatch.setattr(
+ main_loop.context.state_processing_manager,
+ "get_output_state",
+ lambda: None,
+ )
+
+ body = StateFrame(
+ State({"note": "from-body-op"}), loop_counter=0,
loop_start_id=""
+ )
+ loop = StateFrame(
+ State({"i": 0}), loop_counter=0, loop_start_id="outer-loop"
+ )
+ for frame in (body, loop) if body_state_first else (loop, body):
+ main_loop._process_state_frame(frame)
+
+ # Must not raise, in either order.
+ main_loop._check_loop_state_arrived()
+ assert main_loop._forwarded_unstamped_state is True
+ assert main_loop._loop_state_consumed is True
+ assert main_loop._loop_start_id == "outer-loop"
+
+ # The guard must key on the DURABLE evidence that a stamped state
+ # was taken (_loop_start_id), not on the fan-in dedup flag: that
+ # flag belongs to the dedup, and a caller may legitimately clear
+ # it once the iteration's state has been taken. Keyed on the flag,
+ # this legitimate shape would start raising the moment anything
+ # re-armed it -- so simulate that and require silence.
+ main_loop._loop_state_consumed = False
+ main_loop._check_loop_state_arrived()
+
def test_loopend_consumes_its_loop_state_once_per_iteration(
self, main_loop, monkeypatch
):
