Skip to content

feat(iswf): Adds silenced_exceptions parameter to tasks, exposes this and report_timeout_errors in task registration #608

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
GabeVillalobos merged 7 commits into from
May 4, 2026
Merged

feat(iswf): Adds silenced_exceptions parameter to tasks, exposes this and report_timeout_errors in task registration #608

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
498c91f
Adds additional task parameter to silence swaths of exceptions
GabeVillalobos Apr 27, 2026
f37ed8f
removes extraneous init param
GabeVillalobos Apr 27, 2026
7f580b4
Addresses comment from cursorbot
GabeVillalobos Apr 28, 2026
3e53af7
Updates silence param name, adds dosctring comments for new parameters
GabeVillalobos Apr 29, 2026
a2800de
Adds task stubs, new example tasks
GabeVillalobos Apr 30, 2026
24f66a6
Adds tests to cover different cases
GabeVillalobos Apr 30, 2026
4908092
Renames parameter to silenced_exceptions
GabeVillalobos May 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 37 additions & 0 deletions clients/python/src/examples/tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from examples.app import app
from taskbroker_client.retry import LastAction, NoRetriesRemainingError, Retry, RetryTaskError
from taskbroker_client.retry import retry_task as retry_task_helper
from taskbroker_client.worker.workerchild import ProcessingDeadlineExceeded

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -72,3 +73,39 @@ def will_retry(failure: str) -> None:
def timed_task(sleep_seconds: float | str, *args: list[Any], **kwargs: dict[str, Any]) -> None:
sleep(float(sleep_seconds))
logger.debug("timed_task complete")


@exampletasks.register(
name="examples.will_timeout_without_reporting",
processing_deadline_duration=1,
report_timeout_errors=False,
)
def will_timeout_without_reporting() -> None:
timed_task(sleep_seconds=2)


@exampletasks.register(
name="examples.will_fail_with_silenced_exception",
retry=Retry(times=2, times_exceeded=LastAction.Discard),
silenced_exceptions=(RuntimeError,),
)
def will_fail_with_silenced_exception() -> None:
raise RuntimeError("oh no")


@exampletasks.register(
name="examples.will_fail_with_silenced_ignored_exception",
retry=Retry(times=2, on=(RuntimeError,), times_exceeded=LastAction.Discard),
silenced_exceptions=(RuntimeError,),
)
def will_fail_with_silenced_ignored_exception() -> None:
raise RuntimeError("oh no")


@exampletasks.register(
name="examples.will_retry_on_deadline_exceeded",
processing_deadline_duration=1,
retry=Retry(times=2, on=(ProcessingDeadlineExceeded,), times_exceeded=LastAction.Discard),
)
def will_retry_on_deadline_exceeded() -> None:
timed_task(sleep_seconds=2)
16 changes: 16 additions & 0 deletions clients/python/src/taskbroker_client/registry.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,8 @@ def register(
at_most_once: bool = False,
wait_for_delivery: bool = False,
compression_type: CompressionType = CompressionType.PLAINTEXT,
report_timeout_errors: bool = True,
silenced_exceptions: tuple[type[BaseException], ...] | None = None,
) -> Callable[[Callable[P, R]], Task[P, R]]:
"""
Register a task.
Expand Down Expand Up @@ -115,6 +117,10 @@ def register(
before returning.
compression_type: CompressionType
The compression type to use to compress the task parameters.
report_timeout_errors: bool
Enable reporting of ProcessingDeadlineExceededError to Sentry.
silenced_exceptions: tuple[type[BaseException], ...] | None
A tuple of exception types that will not be reported by Sentry.
"""

