Rust #325: query-authority clients wait for the acknowledgement before stopping

scripts/conformance/server-parity/query-authority/python.py

@@ -117,6 +117,17 @@ async def main():
                 if MODE == 'timely':
                     result = await client.query_workflow(namespace, 'value', args=[])
                     assert result['result'] == EXPECTED and type(result['result']['long']) is int
+                    # The independent caller can observe the commit before the
+                    # Worker receives its acknowledgement. Wait inside the
+                    # original budget before stopping it, as worker.php does.
+                    PHASE = 'acknowledgement'
+                    while not any(x['event'] == 'acknowledgement' for x in events()):
+                        if runner.done():
+                            await runner
+                            raise RuntimeError('Worker stopped before its acknowledgement')
+                        if time.monotonic() - started >= 30:
+                            raise AssertionError('No acknowledgement inside the original budget')
+                        await asyncio.sleep(.02)
                 else:
                     try:
                         await client.query_workflow(namespace, 'value', args=[])
@@ -125,12 +136,19 @@ async def main():
                         assert error.status == 504 and error.reason == 'query_worker_execution_timeout'
                     # Delivery is recorded before the SDK receives its response.
                     # Wait for its actual refusal before stopping the worker.
+                    # A late handler's callback counter moves before the SDK
+                    # logs the lost authority, so every mode also waits for
+                    # that evidence, as worker.php waits for its refusal.
                     PHASE = 'expiry-observation'
-                    while (MODE == 'late-poll' and not any(
-                               expired_losses(event['claim']['query_task_id'])
-                               for event in events() if event['event'] == 'claim')
+                    while (not any(expired_losses(event['claim']['query_task_id'])
+                                   for event in events() if event['event'] == 'claim')
                            or MODE == 'late-handler-async' and COUNTERS['cancelled'] != 1
                            or MODE == 'late-handler-sync' and COUNTERS['returned'] != 1):
+                        if runner.done():
+                            await runner
+                            raise RuntimeError('Worker stopped before its expiry evidence')
+                        if time.monotonic() - started >= 30:
+                            raise AssertionError('No expiry evidence inside the original budget')
                         await asyncio.sleep(.02)
             finally:
                 PHASE = 'worker-stop'

scripts/conformance/server-parity/query-authority/rust/src/main.rs

@@ -169,6 +169,13 @@ async fn main() -> Result<()> {
         let query = client.query_workflow(&namespace, "value", json!([])).await;
         if mode == "timely" {
             assert_eq!(query?, expected());
+            // The independent caller can observe the commit before the Worker
+            // receives its acknowledgement. Wait inside the original budget
+            // before stopping it, as worker.php does.
+            wait(until, || {
+                !named(&events(Path::new(journal)), "acknowledgement").is_empty()
+            })
+            .await?;
         } else {
             match query {
                 Err(Error::QueryFailed(failure)) => {