22868324715

Committed 09 Mar 2026 06:21PM UTC coverage: 97.577%. First build

Build # 22868324715

Build Type

Pull #2

github

Committed by

web-flow

Commit Message

Merge ad0ba7df0 into d04f85da2

Pull Request Pull Request #2: Improve task management with Redis Streams, leader election and management API

Coverage Stats

3182 of 3264 new or added lines in 32 files covered. (97.49%)

3342 of 3425 relevant lines covered (97.58%)

4.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.36

/src/fastapi_task_manager/reconciler.py

"""Reconciliation module - detects and recovers lost tasks (leader only).

This module provides the Reconciler class which periodically verifies the
consistency between what should be running and what is actually running.

It handles three recovery scenarios:
1. **Overdue tasks**: Tasks whose next_run is in the past but are not in the
   stream and not currently executing. This happens when the leader crashes
   before publishing to the stream, or during a leader failover gap.
2. **Stale tracking entries**: Entries in the scheduled SET that no longer have
   a corresponding message in the stream or a running indicator. These are
   cleaned up to prevent the SET from growing indefinitely.
3. **Stuck pending messages**: Messages in the stream's Pending Entries List (PEL)
   that have been idle too long (consumer crashed before ACK). These are
   reclaimed via XAUTOCLAIM and reassigned to active consumers.

The Reconciler only runs on the leader worker to avoid duplicate recovery work.
"""

import asyncio
import logging
import math
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING

from redis.asyncio import Redis
from redis.exceptions import ResponseError

from fastapi_task_manager.async_utils import interruptible_sleep
from fastapi_task_manager.coordinator import STREAM_MAX_LEN
from fastapi_task_manager.leader_election import LeaderElector
from fastapi_task_manager.redis_keys import RedisKeyBuilder
from fastapi_task_manager.schema.task import Task
from fastapi_task_manager.task_group import TaskGroup

if TYPE_CHECKING:
    from fastapi_task_manager import TaskManager

logger = logging.getLogger("fastapi.task-manager.reconciler")


