feldera · gz · Aug 4, 2026 · Aug 4, 2026 · Aug 4, 2026 · Aug 4, 2026
diff --git a/.github/workflows/test-integration-runtime.yml b/.github/workflows/test-integration-runtime.yml
@@ -79,3 +79,15 @@ jobs:
           KAFKA_BOOTSTRAP_SERVERS: ${{ vars.CI_KAFKA_BOOTSTRAP }}
           SCHEMA_REGISTRY_URL: ${{ vars.CI_SCHEMA_REGISTRY }}
           RUN_ID: 1 # we use this to run a single variant of the Kafka tests in test_kafka_avro.py
+
+      # Test teardown only runs while pytest is alive. A cancel sentinel firing
+      # for some other job, a runner eviction or a job timeout leaves tests
+      # running on the shared instance to burn compute.
+      # This step reclaims the pipelines this run named after itself
+      - name: Stop pipelines this run leaked
+        if: ${{ always() && vars.CI_DRY_RUN != 'true' }}
+        timeout-minutes: 10
+        run: uv run --locked python -m tests.stop_ci_run_pipelines
+        working-directory: python
+        env:
+          PYTHONPATH: ${{ github.workspace }}/python
diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py
@@ -99,18 +99,19 @@ def _token_expiry(token: str) -> float:
         return time.time()
 
 
