Skip to content

Commit 8e43a0c

Browse files
durable-workflow-opsdurable-workflow-ops
committed
Conformance: restore signals/queries Python baseline current readout (#207)
1 parent dced72d commit 8e43a0c

4 files changed

Lines changed: 193 additions & 1 deletion

File tree

‎CHANGELOG.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
3434
`NamespaceDescription.deleted`.
3535

3636
### Fixed
37+
- Worker-routed queries now replay signal-woken condition waits that reopen the
38+
same logical wait until all recorded signals have been applied, so long-lived
39+
queryable workflows report the latest signal-mutated state.
3740
- Python conformance CLI evidence now accepts terminal stdout/JSON captured on
3841
nested CLI command records while still rejecting generic scenario outputs as
3942
proof of the public CLI result path.

‎src/durable_workflow/workflow.py‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2762,7 +2762,17 @@ def _terminal_state(value: Any, *, include_pending: bool) -> _ReplayState:
27622762
exception_type=type(exc).__name__,
27632763
)])
27642764
if resolution == "satisfied":
2765-
terminal_condition_reopen_cmd = None
2765+
if has_reopened_same_wait:
2766+
if not satisfied:
2767+
consumed_reopen_for_current_wait = True
2768+
wait_yield_count = next_wait_index
2769+
continue
2770+
2771+
terminal_condition_reopen_cmd = (
2772+
cmd
2773+
if has_reopened_same_wait and consumed_reopen_for_current_wait
2774+
else None
2775+
)
27662776
next_value = True
27672777
wait_yield_count += 1
27682778
break

‎tests/test_wait_condition.py‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -172,6 +172,41 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def]
172172
return self.current()
173173

174174

175+
@workflow.defn(name="signal-counter-single-predicate-per-reopen")
176+
class SignalCounterSinglePredicatePerReopen:
177+
def __init__(self) -> None:
178+
self.count = 0
179+
self.done = False
180+
self.receiver_generation = 0
181+
self.last_predicate_generation: int | None = None
182+
self.predicate_calls = 0
183+
184+
@workflow.signal("increment")
185+
def increment(self, amount: int) -> None:
186+
self.count += amount
187+
self.receiver_generation += 1
188+
189+
@workflow.signal("finish")
190+
def finish(self) -> None:
191+
self.done = True
192+
self.receiver_generation += 1
193+
194+
def done_once_per_receiver(self) -> bool:
195+
self.predicate_calls += 1
196+
if self.last_predicate_generation == self.receiver_generation:
197+
raise RuntimeError("predicate re-evaluated before replay advanced")
198+
self.last_predicate_generation = self.receiver_generation
199+
return self.done
200+
201+
def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def]
202+
satisfied = yield ctx.wait_condition(self.done_once_per_receiver, key="done")
203+
return {
204+
"count": self.count,
205+
"satisfied": satisfied,
206+
"predicate_calls": self.predicate_calls,
207+
}
208+
209+
175210
@workflow.defn(name="two-same-key-signal-waits")
176211
class TwoSameKeySignalWaits:
177212
def __init__(self) -> None:
@@ -455,6 +490,38 @@ def test_repeated_physical_waits_keep_query_state_current_before_finish(self) ->
455490
],
456491
}
457492

