Silence error logging when exception is silenced

clients/python/src/taskbroker_client/worker/workerchild.py

@@ -510,17 +510,19 @@ def check_task_future_completion(
                         next_state = TASK_ACTIVATION_STATUS_RETRY
                     elif retry.max_attempts_reached(inflight.activation.retry_state):
                         with sentry_sdk.isolation_scope() as scope:
-                            retry_error = NoRetriesRemainingError(
-                                f"{inflight.activation.taskname} has consumed all of its retries"
-                            )
-                            retry_error.__cause__ = err
-                            scope.fingerprint = [
-                                "taskworker.no_retries_remaining",
-                                inflight.activation.namespace,
-                                inflight.activation.taskname,
-                            ]
-                            scope.set_transaction_name(inflight.activation.taskname)
-                            sentry_sdk.capture_exception(retry_error)
+                            # Only report to Sentry when the underlying error is not silenced.
+                            if should_capture_error:
+                                retry_error = NoRetriesRemainingError(
+                                    f"{inflight.activation.taskname} has consumed all of its retries"
+                                )
+                                retry_error.__cause__ = err
+                                scope.fingerprint = [
+                                    "taskworker.no_retries_remaining",
+                                    inflight.activation.namespace,
+                                    inflight.activation.taskname,
+                                ]
+                                scope.set_transaction_name(inflight.activation.taskname)
+                                sentry_sdk.capture_exception(retry_error)
                             # Emit a structured worker log without logger.exception so this
                             # branch does not create a second Sentry error event.
                             _log_task_retry_exhausted(

clients/python/tests/worker/test_worker.py

@@ -1797,6 +1797,62 @@ def test_child_process_expected_ignored_exception_max_attempts(mock_capture: moc
     assert mock_capture.call_count == 0
 
 
+@mock.patch("taskbroker_client.worker.workerchild.logger")
+@mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception")
+def test_child_process_silenced_exception_max_attempts(
+    mock_capture: mock.Mock, mock_logger: mock.Mock
+) -> None:
+    """Silenced exceptions do not raise on retry exhaustion."""
+    activation = InflightTaskActivation(
+        host="localhost:50051",
+        receive_timestamp=0,
+        activation=TaskActivation(
+            id="silenced-max-attempts",
+            taskname="examples.will_fail_with_silenced_ignored_exception",
+            namespace="examples",
+            parameters_bytes=msgpack.packb({"args": [], "kwargs": {}}, use_bin_type=True),
+            processing_deadline_duration=2,
+            retry_state=RetryState(
+                # No retries left
+                attempts=1,
+                max_attempts=2,
+                on_attempts_exceeded=ON_ATTEMPTS_EXCEEDED_DISCARD,
+            ),
+        ),
+    )
+    todo: queue.Queue[InflightTaskActivation] = queue.Queue()
+    processed: queue.Queue[ProcessingResult] = queue.Queue()
+    shutdown = Event()
+
+    todo.put(activation)
+    child_process(
+        "examples.app:app",
+        todo,
+        processed,
+        shutdown,
+        max_task_count=1,
+        processing_pool_name="test",
+        process_type="fork",
+        skip_awaiting_futures=False,
+        future_checking_frequency=0.1,
+    )
+
+    assert todo.empty()
+    result = processed.get()
+    assert result.task_id == activation.activation.id
+    assert result.status == TASK_ACTIVATION_STATUS_FAILURE
+
+    # Silenced error: no Sentry event even though retries are exhausted.
+    assert mock_capture.call_count == 0
+
+    # The structured retry-exhausted log still fires (without logger.exception).
+    mock_logger.exception.assert_not_called()
+    mock_logger.warning.assert_called_once()
+    args, kwargs = mock_logger.warning.call_args
+    assert args[0] == "taskworker.task.retry_exhausted"
+    assert kwargs["extra"]["exception_type"] == "RuntimeError"
+
+
 @mock.patch("taskbroker_client.worker.workerchild.logger")
 def test_child_process_retry_on_deadline_exceeded(mock_logger: mock.Mock) -> None:
     todo: queue.Queue[InflightTaskActivation] = queue.Queue()