-def _feldera_credential():
-    """Bearer credential for the test client.
-
-    Under GitHub Actions the credential is an ID token that expires well inside
-    a long test run, so hand the SDK the callable rather than a token: it
-    re-resolves per request and retries once on 401. A configured OIDC login
-    flow takes precedence over both.
+def feldera_bearer_token() -> Optional[str]:
+    """The bearer token to send with a request issued now.
+
+    A configured OIDC login flow wins, then a GitHub Actions ID token, then the
+    static API key. None when nothing is configured, which is how a local
+    instance without authentication runs. Callers that build their own requests
+    resolve this per request: under Actions the token expires well inside a test
+    run, and one read at import time starts returning 401 partway through.
     """
     if os.environ.get("OIDC_TEST_ISSUER"):
         return _get_effective_api_key()
     if os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL"):
-        return _github_oidc_token
+        return _github_oidc_token()
     return API_KEY
 
 
@@ -155,9 +156,16 @@ def __init__(self):
 
     def _ensure(self):
         if self._client is None:
+            # Under Actions the token expires inside a run, so the SDK gets the
+            # resolver itself: it re-resolves per request and retries once on
+            # 401. Elsewhere the credential is fixed for the process.
             self._client = FelderaClient(
                 connection_timeout=10,
-                api_key=_feldera_credential(),
+                api_key=(
+                    feldera_bearer_token
+                    if os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL")
+                    else feldera_bearer_token()
+                ),
             )
         return self._client
 
@@ -287,6 +295,42 @@ def unique_pipeline_name(base_name: str) -> str:
     return name
 
 
+# Teardown must not block on one wedged pipeline: the SDK otherwise polls for
+# `Stopped` forever, and when the runner finally kills the job every pipeline
+# the run started keeps consuming compute on the shared instance.
+RECLAIM_TIMEOUT_SECONDS = 60.0
+
+
+def reclaim_pipeline(name: str, delete: bool = False) -> List[str]:
+    """Force-stop `name`, clear its storage, and optionally delete it.
+
+    Every step runs even when the one before it raised: a pipeline that never
+    reports `Stopped` can still release its storage, and abandoning the rest on
+    the first error leaves them running on a shared instance. Returns one
+    message per failed step, empty when the pipeline is fully reclaimed, so the
+    caller decides whether a cleanup failure is worth failing a test over.
+    """
+    failures = []
+
+    try:
+        TEST_CLIENT.stop_pipeline(name, force=True, timeout_s=RECLAIM_TIMEOUT_SECONDS)
+    except Exception as error:
+        failures.append(f"{name}: stop: {error}")
+
+    try:
+        TEST_CLIENT.clear_storage(name, timeout_s=RECLAIM_TIMEOUT_SECONDS)
+    except Exception as error:
+        failures.append(f"{name}: clear storage: {error}")
+
+    if delete:
+        try:
+            TEST_CLIENT.delete_pipeline(name)
+        except Exception as error:
+            failures.append(f"{name}: delete: {error}")
+
+    return failures
+
+
 def enterprise_only(fn):
     fn._enterprise_only = True
     return unittest.skipUnless(

diff --git a/python/tests/README.md b/python/tests/README.md
@@ -111,3 +111,21 @@ after the test completed.
 def test_some_property(pipeline_name):
     pass
 ```
+
+## Cleaning up after a run
+
+Teardown only runs while the test process is alive. A cancelled workflow run, an
+evicted runner or a job timeout kills pytest outright, and every pipeline that
+was running at that moment keeps consuming compute on the shared CI instance.
+
+`tests/stop_ci_run_pipelines.py` sweeps up what such a run left behind. It
+matches the prefix `unique_pipeline_name` stamps on every test pipeline.
+
+```bash
+cd python
+PYTHONPATH=`pwd` uv run python -m tests.stop_ci_run_pipelines --prefix <sha5>_
+```
+
+Whatever cleanup a test does itself should go through
+`feldera.testutils.reclaim_pipeline`: it bounds every wait, and it clears
+storage even when the stop failed instead of abandoning a running pipeline.
diff --git a/python/tests/platform/TEST_AUTH.md b/python/tests/platform/TEST_AUTH.md
@@ -136,8 +136,13 @@ uv run pytest tests/platform -v
 The platform tests automatically select authentication in this order:
 
 1. **OIDC Token** (if all OIDC environment variables are set)
-2. **API Key** (if `FELDERA_API_KEY` is set)
-3. **No Authentication** (fallback for development)
+2. **GitHub Actions ID token** (if `ACTIONS_ID_TOKEN_REQUEST_URL` is set)
+3. **API Key** (if `FELDERA_API_KEY` is set)
+4. **No Authentication** (fallback for development)
+
+`feldera.testutils.feldera_bearer_token()` resolves this order. Call it per
+request rather than caching what it returns: an ID token expires well inside a
+long suite, and the function re-mints one shortly before its expiry.
 
 ## CI/CD Configuration
 

diff --git a/python/tests/platform/helper.py b/python/tests/platform/helper.py
@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 import unittest
 from http import HTTPStatus
@@ -21,11 +22,14 @@
 
 import pytest
 import requests
-from feldera.testutils import FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS
-from feldera.testutils_oidc import get_oidc_test_helper
+from feldera.testutils import (
+    FELDERA_TEST_NUM_HOSTS,
+    FELDERA_TEST_NUM_WORKERS,
+    feldera_bearer_token,
+    reclaim_pipeline,
+)
 
 from tests import (
-    API_KEY,
     BASE_URL,
     FELDERA_REQUESTS_VERIFY,
     TEST_CLIENT,
@@ -35,19 +39,18 @@
 
 API_PREFIX = "/v0"
 
+logger = logging.getLogger(__name__)
+
 
 def _base_headers() -> Dict[str, str]:
     headers = {
         "Accept": "application/json",
     }
 
-    # Try OIDC authentication first, then fall back to API_KEY
-    oidc_helper = get_oidc_test_helper()
-    if oidc_helper is not None:
-        token = oidc_helper.obtain_access_token()
+    # Resolved per request: a GitHub Actions ID token expires mid-suite.
+    token = feldera_bearer_token()
+    if token:
         headers["Authorization"] = f"Bearer {token}"
-    elif API_KEY:
-        headers["Authorization"] = f"Bearer {API_KEY}"
 
     return headers
 
@@ -162,22 +165,25 @@ def pause_pipeline(name: str, wait: bool = True):
     TEST_CLIENT.pause_pipeline(name, wait=wait)
 
 
-def stop_pipeline(name: str, force: bool = True, wait: bool = True):
-    TEST_CLIENT.stop_pipeline(name, force=force, wait=wait)
+def stop_pipeline(
+    name: str, force: bool = True, wait: bool = True, timeout_s: float | None = None
+):
+    TEST_CLIENT.stop_pipeline(name, force=force, wait=wait, timeout_s=timeout_s)
 
 
-def clear_pipeline(name: str, wait: bool = True):
+def clear_pipeline(name: str, wait: bool = True, timeout_s: float | None = None):
     r = post_no_body(f"{API_PREFIX}/pipelines/{name}/clear")
     if wait and r.status_code == HTTPStatus.ACCEPTED:
-        TEST_CLIENT.clear_storage(name)
+        TEST_CLIENT.clear_storage(name, timeout_s=timeout_s)
     return r
 
 
 def reset_pipeline(name: str):
+    """Force-stop `name` and clear its storage."""
     if get_pipeline(name, "status").status_code != HTTPStatus.OK:
         return
-    stop_pipeline(name, force=True)
-    clear_pipeline(name)
+    for failure in reclaim_pipeline(name):
+        logger.warning("pipeline cleanup: %s", failure)
 
 
 def delete_pipeline(name: str):

diff --git a/python/tests/runtime_aggtest/aggtst_base.py b/python/tests/runtime_aggtest/aggtst_base.py
@@ -9,7 +9,11 @@
 from feldera.enums import CompilationProfile
 from feldera.rest.errors import FelderaAPIError
 from feldera.runtime_config import Resources, RuntimeConfig
-from feldera.testutils import FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS
+from feldera.testutils import (
+    FELDERA_TEST_NUM_WORKERS,
+    FELDERA_TEST_NUM_HOSTS,
+    reclaim_pipeline,
+)
 from tests import TEST_CLIENT, unique_pipeline_name
 
 JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | str | int | float | bool | None
@@ -325,8 +329,8 @@ def run_pipeline(
                         raise
 
             if pipeline is not None:
-                pipeline.stop(force=True)
-                pipeline.delete(True)
+                for failure in reclaim_pipeline(pipeline_name, delete=True):
+                    print(f"WARNING: {failure}")
 
     def assert_expected_error(self, obj: SqlObject, actual_exception: Exception):
         """Validate the error produced by the failing pipeline with the expected error type"""

diff --git a/python/tests/shared_test_pipeline.py b/python/tests/shared_test_pipeline.py
@@ -1,14 +1,18 @@
+import logging
 import unittest
 
 from feldera import Pipeline, PipelineBuilder
 from feldera.runtime_config import RuntimeConfig
 from feldera.testutils import (
     FELDERA_TEST_NUM_HOSTS,
     FELDERA_TEST_NUM_WORKERS,
+    reclaim_pipeline,
     unique_pipeline_name,
 )
 from tests import TEST_CLIENT
 
+logger = logging.getLogger(__name__)
+
 
 def sql(text_or_iterable):
     """