class Reconciler:
    """Periodic reconciliation loop that detects and recovers lost tasks.

    Runs on the leader only, at a configurable interval (default: 30s).
    Each cycle performs three checks:
    1. Find overdue tasks that are not scheduled or running, and republish them.
    2. Clean up stale entries in the scheduled tracking SET.
    3. Reclaim stuck pending messages from crashed consumers.
    """

    def __init__(
        self,
        redis_client: Redis,
        key_builder: RedisKeyBuilder,
        leader_elector: LeaderElector,
        task_manager: "TaskManager",
    ) -> None:
        """Initialize the Reconciler.

        Args:
            redis_client: Async Redis client for all Redis operations.
            key_builder: RedisKeyBuilder instance for key construction.
            leader_elector: LeaderElector to verify leadership before acting.
            task_manager: TaskManager containing registered task groups and config.
        """
        self._redis = redis_client
        self._keys = key_builder
        self._leader = leader_elector
        self._task_manager = task_manager
        self._running = False

    async def start(self) -> asyncio.Task:
        """Start the reconciliation loop.

        Returns:
            The asyncio Task running the reconciliation loop.
        """
        self._running = True
        return asyncio.create_task(self._run(), name="Reconciler")

    async def stop(self) -> None:
        """Stop the reconciliation loop."""
        self._running = False

    async def _run(self) -> None:
        """Main reconciliation loop.

        Sleeps for the configured interval between checks using short sleep
        increments so that stop() can interrupt promptly. Only performs
        reconciliation when this worker is the leader.
        """
        config = self._task_manager.config

        while self._running:
            try:
                # Sleep in short increments to allow prompt shutdown
                await interruptible_sleep(config.reconciliation_interval, lambda: self._running)
                if not self._running:
                    break

                if not self._leader.is_leader:
                    continue

                logger.debug("Reconciliation cycle starting")
                await self._check_overdue_tasks()
                await self._cleanup_stale_tracking()
                await self._reclaim_stuck_pending()
                logger.debug("Reconciliation cycle completed")

            except asyncio.CancelledError:
                logger.info("Reconciler loop cancelled")
                break
            except Exception:
                logger.exception("Error in reconciliation cycle")
                await interruptible_sleep(1, lambda: self._running)

    async def _check_overdue_tasks(self) -> None:
        """Find tasks that are overdue and not scheduled or running, then republish.

        A task is considered lost if:
        - It is not disabled
        - It is not in the scheduled tracking SET (not in the stream)
        - It does not have an active running key (not being executed)
        - Its next_run timestamp is in the past by more than the safety margin

        The safety margin (reconciliation_interval * 5) avoids race conditions
        with the Coordinator: it ensures the Coordinator has had plenty of
        scheduling cycles to publish the task before the Reconciler steps in.
        """
        config = self._task_manager.config
        # Safety margin to avoid double-publishing with the Coordinator
        overdue_margin = timedelta(seconds=config.reconciliation_interval * 5)
        scheduled_set_key = self._keys.scheduled_set_key()

        for task_group in self._task_manager.task_groups:
            for task in task_group.tasks:
                try:
                    await self._check_single_task(
                        task_group,
                        task,
                        scheduled_set_key,
                        overdue_margin,
                    )
                except Exception:
                    logger.exception(
                        "Error checking task %s/%s during reconciliation",
                        task_group.name,
                        task.name,
                    )

    async def _check_single_task(
        self,
        task_group: TaskGroup,
        task: Task,
        scheduled_set_key: str,
        overdue_margin: timedelta,
    ) -> None:
        """Check a single task for overdue status and republish if needed.

        Args:
            task_group: The TaskGroup containing the task.
            task: The Task to check.
            scheduled_set_key: Redis key for the scheduled tracking SET.
            overdue_margin: Safety margin past next_run before considering overdue.
        """
        task_id = f"{task_group.name}:{task.name}"
        keys = self._keys.get_task_keys(task_group.name, task.name)
        running_key = self._keys.running_task_key(task_group.name, task.name)

        # Skip disabled tasks
        disabled = await self._redis.get(keys.disabled)
        if disabled is not None:
            return

        # Skip if already scheduled (in stream) or currently running
        is_scheduled = await self._redis.sismember(scheduled_set_key, task_id)  # ty: ignore[invalid-await]
        if is_scheduled:
            return

        is_running = await self._redis.exists(running_key)
        if is_running:
            return

        # Check if the task is overdue
        next_run_b = await self._redis.get(keys.next_run)
        if next_run_b is None:
            # Never scheduled and not in stream -> publish it
            logger.warning(
                "Reconciliation: task %s/%s has no next_run and is not scheduled, republishing",
                task_group.name,
                task.name,
            )
            await self._republish_task(task_group, task)
            return

        next_run = datetime.fromtimestamp(float(next_run_b), tz=timezone.utc)
        now = datetime.now(timezone.utc)

        if now > next_run + overdue_margin:
            logger.warning(
                "Reconciliation: task %s/%s is overdue by %s, republishing",
                task_group.name,
                task.name,
                now - next_run,
            )
            await self._republish_task(task_group, task)

    async def _republish_task(self, task_group: TaskGroup, task: Task) -> None:
        """Republish a lost task to the appropriate stream.

        Publishes the task to either the high or low priority stream based on
        the task's priority setting, and adds it to the scheduled tracking SET.

        Args:
            task_group: The TaskGroup containing the task.
            task: The Task to republish.
        """
        task_id = f"{task_group.name}:{task.name}"
        stream_key = self._keys.task_stream_high_key() if task.high_priority else self._keys.task_stream_low_key()

        message_id = await self._redis.xadd(
            stream_key,
            {
                "group": task_group.name,
                "task": task.name,
                "scheduled_at": str(datetime.now(timezone.utc).timestamp()),
                "reconciled": "1",
            },
            maxlen=STREAM_MAX_LEN,
            approximate=True,
        )

        # Track in the scheduled set
        await self._redis.sadd(self._keys.scheduled_set_key(), task_id)  # ty: ignore[invalid-await]

        priority = "high" if task.high_priority else "low"
        logger.info(
            "Reconciliation: republished task %s/%s to %s stream (msg_id: %s)",
            task_group.name,
            task.name,
            priority,
            message_id,
        )

    async def _cleanup_stale_tracking(self) -> None:
        """Remove stale entries from the scheduled tracking SET.

        An entry is stale if the task is neither running (no running key)
        nor pending in the stream. This can happen when a consumer finishes
        a task but fails to remove the tracking entry (e.g., partial failure
        during cleanup).
        """
        scheduled_set_key = self._keys.scheduled_set_key()
        members = await self._redis.smembers(scheduled_set_key)  # ty: ignore[invalid-await]

        for member in members:
            task_id = member

            # Parse "group:task" format
            if ":" not in task_id:
                # Malformed entry, remove it
                await self._redis.srem(scheduled_set_key, member)  # ty: ignore[invalid-await]
                continue

            group_name, task_name = task_id.split(":", 1)
            running_key = self._keys.running_task_key(group_name, task_name)

            # If the task is still running, the entry is valid
            is_running = await self._redis.exists(running_key)
            if is_running:
                continue

            # Check if the task still exists in registered task groups
            task = self._find_task(group_name, task_name)
            if task is None:
                # Task no longer registered, clean up the tracking entry
                logger.info(
                    "Reconciliation: removing stale tracking for unregistered task %s",
                    task_id,
                )
                await self._redis.srem(scheduled_set_key, member)  # ty: ignore[invalid-await]

    async def _reclaim_stuck_pending(self) -> None:
        """Reclaim messages stuck in the Pending Entries List (PEL).

        Finds messages that have been pending for longer than
        running_heartbeat_interval * 9 (in ms) (consumer crashed before ACK). For each
        stuck message, the Reconciler:
        1. ACKs the old pending message (removes it from the PEL)
        2. Re-publishes it as a new message via XADD

        This approach ensures the message goes back into the stream as a fresh
        entry, so the consumer group distributes it evenly across all active
        workers - rather than forcing it onto a single consumer via XCLAIM.
        """
        config = self._task_manager.config
        stream_keys = self._keys.get_stream_keys()
        # Derived from running_heartbeat_interval: TTL is interval * 3,
        # and we wait 3x the TTL to be sure the heartbeat has expired
        min_idle = math.ceil(config.running_heartbeat_interval * 9) * 1000

        for stream_key, label in [
            (stream_keys.task_stream_high, "high"),
            (stream_keys.task_stream_low, "low"),
        ]:
            try:
                # Find messages that have been pending too long
                pending_info = await self._redis.xpending_range(
                    stream_key,
                    stream_keys.consumer_group,
                    min="-",
                    max="+",
                    count=100,
                )

                if not pending_info:
                    continue

                for entry in pending_info:
                    idle_time = entry.get("time_since_delivered", 0)
                    if idle_time <= min_idle:
                        continue

                    message_id = entry.get("message_id")
                    if message_id is None:
                        continue

                    await self._requeue_pending_message(
                        stream_key,
                        stream_keys.consumer_group,
                        message_id,
                        label,
                    )

            except ResponseError as e:
                if "unknown command" in str(e).lower():
                    logger.debug("XPENDING not available, skipping pending check for %s stream", label)
                else:
                    logger.exception("Error checking pending messages in %s stream", label)
            except Exception:
                logger.exception("Error checking pending messages in %s stream", label)

    async def _requeue_pending_message(
        self,
        stream_key: str,
        group_name: str,
        message_id: str,
        priority_label: str,
    ) -> None:
        """ACK a stuck pending message and re-publish it as a new stream entry.

        This makes the message available to all consumers via normal consumer
        group distribution, instead of forcing it onto a specific consumer.

        Args:
            stream_key: The Redis stream key (high or low priority).
            group_name: The consumer group name.
            message_id: The stuck message ID to requeue.
            priority_label: "high" or "low" for logging.
        """
        # Read the original message data before ACK'ing
        messages = await self._redis.xrange(stream_key, min=message_id, max=message_id, count=1)

        if not messages:
            # Message was already trimmed from the stream, just ACK to clean PEL
            await self._redis.xack(stream_key, group_name, message_id)
            logger.debug(
                "Reconciliation: ACK'd orphan pending message %s (already trimmed from %s stream)",
                message_id,
                priority_label,
            )
            return

        # Extract the original message data
        _msg_id, data = messages[0]

        # ACK the old message to remove it from the PEL and delete it from the stream
        await self._redis.xack(stream_key, group_name, message_id)
        await self._redis.xdel(stream_key, message_id)

        # Re-publish as a new message so all consumers can compete for it
        new_message_id = await self._redis.xadd(
            stream_key,
            {
                **data,
                "requeued_from": message_id,
            },
            maxlen=STREAM_MAX_LEN,
            approximate=True,
        )

        logger.warning(
            "Reconciliation: requeued stuck %s priority message %s -> %s",
            priority_label,
            message_id,
            new_message_id,
        )

    def _find_task(self, group_name: str, task_name: str) -> Task | None:
        """Find a task by group and task name in registered task groups.

        Args:
            group_name: The task group name.
            task_name: The task name.

        Returns:
            The Task if found, None otherwise.
        """
        for task_group in self._task_manager.task_groups:
            if task_group.name == group_name:
                for task in task_group.tasks:
                    if task.name == task_name:
                        return task
        return None

