refactor(publisher): move enable/disable decision into a make_publisher factory

Address review: rather than PubSubPublisher disabling itself, build publishers
through a make_publisher() factory that returns the existing NullPublisher when
publishing is off (PUBSUB_ENABLED false, or GCP credentials absent). Keeps the
transport class a plain sender and reuses the noop publisher we already have.

- New make_publisher() factory; the ~5 construction sites (cli/sinks.py,
  validation_result_exporter.py, ui_api/singletons.py) route through it.
- Missing credentials emit a one-time configuration.errors metric from the
  factory; the deliberate PUBSUB_ENABLED opt-out stays silent.
- Read the flag via config.get_bool('PUBSUB_ENABLED', True) (positive form) instead
  of a raw os.environ DISABLE_GCP_PUBSUB read.
- Move the shared credential check to lib/utils/gcp_credentials.py.
- Revert AsyncPubSubPublisher to a plain transport; it has no callers yet, so a
  factory/noop there would be dead code until it is wired into a sink.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TEcfpqhza3dPepZUVWw36X

Author: julietshen

CHANGELOG.md

@@ -31,7 +31,7 @@ For more information about each release including git tags and artifacts, see [R
 - Event stream shows sensible defaults so first-load isn't empty ([#297](https://github.com/roostorg/osprey/pull/297) by [@haileyok](https://github.com/haileyok))
 - Replace `react-scripts` with `rsbuild`/`rspack` for UI builds ([#235](https://github.com/roostorg/osprey/pull/235) by [@chimosky](https://github.com/chimosky))
 - Migrate from npm to pnpm via Corepack ([#252](https://github.com/roostorg/osprey/pull/252) by [@haileyok](https://github.com/haileyok))
-- Pub/Sub publishers degrade to a noop when GCP credentials are absent or when `DISABLE_GCP_PUBSUB` is set, so Osprey runs without GCP config instead of failing: `PubSubPublisher` ([#388](https://github.com/roostorg/osprey/pull/388)) and `AsyncPubSubPublisher` ([#411](https://github.com/roostorg/osprey/pull/411)) (by [@julietshen](https://github.com/julietshen))
+- Osprey runs without GCP config instead of failing: publishers are built through a `make_publisher` factory that returns a noop publisher when GCP credentials are absent or `PUBSUB_ENABLED` is false ([#388](https://github.com/roostorg/osprey/pull/388) by [@julietshen](https://github.com/julietshen))
 
 ### Fixed
 

docker-compose.override.yaml

@@ -0,0 +1,16 @@
+services:
+  osprey-ui-api:
+    environment:
+      - OSPREY_RULES_SUBMISSION_BACKEND=tangled
+      - OSPREY_TANGLED_HANDLE=julietshen.bsky.social
+      - OSPREY_TANGLED_REPO=julietshen.bsky.social/osprey-experiment
+      - OSPREY_TANGLED_REPO_DID=did:plc:55uxnkmuoiwwgyoqjbjo4z5t
+      - OSPREY_TANGLED_APP_PASSWORD=${OSPREY_TANGLED_APP_PASSWORD}
+      - OSPREY_RULES_BASE_BRANCH=main
+      - OSPREY_RULES_PATH_IN_REPO=example_rules
+      # Other backends' vars kept so switching back is a one-line change:
+      - OSPREY_GITLAB_PROJECT=julietshen/osprey-proof-of-concept
+      - OSPREY_GITLAB_TOKEN=${OSPREY_GITLAB_TOKEN}
+      - OSPREY_RULES_REPO=julietshen/osprey
+      - OSPREY_GITHUB_TOKEN=${OSPREY_GITHUB_TOKEN}
+      - OSPREY_RULES_LOCAL_PATH=/osprey/example_rules

osprey_async_worker/src/osprey/async_worker/lib/publisher.py

@@ -9,7 +9,6 @@
 
 from google.api_core.retry import Retry
 from google.cloud import pubsub_v1
-from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled
 from osprey.worker.lib.instruments import metrics
 from pydantic import BaseModel
 
@@ -31,12 +30,9 @@ class AsyncPubSubPublisher:
     Messages are buffered in an asyncio.Queue and flushed either when
     the batch reaches max_messages or after max_latency_seconds.
 
-    Degrades to noop mode when the DISABLE_GCP_PUBSUB env var is set, or when GCP
-    credentials cannot be resolved at construction time (e.g. local dev or adopter
-    environments without GCP). In noop mode no underlying client is built and the
-    publish paths return immediately. A one-time warning is logged at construction
-    so the inert state is visible, and missing credentials (unlike the deliberate
-    opt-out) also emit a startup `configuration.errors` metric.
+    This is a plain transport that assumes GCP is configured. A callers-side
+    factory (as on the sync side) should decide whether to build this or a noop
+    publisher once the async worker wires it into a sink.
     """
 
     def __init__(
@@ -49,25 +45,6 @@ def __init__(
         self._topic_path = f'projects/{project_id}/topics/{topic_id}'
         self._metric_tags = [f'project:{project_id}', f'topic:{topic_id}']
         self._flush_task: asyncio.Task[None] | None = None
-        if gcp_pubsub_disabled():
-            self._enabled = False
-            logger.warning(
-                'DISABLE_GCP_PUBSUB is set, AsyncPubSubPublisher disabled (project=%s, topic=%s)',
-                project_id,
-                topic_id,
-            )
-            return
-        self._enabled = gcp_credentials_available()
-        if not self._enabled:
-            logger.warning(
-                'GCP credentials not detected, AsyncPubSubPublisher running in noop mode (project=%s, topic=%s)',
-                project_id,
-                topic_id,
-            )
-            # Startup-only signal: missing credentials is a misconfiguration, unlike the
-            # deliberate DISABLE_GCP_PUBSUB opt-out above, which is silent.
-            metrics.increment('configuration.errors', tags=self._metric_tags + ['reason:gcp_credentials_missing'])
-            return
         self._client = pubsub_v1.PublisherClient(
             batch_settings=pubsub_v1.types.BatchSettings(max_messages=1),
         )
@@ -156,8 +133,6 @@ def publish(self, data: BaseModel) -> None:
 
     def publish_bytes(self, data: bytes) -> None:
         """Queue raw bytes for async batched publishing."""
-        if not self._enabled:
-            return
         self._ensure_started()
         try:
             self._queue.put_nowait(data)
@@ -167,8 +142,6 @@ def publish_bytes(self, data: bytes) -> None:
 
     async def stop(self) -> None:
         """Flush remaining messages and stop."""
-        if not self._enabled:
-            return
         if self._flush_task is not None:
             self._flush_task.cancel()
             try:

osprey_async_worker/src/osprey/async_worker/tests/test_publisher.py

@@ -2,18 +2,13 @@
 
 from unittest.mock import MagicMock, patch
 
-import pytest
 from google.api_core.exceptions import NotFound
-from osprey.async_worker.lib import publisher as publisher_module
 from osprey.async_worker.lib.publisher import _PUBLISH_RETRY, AsyncPubSubPublisher
 
 
 def _make_publisher():
-    """Return an enabled AsyncPubSubPublisher with a mocked PublisherClient."""
-    with (
-        patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=True),
-        patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient'),
-    ):
+    """Return an AsyncPubSubPublisher with a mocked PublisherClient."""
+    with patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient'):
         publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
     publisher._client = MagicMock()
     return publisher
@@ -53,59 +48,3 @@ def test_permanent_failure_metric_fires(mock_metrics):
     assert len(failure_calls) == 1
     assert failure_calls[0][0][0] == 'async_pubsub_publisher.publish.failure'
     assert f'error:{exc.__class__.__name__}' in failure_calls[0][1]['tags']
-
-
-def test_noops_when_creds_absent(caplog: pytest.LogCaptureFixture) -> None:
-    with (
-        patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False),
-        patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient') as client_cls,
-        patch('osprey.async_worker.lib.publisher.metrics') as mock_metrics,
-    ):
-        with caplog.at_level('WARNING', logger=publisher_module.logger.name):
-            publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
-    assert publisher._enabled is False
-    assert not client_cls.called
-    assert not hasattr(publisher, '_client')
-    assert 'noop mode' in caplog.text
-    assert 'project=proj' in caplog.text
-    assert 'topic=topic' in caplog.text
-    mock_metrics.increment.assert_called_once_with(
-        'configuration.errors',
-        tags=['project:proj', 'topic:topic', 'reason:gcp_credentials_missing'],
-    )
-
-
-def test_disabled_via_env(monkeypatch, caplog: pytest.LogCaptureFixture) -> None:
-    monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'true')
-    with (
-        patch('osprey.async_worker.lib.publisher.gcp_credentials_available') as cred_check,
-        patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient') as client_cls,
-        patch('osprey.async_worker.lib.publisher.metrics') as mock_metrics,
-    ):
-        with caplog.at_level('WARNING', logger=publisher_module.logger.name):
-            publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
-    assert publisher._enabled is False
-    assert not client_cls.called
-    # The opt-out short-circuits before probing credentials.
-    assert not cred_check.called
-    assert 'DISABLE_GCP_PUBSUB' in caplog.text
-    # A deliberate opt-out is not a misconfiguration, so no config-error metric.
-    mock_metrics.increment.assert_not_called()
-
-
-@patch('osprey.async_worker.lib.publisher.metrics')
-def test_publish_bytes_short_circuits_silently_when_disabled(mock_metrics) -> None:
-    with patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False):
-        publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
-        mock_metrics.increment.reset_mock()  # ignore the startup configuration.errors metric
-        publisher.publish_bytes(b'data')
-
-    mock_metrics.increment.assert_not_called()
-
-
-async def test_stop_short_circuits_when_disabled() -> None:
-    with patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False):
-        publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic')
-        await publisher.stop()
-
-    assert not hasattr(publisher, '_client')

osprey_worker/src/osprey/worker/cli/sinks.py

@@ -37,7 +37,7 @@
 from osprey.worker.lib.config import Config
 from osprey.worker.lib.osprey_engine import bootstrap_engine, bootstrap_engine_with_helpers, get_sources_provider
 from osprey.worker.lib.osprey_shared.logging import get_logger
-from osprey.worker.lib.publisher import PubSubPublisher
+from osprey.worker.lib.publisher import make_publisher
 from osprey.worker.lib.singletons import CONFIG, LABELS_PROVIDER
 from osprey.worker.lib.storage import postgres
 from osprey.worker.lib.storage.bigtable import osprey_bigtable
@@ -263,7 +263,7 @@ def run_bulk_label_sink(pooled: bool) -> None:
 
     analytics_pubsub_project_id = config.get_str('PUBSUB_DATA_PROJECT_ID', 'osprey-dev')
     analytics_pubsub_topic_id = config.get_str('PUBSUB_ANALYTICS_EVENT_TOPIC_ID', 'osprey-analytics')
-    analytics_publisher = PubSubPublisher(analytics_pubsub_project_id, analytics_pubsub_topic_id)
+    analytics_publisher = make_publisher(analytics_pubsub_project_id, analytics_pubsub_topic_id)
 
     def factory() -> BulkLabelSink:
         # NOTE: It's very important the input stream is created per-webhook sink
@@ -304,11 +304,11 @@ def rollback_bulk_label_effects(ctx: click.Context, task_id: int, include_ids_fr
 
     analytics_pubsub_project_id = config.get_str('PUBSUB_DATA_PROJECT_ID', 'osprey-dev')
     analytics_pubsub_topic_id = config.get_str('PUBSUB_ANALYTICS_EVENT_TOPIC_ID', 'osprey-analytics')
-    analytics_publisher = PubSubPublisher(analytics_pubsub_project_id, analytics_pubsub_topic_id)
+    analytics_publisher = make_publisher(analytics_pubsub_project_id, analytics_pubsub_topic_id)
 
     osprey_webhook_pubsub_project = config.get_str('PUBSUB_OSPREY_WEBHOOKS_PROJECT_ID', 'osprey-dev')
     osprey_webhook_pubsub_topic = config.get_str('PUBSUB_OSPREY_WEBHOOKS_TOPIC_ID', 'osprey-webhooks')
-    webhooks_publisher = PubSubPublisher(osprey_webhook_pubsub_project, osprey_webhook_pubsub_topic)
+    webhooks_publisher = make_publisher(osprey_webhook_pubsub_project, osprey_webhook_pubsub_topic)
 
     task = BulkLabelTask.get_one(task_id)
     if task is None:

osprey_worker/src/osprey/worker/lib/data_exporters/validation_result_exporter.py

@@ -5,7 +5,7 @@
     ValidateExperiments,
 )
 from osprey.worker.lib.data_exporters.models import ospreyExperimentMetadataUpdate
-from osprey.worker.lib.publisher import BasePublisher, PubSubPublisher
+from osprey.worker.lib.publisher import BasePublisher, make_publisher
 from osprey.worker.lib.singletons import CONFIG
 
 
@@ -56,4 +56,4 @@ def get_validation_result_exporter() -> BaseValidationResultExporter:
 
     pubsub_project_id = config.get_str('PUBSUB_DATA_PROJECT_ID', 'osprey-dev')
     pubsub_topic_id = config.get_str('PUBSUB_ANALYTICS_EVENT_TOPIC_ID', 'osprey-analytics')
-    return ExperimentValidationResultExporter(publisher=PubSubPublisher(pubsub_project_id, pubsub_topic_id))
+    return ExperimentValidationResultExporter(publisher=make_publisher(pubsub_project_id, pubsub_topic_id))

osprey_worker/src/osprey/worker/lib/publisher.py

@@ -3,9 +3,9 @@
 from typing import TypeVar
 
 from google.cloud import pubsub_v1
-from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled
 from osprey.worker.lib.instruments import metrics
 from osprey.worker.lib.pubsub.publisher_client import BatchPubsubPublisherClient
+from osprey.worker.lib.utils.gcp_credentials import gcp_credentials_available
 from pydantic import BaseModel
 
 _PydanticModelT = TypeVar('_PydanticModelT', bound=BaseModel)
@@ -46,12 +46,9 @@ def stop(self) -> None:
 class PubSubPublisher(BasePublisher):
     """Publishes Pydantic models to a Google Cloud Pub/Sub topic.
 
-    Degrades to noop mode when the DISABLE_GCP_PUBSUB env var is set, or when GCP
-    credentials cannot be resolved at construction time (e.g. local dev or adopter
-    environments without GCP). In noop mode no underlying client is built and
-    publish() and stop() return immediately. A one-time warning is logged at
-    construction so the inert state is visible, and missing credentials (unlike the
-    deliberate opt-out) also emit a startup `configuration.errors` metric.
+    This is a plain transport: it assumes GCP is configured and builds its client
+    eagerly. Callers that may run without GCP should build publishers via
+    make_publisher(), which hands back a NullPublisher when publishing is off.
     """
 
     def __init__(
@@ -68,34 +65,15 @@ def __init__(
             topic_id=topic_id,
         )
         self._project = project_id
-        self._raise_on_error = raise_on_error
-        self._tags = [f'project:{self._project}', f'topic:{self._topic_name}']
-        if gcp_pubsub_disabled():
-            self._enabled = False
-            logger.warning(
-                'DISABLE_GCP_PUBSUB is set, PubSubPublisher disabled (project=%s, topic=%s)',
-                project_id,
-                topic_id,
-            )
-            return
-        self._enabled = gcp_credentials_available()
-        if not self._enabled:
-            logger.warning(
-                'GCP credentials not detected, PubSubPublisher running in noop mode (project=%s, topic=%s)',
-                project_id,
-                topic_id,
-            )
-            # Startup-only signal: missing credentials is a misconfiguration, unlike the
-            # deliberate DISABLE_GCP_PUBSUB opt-out above, which is silent.
-            metrics.increment('configuration.errors', tags=self._tags + ['reason:gcp_credentials_missing'])
-            return
         batch_settings = pubsub_v1.types.BatchSettings(
             max_bytes=max_bytes,  # default 1MB
             max_messages=max_messages,  # default 100 messages
             max_latency=max_latency,
         )
         # self._publisher = pubsub_v1.PublisherClient(batch_settings=batch_settings)
         self._publisher = BatchPubsubPublisherClient(batch_settings=batch_settings)
+        self._raise_on_error = raise_on_error
+        self._tags = [f'project:{self._project}', f'topic:{self._topic_name}']
 
     def prepare_data(self, data: _PydanticModelT) -> bytes:
         """
@@ -105,8 +83,6 @@ def prepare_data(self, data: _PydanticModelT) -> bytes:
         return data.json(exclude_none=True).encode()
 
     def publish(self, data: _PydanticModelT, attributes: dict[str, str] | None = None) -> None:
-        if not self._enabled:
-            return
         if attributes is None:
             attributes = {}
 
@@ -124,8 +100,6 @@ def publish(self, data: _PydanticModelT, attributes: dict[str, str] | None = Non
         metrics.increment(f'{self.__class__.__name__}.publisher.success', tags=self._tags)
 
     def stop(self) -> None:
-        if not self._enabled:
-            return
         self._publisher.stop()
 
 
@@ -140,3 +114,41 @@ def publish(self, data: str, attributes: dict[str, str] | None = None) -> None:
             attributes = {}
 
         super().publish(data, attributes)  # type: ignore[type-var]
+
+
+def make_publisher(
+    project_id: str,
+    topic_id: str,
+    raise_on_error: bool = False,
+    max_bytes: int = 2000000,
+    max_messages: int = 250,
+    max_latency: float = 1.0,
+) -> BasePublisher:
+    """Build a Pub/Sub publisher, or a NullPublisher when publishing is off.
+
+    Publishing is off when PUBSUB_ENABLED is false, or when GCP credentials cannot
+    be resolved (e.g. local dev or adopter environments without GCP). Missing
+    credentials are a misconfiguration, so they also emit a one-time
+    configuration.errors metric; the deliberate PUBSUB_ENABLED opt-out is silent.
+    """
+    # Imported here to avoid pulling the config/singletons stack into this low-level module.
+    from osprey.worker.lib.singletons import CONFIG
+
+    if not CONFIG.instance().get_bool('PUBSUB_ENABLED', True):
+        logger.warning('PUBSUB_ENABLED is false, publishing disabled (project=%s, topic=%s)', project_id, topic_id)
+        return NullPublisher()
+    if not gcp_credentials_available():
+        logger.warning('GCP credentials not detected, publishing disabled (project=%s, topic=%s)', project_id, topic_id)
+        metrics.increment(
+            'configuration.errors',
+            tags=[f'project:{project_id}', f'topic:{topic_id}', 'reason:gcp_credentials_missing'],
+        )
+        return NullPublisher()
+    return PubSubPublisher(
+        project_id,
+        topic_id,
+        raise_on_error=raise_on_error,
+        max_bytes=max_bytes,
+        max_messages=max_messages,
+        max_latency=max_latency,
+    )

osprey_worker/src/osprey/worker/lib/tests/test_gcp_credentials.py

@@ -3,8 +3,8 @@
 
 import pytest
 from google.auth.exceptions import DefaultCredentialsError
-from osprey.worker.lib import gcp_credentials
-from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled
+from osprey.worker.lib.utils import gcp_credentials
+from osprey.worker.lib.utils.gcp_credentials import gcp_credentials_available
 
 
 @pytest.fixture(autouse=True)
@@ -30,12 +30,3 @@ def test_result_is_cached() -> None:
         gcp_credentials_available()
         gcp_credentials_available()
     assert default_mock.call_count == 1
-
-
-def test_pubsub_disabled_reads_env(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.delenv('DISABLE_GCP_PUBSUB', raising=False)
-    assert gcp_pubsub_disabled() is False
-    monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'true')
-    assert gcp_pubsub_disabled() is True
-    monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'false')
-    assert gcp_pubsub_disabled() is False