Speed up CI

.github/workflows/run-tests.yaml

@@ -97,11 +97,14 @@ jobs:
         with:
           commit: ${{ github.event.inputs.daprdapr_commit }}
           github-token: ${{ secrets.GITHUB_TOKEN }}
+      # The official install.sh spends ~1.5 minutes creating a system user and
+      # a systemd service that CI does not need; extracting the tarball and
+      # running `ollama serve` directly takes seconds.
       - name: Set up Llama
         run: |
-          curl -fsSL https://ollama.com/install.sh | sh
-          nohup ollama serve &
-          sleep 10
+          curl -fsSL https://github.com/ollama/ollama/releases/latest/download/ollama-linux-amd64.tar.zst | sudo tar --zstd -xf - -C /usr/local
+          nohup ollama serve > "$RUNNER_TEMP/ollama-serve.log" 2>&1 &
+          timeout 60 bash -c 'until curl -fsS http://127.0.0.1:11434/api/tags > /dev/null; do sleep 1; done'
           ollama pull llama3.2:latest
       - name: Run integration tests
         run: |

examples/demo_actor/README.md

@@ -87,7 +87,7 @@ expected_stdout_lines:
   - 'call GetMyData actor method to get the state'
   - 'Register reminder'
   - 'Register timer'
-  - 'waiting for 30 seconds'
+  - 'waiting for 12 seconds'
   - 'stop reminder'
   - 'stop timer'
   - 'clear actor state'
@@ -115,7 +115,7 @@ expected_stdout_lines:
    {'data': 'new_data', 'ts': datetime.datetime(2020, 11, 13, 0, 38, 36, 163000, tzinfo=tzutc())}
    Register reminder
    Register timer
-   waiting for 30 seconds
+   waiting for 12 seconds
    stop reminder
    stop timer
    clear actor state

examples/demo_actor/demo_actor/demo_actor_client.py

@@ -63,9 +63,9 @@ async def main():
     print('Register timer', flush=True)
     await proxy.SetTimer(True)
 
-    # Wait for 30 seconds to see reminder and timer is triggered
-    print('waiting for 30 seconds', flush=True)
-    await asyncio.sleep(30)
+    # Both fire at due_time=5s, so 12 seconds is enough to observe them
+    print('waiting for 12 seconds', flush=True)
+    await asyncio.sleep(12)
 
     # Stop reminder and timer
     print('stop reminder', flush=True)

tests/examples/conftest.py

@@ -16,6 +16,7 @@
     'bind: address already in use',
     'failed to start internal gRPC server: could not listen on any endpoint',
 )
+DAPR_SIDECAR_READY_MARKER = "You're up and running!"
 
 
 def pytest_configure(config: pytest.Config) -> None:
@@ -113,7 +114,7 @@ def _run_once(self, args: str, *, timeout: int, until: list[str] | None) -> str:
 
         return ''.join(lines)
 
-    def start(self, args: str, *, wait: int = 5, port_bind_retries: int = 3) -> None:
+    def start(self, args: str, *, wait: int = 30, port_bind_retries: int = 3) -> None:
         """Start a long-lived background service.
 
         Use this for servers/subscribers that must stay alive while a second
@@ -122,7 +123,8 @@ def start(self, args: str, *, wait: int = 5, port_bind_retries: int = 3) -> None
 
         Args:
             args: Arguments passed to ``dapr run``.
-            wait: Seconds to wait for the sidecar to come up before returning.
+            wait: Maximum seconds to poll for sidecar readiness before
+                proceeding anyway.
             port_bind_retries: Retry count for Dapr sidecar startup failures
                 caused by a transient random-port collision.
         """
@@ -137,7 +139,7 @@ def start(self, args: str, *, wait: int = 5, port_bind_retries: int = 3) -> None
                 text=True,
                 **get_kwargs_for_process_group(),
             )
-            time.sleep(wait)
+            self._wait_until_ready(proc, output_file, timeout=wait)
 
             can_retry = attempt < attempts - 1
             if can_retry and self._started_with_port_bind_failure(proc, output_file):
@@ -156,6 +158,29 @@ def start(self, args: str, *, wait: int = 5, port_bind_retries: int = 3) -> None
             self._bg_output_file = output_file
             return
 