1	"""Reconciliation module - detects and recovers lost tasks (leader only).
2
3	This module provides the Reconciler class which periodically verifies the
4	consistency between what should be running and what is actually running.
5
6	It handles three recovery scenarios:
7	1. Overdue tasks: Tasks whose next_run is in the past but are not in the
8	stream and not currently executing. This happens when the leader crashes
9	before publishing to the stream, or during a leader failover gap.
10	2. Stale tracking entries: Entries in the scheduled SET that no longer have
11	a corresponding message in the stream or a running indicator. These are
12	cleaned up to prevent the SET from growing indefinitely.
13	3. Stuck pending messages: Messages in the stream's Pending Entries List (PEL)
14	that have been idle too long (consumer crashed before ACK). These are
15	reclaimed via XAUTOCLAIM and reassigned to active consumers.
16
17	The Reconciler only runs on the leader worker to avoid duplicate recovery work.
18	"""
19
20	import asyncio	5✔
21	import logging	5✔
22	import math	5✔
23	from datetime import datetime, timedelta, timezone	5✔
24	from typing import TYPE_CHECKING	5✔
25
26	from redis.asyncio import Redis	5✔
27	from redis.exceptions import ResponseError	5✔
28
29	from fastapi_task_manager.async_utils import interruptible_sleep	5✔
30	from fastapi_task_manager.coordinator import STREAM_MAX_LEN	5✔
31	from fastapi_task_manager.leader_election import LeaderElector	5✔
32	from fastapi_task_manager.redis_keys import RedisKeyBuilder	5✔
33	from fastapi_task_manager.schema.task import Task	5✔
34	from fastapi_task_manager.task_group import TaskGroup	5✔
35
36	if TYPE_CHECKING:
37	from fastapi_task_manager import TaskManager
38
39	logger = logging.getLogger("fastapi.task-manager.reconciler")	5✔
40
41
42	class Reconciler:	5✔
43	"""Periodic reconciliation loop that detects and recovers lost tasks.
44
45	Runs on the leader only, at a configurable interval (default: 30s).
46	Each cycle performs three checks:
47	1. Find overdue tasks that are not scheduled or running, and republish them.
48	2. Clean up stale entries in the scheduled tracking SET.
49	3. Reclaim stuck pending messages from crashed consumers.
50	"""
51
52	def __init__(	5✔
53	self,
54	redis_client: Redis,
55	key_builder: RedisKeyBuilder,
56	leader_elector: LeaderElector,
57	task_manager: "TaskManager",
58	) -> None:
59	"""Initialize the Reconciler.
60
61	Args:
62	redis_client: Async Redis client for all Redis operations.
63	key_builder: RedisKeyBuilder instance for key construction.
64	leader_elector: LeaderElector to verify leadership before acting.
65	task_manager: TaskManager containing registered task groups and config.
66	"""
67	self._redis = redis_client	5✔
68	self._keys = key_builder	5✔
69	self._leader = leader_elector	5✔
70	self._task_manager = task_manager	5✔
71	self._running = False	5✔
72
73	async def start(self) -> asyncio.Task:	5✔
74	"""Start the reconciliation loop.
75
76	Returns:
77	The asyncio Task running the reconciliation loop.
78	"""
79	self._running = True	5✔
80	return asyncio.create_task(self._run(), name="Reconciler")	5✔
81
82	async def stop(self) -> None:	5✔
83	"""Stop the reconciliation loop."""
84	self._running = False	5✔
85
86	async def _run(self) -> None:	5✔
87	"""Main reconciliation loop.
88
89	Sleeps for the configured interval between checks using short sleep
90	increments so that stop() can interrupt promptly. Only performs
91	reconciliation when this worker is the leader.
92	"""
93	config = self._task_manager.config	5✔
94
95	while self._running:	5✔
96	try:	5✔
97	# Sleep in short increments to allow prompt shutdown
98	await interruptible_sleep(config.reconciliation_interval, lambda: self._running)	5✔
99	if not self._running:	5✔
100	break	5✔
101
102	if not self._leader.is_leader:	5✔
103	continue	5✔
104
105	logger.debug("Reconciliation cycle starting")	5✔
106	await self._check_overdue_tasks()	5✔
107	await self._cleanup_stale_tracking()	5✔
108	await self._reclaim_stuck_pending()	5✔
109	logger.debug("Reconciliation cycle completed")	5✔
110
NEW 111	except asyncio.CancelledError:	×
NEW 112	logger.info("Reconciler loop cancelled")	×
NEW 113	break	×
NEW 114	except Exception:	×
NEW 115	logger.exception("Error in reconciliation cycle")	×
NEW 116	await interruptible_sleep(1, lambda: self._running)	×
117
118	async def _check_overdue_tasks(self) -> None:	5✔
119	"""Find tasks that are overdue and not scheduled or running, then republish.
120
121	A task is considered lost if:
122	- It is not disabled
123	- It is not in the scheduled tracking SET (not in the stream)
124	- It does not have an active running key (not being executed)
125	- Its next_run timestamp is in the past by more than the safety margin
126
127	The safety margin (reconciliation_interval * 5) avoids race conditions
128	with the Coordinator: it ensures the Coordinator has had plenty of
129	scheduling cycles to publish the task before the Reconciler steps in.
130	"""
131	config = self._task_manager.config	5✔
132	# Safety margin to avoid double-publishing with the Coordinator
133	overdue_margin = timedelta(seconds=config.reconciliation_interval * 5)	5✔
134	scheduled_set_key = self._keys.scheduled_set_key()	5✔
135
136	for task_group in self._task_manager.task_groups:	5✔
137	for task in task_group.tasks:	5✔
138	try:	5✔
139	await self._check_single_task(	5✔
140	task_group,
141	task,
142	scheduled_set_key,
143	overdue_margin,
144	)
145	except Exception:	5✔
146	logger.exception(	5✔
147	"Error checking task %s/%s during reconciliation",
148	task_group.name,
149	task.name,
150	)
151
152	async def _check_single_task(	5✔
153	self,
154	task_group: TaskGroup,
155	task: Task,
156	scheduled_set_key: str,
157	overdue_margin: timedelta,
158	) -> None:
159	"""Check a single task for overdue status and republish if needed.
160
161	Args:
162	task_group: The TaskGroup containing the task.
163	task: The Task to check.
164	scheduled_set_key: Redis key for the scheduled tracking SET.
165	overdue_margin: Safety margin past next_run before considering overdue.
166	"""
167	task_id = f"{task_group.name}:{task.name}"	5✔
168	keys = self._keys.get_task_keys(task_group.name, task.name)	5✔
169	running_key = self._keys.running_task_key(task_group.name, task.name)	5✔
170
171	# Skip disabled tasks
172	disabled = await self._redis.get(keys.disabled)	5✔
173	if disabled is not None:	5✔
174	return	5✔
175
176	# Skip if already scheduled (in stream) or currently running
177	is_scheduled = await self._redis.sismember(scheduled_set_key, task_id) # ty: ignore[invalid-await]	5✔
178	if is_scheduled:	5✔
179	return	5✔
180
181	is_running = await self._redis.exists(running_key)	5✔
182	if is_running:	5✔
183	return	5✔
184
185	# Check if the task is overdue
186	next_run_b = await self._redis.get(keys.next_run)	5✔
187	if next_run_b is None:	5✔
188	# Never scheduled and not in stream -> publish it
189	logger.warning(	5✔
190	"Reconciliation: task %s/%s has no next_run and is not scheduled, republishing",
191	task_group.name,
192	task.name,
193	)
194	await self._republish_task(task_group, task)	5✔
195	return	5✔
196
NEW 197	next_run = datetime.fromtimestamp(float(next_run_b), tz=timezone.utc)	×
NEW 198	now = datetime.now(timezone.utc)	×
199
NEW 200	if now > next_run + overdue_margin:	×
NEW 201	logger.warning(	×
202	"Reconciliation: task %s/%s is overdue by %s, republishing",
203	task_group.name,
204	task.name,
205	now - next_run,
206	)
NEW 207	await self._republish_task(task_group, task)	×
208
209	async def _republish_task(self, task_group: TaskGroup, task: Task) -> None:	5✔
210	"""Republish a lost task to the appropriate stream.
211
212	Publishes the task to either the high or low priority stream based on
213	the task's priority setting, and adds it to the scheduled tracking SET.
214
215	Args:
216	task_group: The TaskGroup containing the task.
217	task: The Task to republish.
218	"""
219	task_id = f"{task_group.name}:{task.name}"	5✔
220	stream_key = self._keys.task_stream_high_key() if task.high_priority else self._keys.task_stream_low_key()	5✔
221
222	message_id = await self._redis.xadd(	5✔
223	stream_key,
224	{
225	"group": task_group.name,
226	"task": task.name,
227	"scheduled_at": str(datetime.now(timezone.utc).timestamp()),
228	"reconciled": "1",
229	},
230	maxlen=STREAM_MAX_LEN,
231	approximate=True,
232	)
233
234	# Track in the scheduled set
235	await self._redis.sadd(self._keys.scheduled_set_key(), task_id) # ty: ignore[invalid-await]	5✔
236
237	priority = "high" if task.high_priority else "low"	5✔
238	logger.info(	5✔
239	"Reconciliation: republished task %s/%s to %s stream (msg_id: %s)",
240	task_group.name,
241	task.name,
242	priority,
243	message_id,
244	)
245
246	async def _cleanup_stale_tracking(self) -> None:	5✔
247	"""Remove stale entries from the scheduled tracking SET.
248
249	An entry is stale if the task is neither running (no running key)
250	nor pending in the stream. This can happen when a consumer finishes
251	a task but fails to remove the tracking entry (e.g., partial failure
252	during cleanup).
253	"""
254	scheduled_set_key = self._keys.scheduled_set_key()	5✔
255	members = await self._redis.smembers(scheduled_set_key) # ty: ignore[invalid-await]	5✔
256
257	for member in members:	5✔
258	task_id = member	5✔
259
260	# Parse "group:task" format
261	if ":" not in task_id:	5✔
262	# Malformed entry, remove it
263	await self._redis.srem(scheduled_set_key, member) # ty: ignore[invalid-await]	5✔
264	continue	5✔
265
266	group_name, task_name = task_id.split(":", 1)	5✔
267	running_key = self._keys.running_task_key(group_name, task_name)	5✔
268
269	# If the task is still running, the entry is valid
270	is_running = await self._redis.exists(running_key)	5✔
271	if is_running:	5✔
272	continue	5✔
273
274	# Check if the task still exists in registered task groups
275	task = self._find_task(group_name, task_name)	5✔
276	if task is None:	5✔
277	# Task no longer registered, clean up the tracking entry
278	logger.info(	5✔
279	"Reconciliation: removing stale tracking for unregistered task %s",
280	task_id,
281	)
282	await self._redis.srem(scheduled_set_key, member) # ty: ignore[invalid-await]	5✔
283
284	async def _reclaim_stuck_pending(self) -> None:	5✔
285	"""Reclaim messages stuck in the Pending Entries List (PEL).
286
287	Finds messages that have been pending for longer than
288	running_heartbeat_interval * 9 (in ms) (consumer crashed before ACK). For each
289	stuck message, the Reconciler:
290	1. ACKs the old pending message (removes it from the PEL)
291	2. Re-publishes it as a new message via XADD
292
293	This approach ensures the message goes back into the stream as a fresh
294	entry, so the consumer group distributes it evenly across all active
295	workers - rather than forcing it onto a single consumer via XCLAIM.
296	"""
297	config = self._task_manager.config	5✔
298	stream_keys = self._keys.get_stream_keys()	5✔
299	# Derived from running_heartbeat_interval: TTL is interval * 3,
300	# and we wait 3x the TTL to be sure the heartbeat has expired
301	min_idle = math.ceil(config.running_heartbeat_interval * 9) * 1000	5✔
302
303	for stream_key, label in [	5✔
304	(stream_keys.task_stream_high, "high"),
305	(stream_keys.task_stream_low, "low"),
306	]:
307	try:	5✔
308	# Find messages that have been pending too long
309	pending_info = await self._redis.xpending_range(	5✔
310	stream_key,
311	stream_keys.consumer_group,
312	min="-",
313	max="+",
314	count=100,
315	)
316
317	if not pending_info:	5✔
318	continue	5✔
319
320	for entry in pending_info:	5✔
321	idle_time = entry.get("time_since_delivered", 0)	5✔
322	if idle_time <= min_idle:	5✔
323	continue	5✔
324
325	message_id = entry.get("message_id")	5✔
326	if message_id is None:	5✔
327	continue	5✔
328
329	await self._requeue_pending_message(	5✔
330	stream_key,
331	stream_keys.consumer_group,
332	message_id,
333	label,
334	)
335
336	except ResponseError as e:	5✔
337	if "unknown command" in str(e).lower():	5✔
338	logger.debug("XPENDING not available, skipping pending check for %s stream", label)	5✔
339	else:
340	logger.exception("Error checking pending messages in %s stream", label)	5✔
341	except Exception:	5✔
342	logger.exception("Error checking pending messages in %s stream", label)	5✔
343
344	async def _requeue_pending_message(	5✔
345	self,
346	stream_key: str,
347	group_name: str,
348	message_id: str,
349	priority_label: str,
350	) -> None:
351	"""ACK a stuck pending message and re-publish it as a new stream entry.
352
353	This makes the message available to all consumers via normal consumer
354	group distribution, instead of forcing it onto a specific consumer.
355
356	Args:
357	stream_key: The Redis stream key (high or low priority).
358	group_name: The consumer group name.
359	message_id: The stuck message ID to requeue.
360	priority_label: "high" or "low" for logging.
361	"""
362	# Read the original message data before ACK'ing
363	messages = await self._redis.xrange(stream_key, min=message_id, max=message_id, count=1)	5✔
364
365	if not messages:	5✔
366	# Message was already trimmed from the stream, just ACK to clean PEL
367	await self._redis.xack(stream_key, group_name, message_id)	5✔
368	logger.debug(	5✔
369	"Reconciliation: ACK'd orphan pending message %s (already trimmed from %s stream)",
370	message_id,
371	priority_label,
372	)
373	return	5✔
374
375	# Extract the original message data
376	_msg_id, data = messages[0]	5✔
377
378	# ACK the old message to remove it from the PEL and delete it from the stream
379	await self._redis.xack(stream_key, group_name, message_id)	5✔
380	await self._redis.xdel(stream_key, message_id)	5✔
381
382	# Re-publish as a new message so all consumers can compete for it
383	new_message_id = await self._redis.xadd(	5✔
384	stream_key,
385	{
386	**data,
387	"requeued_from": message_id,
388	},
389	maxlen=STREAM_MAX_LEN,
390	approximate=True,
391	)
392
393	logger.warning(	5✔
394	"Reconciliation: requeued stuck %s priority message %s -> %s",
395	priority_label,
396	message_id,
397	new_message_id,
398	)
399
400	def _find_task(self, group_name: str, task_name: str) -> Task \| None:	5✔
401	"""Find a task by group and task name in registered task groups.
402
403	Args:
404	group_name: The task group name.
405	task_name: The task name.
406
407	Returns:
408	The Task if found, None otherwise.
409	"""
410	for task_group in self._task_manager.task_groups:	5✔
411	if task_group.name == group_name:	5✔
412	for task in task_group.tasks:	5✔
413	if task.name == task_name:	5✔
414	return task	5✔
415	return None	5✔

Morry98 / fastapi-task-manager / 22868324715

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous