feat(async): WARN log when provider showing degraded performance

Diagnostic A/Bs against build.nvidia.com showed runs failing silently
under provider degradation - no log indication that retryable errors
were piling up until the early-shutdown gate fired (or, post-fix,
until salvage exhaustion). Surfacing this earlier helps users
distinguish "DataDesigner is broken" from "the upstream provider is
slow today."

Tracks a separate sliding window over retryable-vs-not for every task
outcome (independent of the early-shutdown gate's window) and emits a
throttled WARN when the rolling fraction crosses the threshold.

Refs #575.

Author: andreatgretel

packages/data-designer-engine/src/data_designer/engine/dataset_builders/async_scheduler.py

@@ -47,6 +47,13 @@
 DEFAULT_TASK_POOL_SIZE: int = 256
 LLM_WAIT_POOL_MULTIPLIER: int = 2
 
+# Degraded-provider WARN: emit at most one warning per interval when the
+# rolling fraction of retryable errors exceeds the threshold. Distinct from
+# the early-shutdown gate (which fires on non-retryable errors).
+DEGRADED_WARN_RATE: float = 0.5
+DEGRADED_WARN_WINDOW: int = 20
+DEGRADED_WARN_INTERVAL_S: float = 60.0
+
 _RETRYABLE_MODEL_ERRORS = (
     ModelRateLimitError,
     ModelTimeoutError,
@@ -105,6 +112,9 @@ def __init__(
         shutdown_error_rate: float = 0.5,
         shutdown_error_window: int = 10,
         disable_early_shutdown: bool = False,
+        degraded_warn_rate: float = DEGRADED_WARN_RATE,
+        degraded_warn_window: int = DEGRADED_WARN_WINDOW,
+        degraded_warn_interval_s: float = DEGRADED_WARN_INTERVAL_S,
         trace: bool = False,
         num_records: int = 0,
         buffer_size: int = 0,
@@ -177,6 +187,14 @@ def __init__(
         self._recent_outcomes: deque[bool] = deque(maxlen=shutdown_error_window)
         self._all_rgs_admitted = False
 
+        # Degraded-provider WARN: separate window tracking retryable-vs-not for
+        # every outcome (success or failure), throttled to one log per interval.
+        self._degraded_warn_rate = degraded_warn_rate
+        self._degraded_warn_window = degraded_warn_window
+        self._degraded_warn_interval_s = degraded_warn_interval_s
+        self._recent_retryable: deque[bool] = deque(maxlen=degraded_warn_window)
+        self._last_degraded_warn_at: float = 0.0
+
         # Pre-compute row-group sizes for O(1) lookup
         self._rg_size_map: dict[int, int] = dict(row_groups)
 
@@ -606,6 +624,33 @@ def _check_error_rate(self, *, success: bool) -> None:
         if errors / self._shutdown_error_window >= self._shutdown_error_rate:
             self._early_shutdown = True
 
+    def _record_retryable_outcome(self, *, retryable: bool) -> None:
+        """Track retryable-error rate and emit a throttled WARN under provider degradation.
+
+        Distinct from ``_check_error_rate``: every outcome (success or failure)
+        feeds this window so the rate reflects the provider's overall health, not
+        just the error mix. Only retryable errors (rate-limit, timeout, 5xx,
+        connection) count toward the rate; non-retryable failures register as 0.
+        """
+        if self._degraded_warn_window <= 0:
+            return
+        self._recent_retryable.append(retryable)
+        if len(self._recent_retryable) < self._degraded_warn_window:
+            return
+        rate = sum(self._recent_retryable) / self._degraded_warn_window
+        if rate < self._degraded_warn_rate:
+            return
+        now = time.monotonic()
+        if now - self._last_degraded_warn_at < self._degraded_warn_interval_s:
+            return
+        self._last_degraded_warn_at = now
+        pct = int(round(rate * 100))
+        logger.warning(
+            f"Provider showing degraded performance: {pct}% of last {self._degraded_warn_window} "
+            "task outcomes were retryable errors (rate-limit, timeout, 5xx, connection). "
+            "Run may take longer than expected; salvage will retry these."
+        )
+
     async def _dispatch_seeds(self, rg_id: int, rg_size: int) -> None:
         """Dispatch from_scratch tasks for a row group."""
         self._rg_states[rg_id].seeds_dispatched = True
@@ -730,6 +775,7 @@ async def _execute_task_inner_impl(self, task: Task) -> None:
                     self._tracker.mark_cell_complete(col, task.row_group, task.row_index)
 
             self._check_error_rate(success=True)
+            self._record_retryable_outcome(retryable=False)
             if self._reporter:
                 if cell_skipped:
                     self._reporter.record_skipped(task.column)
@@ -746,6 +792,7 @@ async def _execute_task_inner_impl(self, task: Task) -> None:
             # and would otherwise trip the gate even when salvage could recover.
             if not retryable:
                 self._check_error_rate(success=False)
+            self._record_retryable_outcome(retryable=retryable)
             if not retryable and self._reporter:
                 self._reporter.record_failure(task.column)
             if self._trace and trace:

packages/data-designer-engine/tests/engine/dataset_builders/test_async_scheduler.py

@@ -919,6 +919,119 @@ async def test_retryable_errors_do_not_trigger_early_shutdown(
     assert tracker.is_row_group_complete(0, 10, ["seed", "col"])
 
 
+@pytest.mark.asyncio(loop_scope="session")
+async def test_degraded_provider_warn_fires_above_threshold(caplog: pytest.LogCaptureFixture) -> None:
+    """When >= threshold of recent outcomes are retryable errors, a WARN log fires."""
+    provider = _mock_provider()
+    configs = [
+        SamplerColumnConfig(name="seed", sampler_type=SamplerType.CATEGORY, params={"values": ["A"]}),
+        LLMTextColumnConfig(name="col", prompt="{{ seed }}", model_alias=MODEL_ALIAS),
+    ]
+    strategies = {
+        "seed": GenerationStrategy.FULL_COLUMN,
+        "col": GenerationStrategy.CELL_BY_CELL,
+    }
+    # 6 retryable failures across 10 cells + their successful retries → ~6/16 retryable.
+    # Set window to 8 and threshold to 0.5 so the WARN can fire.
+    generators = {
+        "seed": MockSeedGenerator(config=_expr_config("seed"), resource_provider=provider),
+        "col": MockRetryableErrorGenerator(
+            config=_expr_config("col"),
+            resource_provider=provider,
+            error_factory=lambda: ModelTimeoutError("read timeout"),
+            retryable_failures=6,
+        ),
+    }
+
+    graph = ExecutionGraph.create(configs, strategies)
+    row_groups = [(0, 10)]
+    tracker = CompletionTracker.with_graph(graph, row_groups)
+
+    storage = MagicMock()
+    storage.dataset_name = "test"
+    storage.get_file_paths.return_value = {}
+    storage.write_batch_to_parquet_file.return_value = "/fake.parquet"
+    storage.move_partial_result_to_final_file_path.return_value = "/fake_final.parquet"
+    buffer_mgr = RowGroupBufferManager(storage)
+
+    scheduler = AsyncTaskScheduler(
+        generators=generators,
+        graph=graph,
+        tracker=tracker,
+        row_groups=row_groups,
+        buffer_manager=buffer_mgr,
+        degraded_warn_rate=0.5,
+        degraded_warn_window=8,
+        degraded_warn_interval_s=0.0,
+    )
+    with caplog.at_level("WARNING"):
+        await scheduler.run()
+
+    degraded_msgs = [r for r in caplog.records if "degraded performance" in r.getMessage()]
+    assert degraded_msgs, "expected a 'degraded performance' WARN to be emitted"
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_degraded_provider_warn_throttled(caplog: pytest.LogCaptureFixture) -> None:
+    """Successive degraded windows within the throttle interval emit only one WARN."""
+    provider = _mock_provider()
+    configs = [
+        SamplerColumnConfig(name="seed", sampler_type=SamplerType.CATEGORY, params={"values": ["A"]}),
+        LLMTextColumnConfig(name="col", prompt="{{ seed }}", model_alias=MODEL_ALIAS),
+    ]
+    strategies = {
+        "seed": GenerationStrategy.FULL_COLUMN,
+        "col": GenerationStrategy.CELL_BY_CELL,
+    }
+    generators = {
+        "seed": MockSeedGenerator(config=_expr_config("seed"), resource_provider=provider),
+        "col": MockRetryableErrorGenerator(
+            config=_expr_config("col"),
+            resource_provider=provider,
+            error_factory=lambda: ModelTimeoutError("read timeout"),
+            retryable_failures=8,
+        ),
+    }
+
+    graph = ExecutionGraph.create(configs, strategies)
+    row_groups = [(0, 12)]
+    tracker = CompletionTracker.with_graph(graph, row_groups)
+
+    storage = MagicMock()
+    storage.dataset_name = "test"
+    storage.get_file_paths.return_value = {}
+    storage.write_batch_to_parquet_file.return_value = "/fake.parquet"
+    storage.move_partial_result_to_final_file_path.return_value = "/fake_final.parquet"
+    buffer_mgr = RowGroupBufferManager(storage)
+
+    scheduler = AsyncTaskScheduler(
+        generators=generators,
+        graph=graph,
+        tracker=tracker,
+        row_groups=row_groups,
+        buffer_manager=buffer_mgr,
+        degraded_warn_rate=0.5,
+        degraded_warn_window=4,
+        degraded_warn_interval_s=3600.0,
+    )
+    with caplog.at_level("WARNING"):
+        await scheduler.run()
+
+    degraded_msgs = [r for r in caplog.records if "degraded performance" in r.getMessage()]
+    assert len(degraded_msgs) == 1, f"expected exactly one throttled WARN, got {len(degraded_msgs)}"
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_degraded_provider_warn_silent_under_threshold(caplog: pytest.LogCaptureFixture) -> None:
+    """Healthy runs (no errors) never emit the degraded-provider WARN."""
+    scheduler, _tracker = _build_simple_pipeline(num_records=5)
+    with caplog.at_level("WARNING"):
+        await scheduler.run()
+
+    degraded_msgs = [r for r in caplog.records if "degraded performance" in r.getMessage()]
+    assert not degraded_msgs
+
+
 @pytest.mark.asyncio(loop_scope="session")
 async def test_scheduler_on_before_checkpoint_callback() -> None:
     """on_before_checkpoint is called before each row group is checkpointed."""