+    @staticmethod
+    def _wait_until_ready(
+        proc: subprocess.Popen[str], output_file: IO[str], *, timeout: int
+    ) -> None:
+        """Polls the sidecar log every second until `dapr run` reports readiness.
+
+        Returns early when the process exits (``start`` then inspects the
+        output for a port-bind failure) and gives up after ``timeout`` seconds,
+        at which point the caller proceeds as if ready.
+
+        The log is re-opened by name for each read: the ``output_file`` handle
+        shares its offset with the child process, so seeking it directly would
+        corrupt the child's writes.
+        """
+        log_path = Path(output_file.name)
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if proc.poll() is not None:
+                return
+            if DAPR_SIDECAR_READY_MARKER in log_path.read_text(errors='replace'):
+                return
+            time.sleep(1)
+
     def _started_with_port_bind_failure(
         self, proc: subprocess.Popen[str], output_file: IO[str]
     ) -> bool:

tests/examples/test_configuration.py

@@ -17,7 +17,6 @@ def test_configuration(dapr, redis_set_config):
 
     dapr.start(
         '--app-id configexample --resources-path components/ -- python3 configuration.py',
-        wait=5,
     )
     # Update Redis to trigger the subscription notification
     redis_set_config('orderId2', '210', version=2)

tests/examples/test_dapr_runner.py

@@ -96,6 +96,9 @@ def fake_popen(*args, **kwargs) -> FakeProcess:
 )
 
 
+SIDECAR_READY_OUTPUT = "✅  You're up and running! Both Dapr and your app logs will appear here.\n"
+
+
 def test_start_retries_transient_dapr_port_bind_failure(
     monkeypatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
 ) -> None:
@@ -117,7 +120,7 @@ def fake_popen(*args, **kwargs) -> FakeBackgroundProcess:
     DaprRunner(tmp_path).start('--app-id demo-actor -- uvicorn demo:app', wait=0)
 
     assert len(popen_calls) == 2
