From 6327b2edef1c3ce8ef0b5810805777f531564c25 Mon Sep 17 00:00:00 2001 From: Markus Unterwaditzer Date: Wed, 5 Aug 2026 14:20:22 +0200 Subject: [PATCH] feat(client): Add is_raw_mode flag to namespaces Raw-mode topics have no producer: taskbroker consumes the topic and builds the activation itself, from the topic's `raw:` block in its own config. Task options that a producer would normally embed in the activation therefore never reach the broker, and today they are dropped silently. `processing_deadline_duration` has already drifted because of this. `sentry.profiles.task.process_profile_from_kafka_raw` declares 80s, while every region deploys the topic with 60s, which is the value that actually applies. Nothing in the code says the decorator value is inert. Mark raw namespaces explicitly and reject the options taskbroker would ignore, so this fails at import instead of at runtime: - expires, processing_deadline_duration and compression_type, which come from the topic's `raw:` block - at_most_once and retry times_exceeded=Deadletter, which the broker hardcodes to false and Discard - a second task in the namespace, which would never run since the broker only spawns the single taskname pinned in its config ref STREAM-1044 Co-Authored-By: Claude Opus 5 (1M context) --- .../python/src/taskbroker_client/registry.py | 82 +++++++++++- clients/python/tests/test_registry.py | 117 ++++++++++++++++++ 2 files changed, 198 insertions(+), 1 deletion(-) diff --git a/clients/python/src/taskbroker_client/registry.py b/clients/python/src/taskbroker_client/registry.py index 48e81536..9e0c7699 100644 --- a/clients/python/src/taskbroker_client/registry.py +++ b/clients/python/src/taskbroker_client/registry.py @@ -17,7 +17,7 @@ CompressionType, ) from taskbroker_client.metrics import MetricsBackend -from taskbroker_client.retry import Retry +from taskbroker_client.retry import LastAction, Retry from taskbroker_client.router import TaskRouter from taskbroker_client.sdk import start_span from taskbroker_client.task import ExternalTask, P, R, Task @@ -47,6 +47,7 @@ def __init__( processing_deadline_duration: int = DEFAULT_PROCESSING_DEADLINE, app_feature: str | None = None, context_hooks: list[ContextHook] | None = None, + is_raw_mode: bool = False, ): self.name = name self.application = application @@ -54,6 +55,7 @@ def __init__( self.default_retry = retry self.default_expires = expires # seconds self.default_processing_deadline_duration = processing_deadline_duration # seconds + self.is_raw_mode = is_raw_mode self.app_feature = app_feature or name self.context_hooks: list[ContextHook] = context_hooks or [] self._registered_tasks: dict[str, Task[Any, Any]] = {} @@ -82,6 +84,66 @@ def topic(self) -> str: """The topic that a namespace is routed to.""" return self.router.route_namespace(self.name) + def _validate_raw_mode( + self, + *, + name: str, + retry: Retry | None, + expires: int | datetime.timedelta | None, + processing_deadline_duration: int | datetime.timedelta | None, + at_most_once: bool, + compression_type: CompressionType, + ) -> None: + """ + Reject task options that taskbroker ignores in raw mode. + + Nothing produces a raw-mode task: taskbroker consumes the topic and builds the + activation itself. Options that a producer would normally embed in the + activation therefore never reach the broker. Some of them it takes from the + topic's `raw:` config instead, and the rest it hardcodes. Either way the + decorator value is inert, so reject it here instead of dropping it silently at + runtime. + """ + if self._registered_tasks: + registered = ", ".join(sorted(self._registered_tasks)) + raise ValueError( + f"Raw-mode namespace {self.name!r} already has a registered task " + f"({registered}). taskbroker only spawns the single taskname configured " + f"for the raw topic, so {name!r} would never run. Give it its own namespace." + ) + + from_topic_config = [] + if expires is not None or self.default_expires is not None: + from_topic_config.append("expires") + if ( + processing_deadline_duration is not None + or self.default_processing_deadline_duration != DEFAULT_PROCESSING_DEADLINE + ): + from_topic_config.append("processing_deadline_duration") + if compression_type != CompressionType.PLAINTEXT: + from_topic_config.append("compression_type") + if from_topic_config: + raise ValueError( + f"Task {name!r} in raw-mode namespace {self.name!r} sets " + f"{', '.join(from_topic_config)} (on the task or as a namespace default). " + "In raw mode taskbroker takes these from the topic's `raw:` block in its " + "own config, not from the decorator, so the value here has no effect and " + "will silently drift from what is deployed." + ) + + if at_most_once: + raise ValueError( + f"Task {name!r} in raw-mode namespace {self.name!r} sets at_most_once. " + "taskbroker hardcodes at_most_once=false for raw activations, so the task " + "would still be retried past its processing deadline." + ) + if retry is not None and retry._times_exceeded == LastAction.Deadletter: + raise ValueError( + f"Task {name!r} in raw-mode namespace {self.name!r} sets " + "retry times_exceeded=LastAction.Deadletter. taskbroker hardcodes Discard " + "for raw activations, so exhausted tasks would be dropped, not deadlettered." + ) + def register( self, *, @@ -135,6 +197,15 @@ def wrapped(func: Callable[P, R]) -> Task[P, R]: task_retry = retry if not at_most_once: task_retry = retry or self.default_retry + if self.is_raw_mode: + self._validate_raw_mode( + name=name, + retry=task_retry, + expires=expires, + processing_deadline_duration=processing_deadline_duration, + at_most_once=at_most_once, + compression_type=compression_type, + ) task = Task( name=name, func=func, @@ -355,6 +426,7 @@ def create_namespace( processing_deadline_duration: int = DEFAULT_PROCESSING_DEADLINE, app_feature: str | None = None, internal: bool = False, + is_raw_mode: bool = False, ) -> TaskNamespace: """ Create a task namespace. @@ -363,6 +435,13 @@ def create_namespace( infrastructure to be scaled based on a region's requirements. Namespaces can define default behavior for tasks defined within a namespace. + + Set `is_raw_mode` when the namespace backs a topic that taskbroker consumes in + "raw mode", where taskbroker builds activations from raw Kafka messages itself + rather than a producer sending them. A raw topic maps 1:1 onto a namespace and + onto a single task, both pinned in taskbroker's own config. The flag makes that + contract explicit and rejects task options taskbroker would ignore; see + `TaskNamespace._validate_raw_mode`. """ if name == INTERNAL_NAMESPACE and not internal: raise ValueError(f"{INTERNAL_NAMESPACE!r} is reserved for internal taskbroker tasks.") @@ -379,6 +458,7 @@ def create_namespace( processing_deadline_duration=processing_deadline_duration, app_feature=app_feature, context_hooks=self._context_hooks, + is_raw_mode=is_raw_mode, ) self._namespaces[name] = namespace diff --git a/clients/python/tests/test_registry.py b/clients/python/tests/test_registry.py index e691e55a..34fab8ec 100644 --- a/clients/python/tests/test_registry.py +++ b/clients/python/tests/test_registry.py @@ -398,3 +398,120 @@ def test_registry_create_namespace_duplicate() -> None: registry.create_namespace(name="tests") with pytest.raises(ValueError, match="tests already exists"): registry.create_namespace(name="tests") + + +def raw_namespace(retry: Retry | None = None, **kwargs: Any) -> TaskNamespace: + return TaskNamespace( + name="tests.raw", + application="acme", + producer_factory=producer_factory, + router=DefaultRouter(), + metrics=NoOpMetricsBackend(), + retry=retry, + is_raw_mode=True, + **kwargs, + ) + + +def test_raw_mode_namespace_allows_plain_task() -> None: + namespace = raw_namespace() + + @namespace.register(name="tests.raw.consume", retry=Retry(times=3, delay=5)) + def consume(message_bytes: bytes) -> None: + raise NotImplementedError + + assert namespace.is_raw_mode + assert namespace.contains("tests.raw.consume") + + +def test_raw_mode_namespace_rejects_second_task() -> None: + namespace = raw_namespace() + + @namespace.register(name="tests.raw.consume") + def consume(message_bytes: bytes) -> None: + raise NotImplementedError + + with pytest.raises(ValueError, match="would never run"): + + @namespace.register(name="tests.raw.other") + def other(message_bytes: bytes) -> None: + raise NotImplementedError + + +@pytest.mark.parametrize( + "kwargs", + [ + pytest.param({"expires": 60}, id="expires"), + pytest.param({"processing_deadline_duration": 90}, id="processing_deadline_duration"), + pytest.param({"compression_type": CompressionType.ZSTD}, id="compression_type"), + ], +) +def test_raw_mode_namespace_rejects_topic_config_params(kwargs: dict[str, Any]) -> None: + namespace = raw_namespace() + + with pytest.raises(ValueError, match="takes these from the topic's `raw:` block"): + + @namespace.register(name="tests.raw.consume", **kwargs) + def consume(message_bytes: bytes) -> None: + raise NotImplementedError + + +@pytest.mark.parametrize( + "namespace_kwargs", + [ + pytest.param({"expires": 60}, id="expires"), + pytest.param({"processing_deadline_duration": 90}, id="processing_deadline_duration"), + ], +) +def test_raw_mode_namespace_rejects_topic_config_defaults(namespace_kwargs: dict[str, Any]) -> None: + namespace = raw_namespace(**namespace_kwargs) + + with pytest.raises(ValueError, match="takes these from the topic's `raw:` block"): + + @namespace.register(name="tests.raw.consume") + def consume(message_bytes: bytes) -> None: + raise NotImplementedError + + +def test_raw_mode_namespace_rejects_at_most_once() -> None: + namespace = raw_namespace() + + with pytest.raises(ValueError, match="hardcodes at_most_once=false"): + + @namespace.register(name="tests.raw.consume", at_most_once=True) + def consume(message_bytes: bytes) -> None: + raise NotImplementedError + + +def test_raw_mode_namespace_rejects_deadletter() -> None: + namespace = raw_namespace() + + with pytest.raises(ValueError, match="hardcodes Discard"): + + @namespace.register( + name="tests.raw.consume", + retry=Retry(times=3, times_exceeded=LastAction.Deadletter), + ) + def consume(message_bytes: bytes) -> None: + raise NotImplementedError + + +def test_raw_mode_namespace_rejects_deadletter_from_namespace_default() -> None: + namespace = raw_namespace(retry=Retry(times=3, times_exceeded=LastAction.Deadletter)) + + with pytest.raises(ValueError, match="hardcodes Discard"): + + @namespace.register(name="tests.raw.consume") + def consume(message_bytes: bytes) -> None: + raise NotImplementedError + + +def test_registry_create_namespace_is_raw_mode() -> None: + registry = TaskRegistry( + application="acme", + producer_factory=producer_factory, + router=DefaultRouter(), + metrics=NoOpMetricsBackend(), + ) + assert not registry.create_namespace("tests").is_raw_mode + assert registry.create_namespace("tests.raw", is_raw_mode=True).is_raw_mode