def wrapped(func: Callable[P, R]) -> Task[P, R]:
Expand All @@ -133,6 +139,8 @@ def wrapped(func: Callable[P, R]) -> Task[P, R]:
at_most_once=at_most_once,
wait_for_delivery=wait_for_delivery,
compression_type=compression_type,
report_timeout_errors=report_timeout_errors,
silenced_exceptions=silenced_exceptions,
)
# TODO(taskworker) tasks should be registered into the registry
# so that we can ensure task names are globally unique
Expand Down Expand Up @@ -224,6 +232,8 @@ def register(
at_most_once: bool = False,
wait_for_delivery: bool = False,
compression_type: CompressionType = CompressionType.PLAINTEXT,
report_timeout_errors: bool = True,
silenced_exceptions: tuple[type[BaseException], ...] | None = None,
) -> Callable[[Callable[P, R]], ExternalTask[P, R]]:
"""
Register an external task stub.
Expand Down Expand Up @@ -251,6 +261,10 @@ def register(
before returning.
compression_type: CompressionType
The compression type to use to compress the task parameters.
report_timeout_errors: bool
Enable reporting of ProcessingDeadlineExceededError to Sentry.
silenced_exceptions: tuple[type[BaseException], ...] | None
A tuple of exception types that will not be reported by Sentry.
"""

def wrapped(func: Callable[P, R]) -> ExternalTask[P, R]:
Expand All @@ -269,6 +283,8 @@ def wrapped(func: Callable[P, R]) -> ExternalTask[P, R]:
at_most_once=at_most_once,
wait_for_delivery=wait_for_delivery,
compression_type=compression_type,
report_timeout_errors=report_timeout_errors,
silenced_exceptions=silenced_exceptions,
)
self._registered_tasks[name] = task
return task
Comment on lines 283 to 290
Copy link
Copy Markdown

@sentry sentry Bot Apr 29, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: The new expected_exceptions and report_timeout_errors parameters on ExternalNamespace.register() have no effect because ExternalTask instances are not executed locally.
Severity: MEDIUM

Suggested Fix

Raise a TypeError or ValueError if expected_exceptions or report_timeout_errors are passed to ExternalNamespace.register(), as these parameters are not supported for external tasks. Alternatively, document this limitation clearly in the method's docstring to prevent misuse.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: clients/python/src/taskbroker_client/registry.py#L283-L290

Potential issue: The `ExternalNamespace.register()` method now accepts
`expected_exceptions` and `report_timeout_errors` parameters. However, these are
silently ignored for external tasks. These parameters control task execution behavior,
but `ExternalTask` instances are only dispatched, not executed, by the local worker; the
remote application's task registration governs error handling. A developer setting
`expected_exceptions` on an external task stub will incorrectly believe they have
suppressed certain error reports, when in fact the setting has no effect at runtime.

Did we get this right? 👍 / 👎 to inform future reviews.

Expand Down
2 changes: 2 additions & 0 deletions clients/python/src/taskbroker_client/task.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@ def __init__(
wait_for_delivery: bool = False,
compression_type: CompressionType = CompressionType.PLAINTEXT,
report_timeout_errors: bool = True,
silenced_exceptions: tuple[type[BaseException], ...] | None = None,
):
self.name = name
self._func = func
Expand All @@ -86,6 +87,7 @@ def __init__(
self.wait_for_delivery = wait_for_delivery
self.compression_type = compression_type
self.report_timeout_errors = report_timeout_errors
self.silenced_exceptions = silenced_exceptions or ()
update_wrapper(self, func)

@property
Expand Down
32 changes: 20 additions & 12 deletions clients/python/src/taskbroker_client/worker/workerchild.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,6 +266,7 @@ def handle_alarm(signum: int, frame: FrameType | None) -> None:
except Exception as err:
retry = task_func.retry
captured_error = False
should_capture_error = not isinstance(err, task_func.silenced_exceptions)
if retry:
if retry.should_retry(inflight.activation.retry_state, err):
logger.info(
Comment on lines 266 to 272
Copy link
Copy Markdown

@sentry sentry Bot May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: The silenced_exceptions parameter does not work for ProcessingDeadlineExceeded because it is caught by an earlier, more specific except block that ignores this parameter.
Severity: MEDIUM

Suggested Fix

Modify the except ProcessingDeadlineExceeded block to also check if the exception type is present in the silenced_exceptions tuple. If it is, the exception should be suppressed according to the intended logic. This will align the behavior with the API's documented purpose.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: clients/python/src/taskbroker_client/worker/workerchild.py#L266-L272

Potential issue: The `silenced_exceptions` parameter has no effect when
`ProcessingDeadlineExceeded` is included in it. This is because
`ProcessingDeadlineExceeded` inherits from `BaseException` and is caught by a dedicated
`except ProcessingDeadlineExceeded` block. This earlier block does not check the
`silenced_exceptions` list. As a result, the general `except Exception` block, which
contains the logic for `silenced_exceptions`, is never reached for this specific
exception. This leads to a misleading API contract where the attempt to silence
`ProcessingDeadlineExceeded` fails without any indication.

Copy link
Copy Markdown
Member Author

@GabeVillalobos GabeVillalobos May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is intentional. report_timeout_errors controls whether or not we report ProcessingDeadlineExceeded errors to Sentry.

Expand All @@ -280,20 +281,27 @@ def handle_alarm(signum: int, frame: FrameType | None) -> None:
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)
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)
# In this branch, all exceptions should be either
# captured or silenced.
Comment thread
cursor[bot] marked this conversation as resolved.
captured_error = True