-    assert sleeps == [0, 1, 0]
+    assert sleeps == [1]
     assert (
         'Dapr background sidecar failed to bind a random port; retrying startup after 1s'
         in capsys.readouterr().out
@@ -136,3 +139,62 @@ def fake_popen(*args, **kwargs) -> FakeBackgroundProcess:
     DaprRunner(tmp_path).start('--app-id demo-actor -- uvicorn demo:app', wait=0)
 
     assert len(popen_calls) == 1
+
+
+def test_start_returns_without_sleeping_when_sidecar_is_already_ready(
+    monkeypatch, tmp_path: Path
+) -> None:
+    def fake_popen(*args, **kwargs) -> FakeBackgroundProcess:
+        return FakeBackgroundProcess(SIDECAR_READY_OUTPUT, None, kwargs['stdout'])
+
+    monkeypatch.setattr(subprocess, 'Popen', fake_popen)
+    sleeps: list[int] = []
+    monkeypatch.setattr(time, 'sleep', sleeps.append)
+
+    DaprRunner(tmp_path).start('--app-id demo-actor -- uvicorn demo:app', wait=30)
+
+    assert sleeps == []
+
+
+def test_start_polls_every_second_until_sidecar_is_ready(monkeypatch, tmp_path: Path) -> None:
+    stdout_files: list[IO[str]] = []
+
+    def fake_popen(*args, **kwargs) -> FakeBackgroundProcess:
+        stdout_files.append(kwargs['stdout'])
+        return FakeBackgroundProcess('sidecar still starting\n', None, kwargs['stdout'])
+
+    monkeypatch.setattr(subprocess, 'Popen', fake_popen)
+    sleeps: list[int] = []
+
+    def sleep_then_become_ready(seconds: int) -> None:
+        sleeps.append(seconds)
+        if len(sleeps) == 2:
+            stdout_files[0].write(SIDECAR_READY_OUTPUT)
+            stdout_files[0].flush()
+
+    monkeypatch.setattr(time, 'sleep', sleep_then_become_ready)
+
+    DaprRunner(tmp_path).start('--app-id demo-actor -- uvicorn demo:app', wait=30)
+
+    assert sleeps == [1, 1]
+
+
+def test_start_gives_up_polling_after_wait_seconds(monkeypatch, tmp_path: Path) -> None:
+    def fake_popen(*args, **kwargs) -> FakeBackgroundProcess:
+        return FakeBackgroundProcess('sidecar never becomes ready\n', None, kwargs['stdout'])
+
+    monkeypatch.setattr(subprocess, 'Popen', fake_popen)
+
+    clock = {'now': 0.0}
+    sleeps: list[int] = []
+
+    def fake_sleep(seconds: int) -> None:
+        sleeps.append(seconds)
+        clock['now'] += seconds
+
+    monkeypatch.setattr(time, 'monotonic', lambda: clock['now'])
+    monkeypatch.setattr(time, 'sleep', fake_sleep)
+
+    DaprRunner(tmp_path).start('--app-id demo-actor -- uvicorn demo:app', wait=3)
+
+    assert sleeps == [1, 1, 1]

tests/examples/test_demo_actor.py

@@ -31,7 +31,6 @@
 def test_demo_actor(dapr):
     dapr.start(
         '--app-id demo-actor --app-port 3000 -- uvicorn --port 3000 demo_actor_service:app',
-        wait=10,
     )
     client_output = dapr.run(
         '--app-id demo-client -- python3 demo_actor_client.py',

tests/examples/test_grpc_proxying.py

@@ -9,7 +9,6 @@
 def test_grpc_proxying(dapr):
     dapr.start(
         '--app-id invoke-receiver --app-protocol grpc --app-port 50051 --config config.yaml -- python3 invoke-receiver.py',
-        wait=5,
     )
     caller_output = dapr.run(
         '--app-id invoke-caller --dapr-grpc-port 50007 --config config.yaml -- python3 invoke-caller.py',

tests/examples/test_invoke_binding.py

@@ -14,6 +14,47 @@
     '{"id":3,"message":"hello world"}',
 ]
 
+KAFKA_TOPIC = 'sample'
+
+
+def _wait_for_kafka_topic(topic: str, timeout: float = 120) -> None:
+    """Polls the broker until the auto-created topic is listable.
+
+    ``docker compose up -d`` returns once containers are created, but the
+    wurstmeister Kafka image takes several seconds of broker registration
+    before it can serve metadata. Without this wait, daprd races the broker
+    and fails component init with "client has run out of available brokers".
+    """
+    list_topics_command = (
+        'docker',
+        'compose',
+        '-f',
+        './docker-compose-single-kafka.yml',
+        'exec',
+        '-T',
+        'kafka',
+        'kafka-topics.sh',
+        '--bootstrap-server',
+        'localhost:9092',
+        '--list',
+    )
+    deadline = time.monotonic() + timeout
+    last_output = ''
+    while time.monotonic() < deadline:
+        result = subprocess.run(
+            list_topics_command,
+            cwd=BINDING_DIR,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            timeout=30,
+        )
+        last_output = result.stdout
+        if result.returncode == 0 and topic in result.stdout.split():
+            return
+        time.sleep(1)
+    pytest.fail(f'Kafka topic {topic!r} not available after {timeout}s:\n{last_output}')
+
 
 @pytest.fixture()
 def kafka():
@@ -30,11 +71,7 @@ def kafka():
         output = (e.stdout or b'').decode(errors='replace')
         pytest.fail(f'Timed out starting Kafka:\n{output}')
 
-    # ``docker compose up -d`` returns once containers are created, but the
-    # wurstmeister Kafka image takes several seconds of broker registration
-    # before it can serve metadata. Without this wait, daprd races the broker
-    # and fails component init with "client has run out of available brokers".
-    time.sleep(20)
+    _wait_for_kafka_topic(KAFKA_TOPIC)
 
     yield
 
@@ -57,7 +94,6 @@ def test_invoke_binding(dapr, kafka):
     dapr.start(
         '--app-id receiver --app-protocol grpc --app-port 50051 '
         '--dapr-http-port 3500 --resources-path ./components -- python3 invoke-input-binding.py',
-        wait=5,
     )
 
     # Publish through the receiver's sidecar (both scripts are infinite,

tests/examples/test_invoke_custom_data.py

@@ -15,7 +15,6 @@
 def test_invoke_custom_data(dapr):
     dapr.start(
         '--app-id invoke-receiver --app-protocol grpc --app-port 50051 -- python3 invoke-receiver.py',
-        wait=5,
     )
     caller_output = dapr.run(
         '--app-id invoke-caller --app-protocol grpc -- python3 invoke-caller.py',

tests/examples/test_invoke_http.py

@@ -20,7 +20,6 @@ def test_invoke_http(dapr):
     dapr.start(
         '--app-id invoke-receiver --app-port 8088 --app-protocol http '
         '-- python3 invoke-receiver.py',
-        wait=5,
     )
     caller_output = dapr.run(
         '--app-id invoke-caller -- python3 invoke-caller.py',

tests/examples/test_invoke_simple.py

@@ -11,7 +11,6 @@ def test_invoke_simple(dapr):
     dapr.start(
         '--app-id invoke-receiver --app-protocol grpc --app-port 50051 '
         '--dapr-http-port 3500 -- python3 invoke-receiver.py',
-        wait=5,
     )
 
     # invoke-caller.py runs an infinite loop, so we invoke the method

tests/examples/test_jobs.py

@@ -41,10 +41,13 @@ def test_job_management(dapr):
 
 @pytest.mark.example_dir('jobs')
 def test_job_processing(dapr):
-    dapr.start(
+    # job_processing.py schedules its jobs ~5s after startup with 1-3s due
+    # times, so run until every expected line has appeared instead of
+    # stopping right after sidecar readiness.
+    output = dapr.run(
         '--app-id jobs-workflow --app-protocol grpc --app-port 50051 -- python3 job_processing.py',
-        wait=15,
+        timeout=60,
+        until=EXPECTED_PROCESSING,
     )
-    output = dapr.stop()
     for line in EXPECTED_PROCESSING:
         assert line in output, f'Missing in output: {line}'

tests/examples/test_pubsub_simple.py

@@ -25,7 +25,6 @@
 def test_pubsub_simple(dapr):
     dapr.start(
         '--app-id python-subscriber --app-protocol grpc --app-port 50051 -- python3 subscriber.py',
-        wait=5,
     )
     publisher_output = dapr.run(
         '--app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 '

tests/examples/test_pubsub_streaming.py

@@ -31,7 +31,6 @@
 def test_pubsub_streaming(dapr):
     dapr.start(
         '--app-id python-subscriber --app-protocol grpc -- python3 subscriber.py --topic=TOPIC_A1',
-        wait=5,
     )
     publisher_output = dapr.run(
         '--app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 '
@@ -50,7 +49,6 @@ def test_pubsub_streaming(dapr):
 def test_pubsub_streaming_handler(dapr):
     dapr.start(
         '--app-id python-subscriber --app-protocol grpc -- python3 subscriber-handler.py --topic=TOPIC_A2',
-        wait=5,
     )
     publisher_output = dapr.run(
         '--app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 '

tests/examples/test_pubsub_streaming_async.py

@@ -31,7 +31,6 @@
 def test_pubsub_streaming_async(dapr):
     dapr.start(
         '--app-id python-subscriber --app-protocol grpc -- python3 -u subscriber.py --topic=TOPIC_B1',
-        wait=5,
     )
     publisher_output = dapr.run(
         '--app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 '
@@ -50,7 +49,6 @@ def test_pubsub_streaming_async(dapr):
 def test_pubsub_streaming_async_handler(dapr):
     dapr.start(
         '--app-id python-subscriber --app-protocol grpc -- python3 -u subscriber-handler.py --topic=TOPIC_B2',
-        wait=5,
     )
     publisher_output = dapr.run(
         '--app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 '

tests/examples/test_state_store_query.py

@@ -41,7 +41,7 @@ def import_data(mongodb, dapr):
     reads these back through the same component. Writing raw documents with
     pymongo would skip that encoding and the query would return nothing.
     """
-    dapr.start('--app-id demo --dapr-http-port 3500 --resources-path components', wait=5)
+    dapr.start('--app-id demo --dapr-http-port 3500 --resources-path components')
     dataset = (EXAMPLES_DIR / 'state_store_query' / 'dataset.json').read_text()
     httpx.post(
         'http://localhost:3500/v1.0/state/statestore',

tests/examples/test_w3c_tracing.py

@@ -13,7 +13,6 @@
 def test_w3c_tracing(dapr):
     dapr.start(
         '--app-id invoke-receiver --app-protocol grpc --app-port 3001 -- python3 invoke-receiver.py',
-        wait=5,
     )
     caller_output = dapr.run(
         '--app-id invoke-caller --app-protocol grpc -- python3 invoke-caller.py',