493+
def test_satisfied_reopened_wait_reuses_current_predicate_result(self) -> None:
494+
history = [
495+
{
496+
"event_type": "ConditionWaitOpened",
497+
"payload": {"condition_wait_id": "wait-count", "condition_key": "done"},
498+
},
499+
_signal_received_event("increment", [3]),
500+
{
501+
"event_type": "ConditionWaitSatisfied",
502+
"payload": {"condition_wait_id": "wait-count", "condition_key": "done"},
503+
},
504+
{
505+
"event_type": "ConditionWaitOpened",
506+
"payload": {"condition_wait_id": "wait-finish", "condition_key": "done"},
507+
},
508+
_signal_received_event("finish", []),
509+
{
510+
"event_type": "ConditionWaitSatisfied",
511+
"payload": {"condition_wait_id": "wait-finish", "condition_key": "done"},
512+
},
513+
]
514+
515+
outcome = replay(SignalCounterSinglePredicatePerReopen, history, [])
516+
517+
assert len(outcome.commands) == 1
518+
assert isinstance(outcome.commands[0], CompleteWorkflow)
519+
assert outcome.commands[0].result == {
520+
"count": 3,
521+
"satisfied": True,
522+
"predicate_calls": 2,
523+
}
524+
458525
def test_signal_received_before_next_wait_uses_declared_workflow_sequence(self) -> None:
459526
history = [
460527
{

‎tests/test_worker.py‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1443,6 +1443,118 @@ async def test_query_task_replays_repeated_condition_wait_signal_arguments(
14431443
)
14441444
mock_client.fail_query_task.assert_not_called()
14451445

1446+
@pytest.mark.asyncio
1447+
async def test_query_task_replays_signal_woken_false_wait_reopens(
1448+
self, mock_client: AsyncMock
1449+
) -> None:
1450+
worker = Worker(mock_client, task_queue="q1", workflows=[CounterQueryWorkflow], activities=[])
1451+
first_signal_arguments = serializer.encode([3], codec="json")
1452+
second_signal_arguments = serializer.encode([5], codec="json")
1453+
task = {
1454+
"query_task_id": "qt-false-wait-reopens",
1455+
"query_task_attempt": 1,
1456+
"workflow_type": "counter-query-wf",
1457+
"workflow_id": "wf-counter",
1458+
"run_id": "run-counter",
1459+
"query_name": "current",
1460+
"history_events": [
1461+
{
1462+
"event_type": "ConditionWaitOpened",
1463+
"payload": {
1464+
"condition_wait_id": "wait-count-0",
1465+
"condition_key": "done",
1466+
},
1467+
},
1468+
{
1469+
"event_type": "SignalReceived",
1470+
"workflow_command_id": "cmd-increment-3",
1471+
"payload": {
1472+
"signal_id": "sig-increment-3",
1473+
"workflow_command_id": "cmd-increment-3",
1474+
"signal_name": "increment",
1475+
},
1476+
},
1477+
{
1478+
"event_type": "ConditionWaitSatisfied",
1479+
"payload": {
1480+
"condition_wait_id": "wait-count-0",
1481+
"condition_key": "done",
1482+
"workflow_signal_id": "sig-increment-3",
1483+
"signal_name": "increment",
1484+
},
1485+
},
1486+
{
1487+
"event_type": "ConditionWaitOpened",
1488+
"payload": {
1489+
"condition_wait_id": "wait-count-3",
1490+
"condition_key": "done",
1491+
},
1492+
},
1493+
{
1494+
"event_type": "SignalReceived",
1495+
"workflow_command_id": "cmd-increment-5",
1496+
"payload": {
1497+
"signal_id": "sig-increment-5",
1498+
"workflow_command_id": "cmd-increment-5",
1499+
"signal_name": "increment",
1500+
},
1501+
},
1502+
{
1503+
"event_type": "ConditionWaitSatisfied",
1504+
"payload": {
1505+
"condition_wait_id": "wait-count-3",
1506+
"condition_key": "done",
1507+
"workflow_signal_id": "sig-increment-5",
1508+
"signal_name": "increment",
1509+
},
1510+
},
1511+
{
1512+
"event_type": "ConditionWaitOpened",
1513+
"payload": {
1514+
"condition_wait_id": "wait-count-8",
1515+
"condition_key": "done",
1516+
},
1517+
},
1518+
],
1519+
"history_export": {
1520+
"payloads": {"codec": "json"},
1521+
"signals": [
1522+
{
1523+
"id": "sig-increment-3",
1524+
"command_id": "cmd-increment-3",
1525+
"name": "increment",
1526+
"payload_codec": "json",
1527+
"arguments": first_signal_arguments,
1528+
},
1529+
{
1530+
"id": "sig-increment-5",
1531+
"command_id": "cmd-increment-5",
1532+
"name": "increment",
1533+
"payload_codec": "json",
1534+
"arguments": second_signal_arguments,
1535+
},
1536+
],
1537+
},
1538+
"workflow_arguments": serializer.envelope([], codec="json"),
1539+
"query_arguments": serializer.envelope([], codec="json"),
1540+
"payload_codec": "json",
1541+
}
1542+
1543+
outcome = await worker._run_query_task(task)
1544+
1545+
assert outcome == "completed"
1546+
mock_client.complete_query_task.assert_awaited_once_with(
1547+
query_task_id="qt-false-wait-reopens",
1548+
lease_owner=worker.worker_id,
1549+
query_task_attempt=1,
1550+
result=8,
1551+
codec="json",
1552+
workflow_id="wf-counter",
1553+
run_id="run-counter",
1554+
query_name="current",
1555+
)
1556+
mock_client.fail_query_task.assert_not_called()
1557+
14461558
@pytest.mark.asyncio
14471559
async def test_query_task_replays_history_from_export_after_worker_restart(
14481560
self, mock_client: AsyncMock

0 commit comments

Comments
 (0)