@@ -59,34 +63,33 @@ def setUpClass(cls):
             ).create_or_replace()
 
     def setUp(self):
-        p = PipelineBuilder(
-            self.client,
-            unique_pipeline_name(self._testMethodName),
-            sql=self.ddl,
-            runtime_config=RuntimeConfig(
-                workers=FELDERA_TEST_NUM_WORKERS,
-                hosts=FELDERA_TEST_NUM_HOSTS,
-                logging="debug",
-            ),
-        ).create_or_replace()
-        self.p = p
+        self._owned_pipelines = []
+        self.p = self._build_pipeline(self._testMethodName)
 
     def tearDown(self):
-        self.p.stop(force=True)
-        self.p.clear_storage()
+        """Force-stop and clear every pipeline the test owns."""
+        for pipeline in self._owned_pipelines:
+            for failure in reclaim_pipeline(pipeline.name):
+                logger.warning("pipeline teardown: %s", failure)
 
     @property
     def pipeline(self) -> Pipeline:
         return self.p
 
     def new_pipeline_with_suffix(self, suffix: str) -> Pipeline:
-        return PipelineBuilder(
+        return self._build_pipeline(f"{self._testMethodName}_{suffix}")
+
+    def _build_pipeline(self, base_name: str) -> Pipeline:
+        """Create a pipeline over the class DDL and hand it to `tearDown`."""
+        pipeline = PipelineBuilder(
             self.client,
-            unique_pipeline_name(f"{self._testMethodName}_{suffix}"),
+            unique_pipeline_name(base_name),
             sql=self.ddl,
             runtime_config=RuntimeConfig(
                 workers=FELDERA_TEST_NUM_WORKERS,
                 hosts=FELDERA_TEST_NUM_HOSTS,
                 logging="debug",
             ),
         ).create_or_replace()
+        self._owned_pipelines.append(pipeline)
+        return pipeline