if not captured_error and next_state != TASK_ACTIVATION_STATUS_RETRY:
if (
should_capture_error
and not captured_error
and next_state != TASK_ACTIVATION_STATUS_RETRY
):
sentry_sdk.capture_exception(err)

clear_current_task()
Expand Down
179 changes: 179 additions & 0 deletions clients/python/tests/worker/test_worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,82 @@
),
)

# Task with Retry logic, expected exceptions to silence reporting
RETRY_TASK_WITH_SILENCED_TIMEOUT = InflightTaskActivation(
host="localhost:50051",
receive_timestamp=0,
activation=TaskActivation(
id="654",
taskname="examples.will_timeout_without_reporting",
namespace="examples",
parameters_bytes=msgpack.packb({"args": [], "kwargs": {}}, use_bin_type=True),
processing_deadline_duration=1,
retry_state=RetryState(
# no more attempts left
attempts=1,
max_attempts=2,
on_attempts_exceeded=ON_ATTEMPTS_EXCEEDED_DISCARD,
),
),
)

# Task with Retry logic, expected exceptions to silence reporting
RETRY_TASK_WITH_SILENCED_UNHANDLED_EXCEPTION = InflightTaskActivation(
host="localhost:50051",
receive_timestamp=0,
activation=TaskActivation(
id="654",
taskname="examples.will_fail_with_silenced_exception",
namespace="examples",
parameters_bytes=msgpack.packb({"args": [], "kwargs": {}}, use_bin_type=True),
processing_deadline_duration=2,
retry_state=RetryState(
# One retry left
attempts=0,
max_attempts=2,
on_attempts_exceeded=ON_ATTEMPTS_EXCEEDED_DISCARD,
),
),
)

# Task set to retry on deadline exceeded exceptions
RETRY_TASK_WITH_SILENCED_IGNORED_EXCEPTION = InflightTaskActivation(
host="localhost:50051",
receive_timestamp=0,
activation=TaskActivation(
id="654",
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(
# One retry left
attempts=0,
max_attempts=2,
on_attempts_exceeded=ON_ATTEMPTS_EXCEEDED_DISCARD,
),
),
)

# Task set to retry on deadline exceeded exceptions
RETRY_TASK_ON_DEADLINE_EXCEEDED = InflightTaskActivation(
host="localhost:50051",
receive_timestamp=0,
activation=TaskActivation(
id="654",
taskname="examples.will_retry_on_deadline_exceeded",
namespace="examples",
parameters_bytes=msgpack.packb({"args": [], "kwargs": {}}, use_bin_type=True),
processing_deadline_duration=1,
retry_state=RetryState(
# One retry left
attempts=0,
max_attempts=2,
on_attempts_exceeded=ON_ATTEMPTS_EXCEEDED_DISCARD,
),
),
)


