diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index da729f9..33e86f8 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -13,6 +13,13 @@ curl -sSL https://raw.githubusercontent.com/codingjoe/naming-things/refs/heads/m - Consistency – We never lose data, even if someone unplugs the power or network. - Utilization – We keep the CPU saturated with tasks, not with idle time or waiting for locks. +We require a persistent Redis without eviction. + +Redis connections use the redis-py default `decode_responses=False`, so all +values read from Redis are bytes. We do not guard against misconfiguration. +We fail loudly instead. The same goes for Redis data altered mid-flight. +These are deliberate design decisions. + ## Testing We have unit tests, integration tests, and benchmarks. Avoid mocking if possible. diff --git a/README.md b/README.md index 052023f..893ef6f 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,9 @@ uv run manage.py threadmill inspector ### Redis Backend Options +> [!IMPORTANT] +> Threadmill requires a persistent Redis without eviction. + The `RedisTaskBackend` accepts the following options under `OPTIONS` in your `TASKS` configuration: diff --git a/tests/backends/test_redis.py b/tests/backends/test_redis.py index 38582f4..cd9bf0c 100644 --- a/tests/backends/test_redis.py +++ b/tests/backends/test_redis.py @@ -128,7 +128,7 @@ def _claim_expired( f"{backend.key_prefix}:task:", ], ) - return [item.decode() if isinstance(item, bytes) else item for item in claimed] + return [item.decode() for item in claimed] class TestRedisBroker: @@ -265,43 +265,32 @@ def test_reap__removes_running_entry_without_task_data(self): finally: backend.close() - def test_reap_running_queue__logs_and_continues_after_task_error(self, caplog): - """A failing reap decision is logged per cause and does not stop the batch.""" + def test_reap_running_queue__logs_and_continues_when_retry_callback_is_gone( + self, caplog + ): + """A batch keeps reaping when a task's retry callback is gone from the code base.""" backend = _make_backend( - "reap_error_test", lease_ttl=datetime.timedelta(seconds=1) + "reap_gone_callback_test", lease_ttl=datetime.timedelta(seconds=1) ) try: - task_ids = [] - for _index in range(3): + for _index in range(2): task_result = backend.enqueue(boom_no_retry, args=[]) acquired = backend.acquire( timeout=datetime.timedelta(seconds=1), worker="worker-1" ) assert acquired is not None _expire_lease(backend, task_result.id) - task_ids.append(task_result.id) - - unreadable_id, gone_id, recovered_id = task_ids - backend.client.hset( - backend.TASK_KEY.format( - prefix=backend.key_prefix, task_id=unreadable_id - ), - "data", - "{not json", - ) - gone_key = backend.TASK_KEY.format( - prefix=backend.key_prefix, task_id=gone_id - ) - payload = json.loads(backend.client.hget(gone_key, "data")) - payload["task"]["retry"] = "tests.testapp.tasks.gone_from_the_code_base" - backend.client.hset(gone_key, "data", json.dumps(payload)) + task_key = backend.TASK_KEY.format( + prefix=backend.key_prefix, task_id=task_result.id + ) + payload = json.loads(backend.client.hget(task_key, "data")) + payload["task"]["retry"] = "tests.testapp.tasks.gone_from_the_code_base" + backend.client.hset(task_key, "data", json.dumps(payload)) with caplog.at_level(logging.ERROR, logger="threadmill.backends.redis"): RedisBroker(backend)._reap_running_queue("default") - assert "has an unreadable payload" in caplog.text - assert "gone from the code base" in caplog.text - assert backend.get_result(recovered_id).status == TaskResultStatus.FAILED + assert caplog.text.count("gone from the code base") == 2 finally: backend.close() diff --git a/threadmill/backends/redis.py b/threadmill/backends/redis.py index 96f3165..eed4936 100644 --- a/threadmill/backends/redis.py +++ b/threadmill/backends/redis.py @@ -86,7 +86,7 @@ def _reap_running_queue(self, queue_name: str) -> None: ], ) for member in claimed_ids: - task_id = member.decode() if isinstance(member, bytes) else member + task_id = member.decode() try: self._reap_task(task_id) except ImportError: @@ -95,10 +95,6 @@ def _reap_running_queue(self, queue_name: str) -> None: "skipping the reap", task_id, ) - except TypeError, ValueError: - logger.exception( - "Task %r has an unreadable payload; skipping the reap", task_id - ) def _reap_task(self, task_id: str) -> None: """Requeue or fail a claimed task.""" @@ -460,7 +456,7 @@ def _peek( ) -> Generator[TaskResult]: pipe = self.client.pipeline() for member in self.client.zrange(zset_key, 0, count - 1): - member_id = member.decode() if isinstance(member, bytes) else member + member_id = member.decode() data_key = data_key_template.format( prefix=self.key_prefix, task_id=member_id, @@ -472,9 +468,7 @@ def _peek( pipe.hget(data_key, field) for data in pipe.execute(): if data: - yield self.deserialize_task_result( - data.decode() if isinstance(data, bytes) else data - ) + yield self.deserialize_task_result(data.decode()) def get_result(self, result_id: str) -> TaskResult: if data := self.client.get( @@ -523,8 +517,7 @@ async def worker_telemetry( try: async for message in pubsub.listen(): if (data := message.get("data")) is not None: - payload = data.decode() if isinstance(data, bytes) else data - direction, _, queue_name = payload.partition(":") + direction, _, queue_name = data.decode().partition(":") try: event = TelemetryEvent( direction=TelemetryDirection(direction),