class TestTaskWorker(TestCase):
def test_fetch_task(self) -> None:
Expand Down Expand Up @@ -901,3 +977,106 @@ def on_execute(self, headers: dict[str, str]) -> contextlib.AbstractContextManag
assert executed_headers[0]["x-viewer-user"] == "7"
finally:
app.context_hooks.remove(hook)


@mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception")
def test_child_process_silenced_timeout(mock_capture: mock.Mock) -> None:
todo: queue.Queue[InflightTaskActivation] = queue.Queue()
processed: queue.Queue[ProcessingResult] = queue.Queue()
shutdown = Event()

todo.put(RETRY_TASK_WITH_SILENCED_TIMEOUT)
child_process(
"examples.app:app",
todo,
processed,
shutdown,
max_task_count=1,
processing_pool_name="test",
process_type="fork",
)

assert todo.empty()
result = processed.get()
assert result.task_id == RETRY_TASK_WITH_SILENCED_TIMEOUT.activation.id
assert result.status == TASK_ACTIVATION_STATUS_FAILURE
assert mock_capture.call_count == 0


@mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception")
def test_child_process_silenced_exception_with_retries(mock_capture: mock.Mock) -> None:
todo: queue.Queue[InflightTaskActivation] = queue.Queue()
processed: queue.Queue[ProcessingResult] = queue.Queue()
shutdown = Event()

todo.put(RETRY_TASK_WITH_SILENCED_UNHANDLED_EXCEPTION)
child_process(
"examples.app:app",
todo,
processed,
shutdown,
max_task_count=1,
processing_pool_name="test",
process_type="fork",
)

assert todo.empty()
result = processed.get()
assert result.task_id == RETRY_TASK_WITH_SILENCED_UNHANDLED_EXCEPTION.activation.id

# No reporting, but the task still raised an unhandled exception
assert result.status == TASK_ACTIVATION_STATUS_FAILURE
assert mock_capture.call_count == 0


@mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception")
def test_child_process_expected_ignored_exception_max_attempts(mock_capture: mock.Mock) -> None:
todo: queue.Queue[InflightTaskActivation] = queue.Queue()
processed: queue.Queue[ProcessingResult] = queue.Queue()
shutdown = Event()

# Task has more retries left, but is set to ignore the raised error type
todo.put(RETRY_TASK_WITH_SILENCED_IGNORED_EXCEPTION)
child_process(
"examples.app:app",
todo,
processed,
shutdown,
max_task_count=1,
processing_pool_name="test",
process_type="fork",
)

# No reporting, but exception type is retriable
assert todo.empty()
result = processed.get()
assert result.task_id == RETRY_TASK_WITH_SILENCED_IGNORED_EXCEPTION.activation.id
assert result.status == TASK_ACTIVATION_STATUS_RETRY
assert mock_capture.call_count == 0


@mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception")
def test_child_process_retry_on_deadline_exceeded(mock_capture: mock.Mock) -> None:
todo: queue.Queue[InflightTaskActivation] = queue.Queue()
processed: queue.Queue[ProcessingResult] = queue.Queue()
shutdown = Event()

# Task will timeout, but should retry, because ProcessingDeadlineExceeded is
# in the Retry.on list
todo.put(RETRY_TASK_ON_DEADLINE_EXCEEDED)
child_process(
"examples.app:app",
todo,
processed,
shutdown,
max_task_count=1,
processing_pool_name="test",
process_type="fork",
)

assert todo.empty()
result = processed.get()
assert result.task_id == RETRY_TASK_ON_DEADLINE_EXCEEDED.activation.id
assert result.status == TASK_ACTIVATION_STATUS_RETRY
assert mock_capture.call_count == 1
assert type(mock_capture.call_args.args[0]) is ProcessingDeadlineExceeded
Loading