22850221594

Committed 09 Mar 2026 10:57AM UTC coverage: 97.615%. First build

Build # 22850221594

Build Type

Pull #2

github

Committed by

web-flow

Commit Message

Merge f2087c771 into d04f85da2

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

Coverage Stats

3196 of 3277 new or added lines in 32 files covered. (97.53%)

3356 of 3438 relevant lines covered (97.61%)

4.86 hits per line

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

91.45

/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

        # Decode the next_run value
        if isinstance(next_run_b, bytes):
            next_run_b = next_run_b.decode("utf-8")

        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]

        # Decode message_id for logging
        if isinstance(message_id, bytes):
            message_id = message_id.decode("utf-8")

        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:
            # Decode member if bytes
            task_id = member.decode("utf-8") if isinstance(member, bytes) else 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

                    # Decode message_id if bytes
                    if isinstance(message_id, bytes):
                        message_id = message_id.decode("utf-8")

                    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,
            {
                **{
                    (k.decode("utf-8") if isinstance(k, bytes) else k): (
                        v.decode("utf-8") if isinstance(v, bytes) else v
                    )
                    for k, v in data.items()
                },
                "requeued_from": message_id,
            },
            maxlen=STREAM_MAX_LEN,
            approximate=True,
        )

        # Decode new_message_id for logging
        if isinstance(new_message_id, bytes):
            new_message_id = new_message_id.decode("utf-8")

        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
197	# Decode the next_run value
NEW 198	if isinstance(next_run_b, bytes):	×
NEW 199	next_run_b = next_run_b.decode("utf-8")	×
200
NEW 201	next_run = datetime.fromtimestamp(float(next_run_b), tz=timezone.utc)	×
NEW 202	now = datetime.now(timezone.utc)	×
203
NEW 204	if now > next_run + overdue_margin:	×
NEW 205	logger.warning(	×
206	"Reconciliation: task %s/%s is overdue by %s, republishing",
207	task_group.name,
208	task.name,
209	now - next_run,
210	)
NEW 211	await self._republish_task(task_group, task)	×
212
213	async def _republish_task(self, task_group: TaskGroup, task: Task) -> None:	5✔
214	"""Republish a lost task to the appropriate stream.
215
216	Publishes the task to either the high or low priority stream based on
217	the task's priority setting, and adds it to the scheduled tracking SET.
218
219	Args:
220	task_group: The TaskGroup containing the task.
221	task: The Task to republish.
222	"""
223	task_id = f"{task_group.name}:{task.name}"	5✔
224	stream_key = self._keys.task_stream_high_key() if task.high_priority else self._keys.task_stream_low_key()	5✔
225
226	message_id = await self._redis.xadd(	5✔
227	stream_key,
228	{
229	"group": task_group.name,
230	"task": task.name,
231	"scheduled_at": str(datetime.now(timezone.utc).timestamp()),
232	"reconciled": "1",
233	},
234	maxlen=STREAM_MAX_LEN,
235	approximate=True,
236	)
237
238	# Track in the scheduled set
239	await self._redis.sadd(self._keys.scheduled_set_key(), task_id) # ty: ignore[invalid-await]	5✔
240
241	# Decode message_id for logging
242	if isinstance(message_id, bytes):	5✔
243	message_id = message_id.decode("utf-8")	5✔
244
245	priority = "high" if task.high_priority else "low"	5✔
246	logger.info(	5✔
247	"Reconciliation: republished task %s/%s to %s stream (msg_id: %s)",
248	task_group.name,
249	task.name,
250	priority,
251	message_id,
252	)
253
254	async def _cleanup_stale_tracking(self) -> None:	5✔
255	"""Remove stale entries from the scheduled tracking SET.
256
257	An entry is stale if the task is neither running (no running key)
258	nor pending in the stream. This can happen when a consumer finishes
259	a task but fails to remove the tracking entry (e.g., partial failure
260	during cleanup).
261	"""
262	scheduled_set_key = self._keys.scheduled_set_key()	5✔
263	members = await self._redis.smembers(scheduled_set_key) # ty: ignore[invalid-await]	5✔
264
265	for member in members:	5✔
266	# Decode member if bytes
267	task_id = member.decode("utf-8") if isinstance(member, bytes) else member	5✔
268
269	# Parse "group:task" format
270	if ":" not in task_id:	5✔
271	# Malformed entry, remove it
272	await self._redis.srem(scheduled_set_key, member) # ty: ignore[invalid-await]	5✔
273	continue	5✔
274
275	group_name, task_name = task_id.split(":", 1)	5✔
276	running_key = self._keys.running_task_key(group_name, task_name)	5✔
277
278	# If the task is still running, the entry is valid
279	is_running = await self._redis.exists(running_key)	5✔
280	if is_running:	5✔
281	continue	5✔
282
283	# Check if the task still exists in registered task groups
284	task = self._find_task(group_name, task_name)	5✔
285	if task is None:	5✔
286	# Task no longer registered, clean up the tracking entry
287	logger.info(	5✔
288	"Reconciliation: removing stale tracking for unregistered task %s",
289	task_id,
290	)
291	await self._redis.srem(scheduled_set_key, member) # ty: ignore[invalid-await]	5✔
292
293	async def _reclaim_stuck_pending(self) -> None:	5✔
294	"""Reclaim messages stuck in the Pending Entries List (PEL).
295
296	Finds messages that have been pending for longer than
297	running_heartbeat_interval * 9 (in ms) (consumer crashed before ACK). For each
298	stuck message, the Reconciler:
299	1. ACKs the old pending message (removes it from the PEL)
300	2. Re-publishes it as a new message via XADD
301
302	This approach ensures the message goes back into the stream as a fresh
303	entry, so the consumer group distributes it evenly across all active
304	workers - rather than forcing it onto a single consumer via XCLAIM.
305	"""
306	config = self._task_manager.config	5✔
307	stream_keys = self._keys.get_stream_keys()	5✔
308	# Derived from running_heartbeat_interval: TTL is interval * 3,
309	# and we wait 3x the TTL to be sure the heartbeat has expired
310	min_idle = math.ceil(config.running_heartbeat_interval * 9) * 1000	5✔
311
312	for stream_key, label in [	5✔
313	(stream_keys.task_stream_high, "high"),
314	(stream_keys.task_stream_low, "low"),
315	]:
316	try:	5✔
317	# Find messages that have been pending too long
318	pending_info = await self._redis.xpending_range(	5✔
319	stream_key,
320	stream_keys.consumer_group,
321	min="-",
322	max="+",
323	count=100,
324	)
325
326	if not pending_info:	5✔
327	continue	5✔
328
329	for entry in pending_info:	5✔
330	idle_time = entry.get("time_since_delivered", 0)	5✔
331	if idle_time <= min_idle:	5✔
332	continue	5✔
333
334	message_id = entry.get("message_id")	5✔
335	if message_id is None:	5✔
336	continue	5✔
337
338	# Decode message_id if bytes
339	if isinstance(message_id, bytes):	5✔
340	message_id = message_id.decode("utf-8")	5✔
341
342	await self._requeue_pending_message(	5✔
343	stream_key,
344	stream_keys.consumer_group,
345	message_id,
346	label,
347	)
348
349	except ResponseError as e:	5✔
350	if "unknown command" in str(e).lower():	5✔
351	logger.debug("XPENDING not available, skipping pending check for %s stream", label)	5✔
352	else:
353	logger.exception("Error checking pending messages in %s stream", label)	5✔
354	except Exception:	5✔
355	logger.exception("Error checking pending messages in %s stream", label)	5✔
356
357	async def _requeue_pending_message(	5✔
358	self,
359	stream_key: str,
360	group_name: str,
361	message_id: str,
362	priority_label: str,
363	) -> None:
364	"""ACK a stuck pending message and re-publish it as a new stream entry.
365
366	This makes the message available to all consumers via normal consumer
367	group distribution, instead of forcing it onto a specific consumer.
368
369	Args:
370	stream_key: The Redis stream key (high or low priority).
371	group_name: The consumer group name.
372	message_id: The stuck message ID to requeue.
373	priority_label: "high" or "low" for logging.
374	"""
375	# Read the original message data before ACK'ing
376	messages = await self._redis.xrange(stream_key, min=message_id, max=message_id, count=1)	5✔
377
378	if not messages:	5✔
379	# Message was already trimmed from the stream, just ACK to clean PEL
380	await self._redis.xack(stream_key, group_name, message_id)	5✔
381	logger.debug(	5✔
382	"Reconciliation: ACK'd orphan pending message %s (already trimmed from %s stream)",
383	message_id,
384	priority_label,
385	)
386	return	5✔
387
388	# Extract the original message data
389	_msg_id, data = messages[0]	5✔
390
391	# ACK the old message to remove it from the PEL and delete it from the stream
392	await self._redis.xack(stream_key, group_name, message_id)	5✔
393	await self._redis.xdel(stream_key, message_id)	5✔
394
395	# Re-publish as a new message so all consumers can compete for it
396	new_message_id = await self._redis.xadd(	5✔
397	stream_key,
398	{
399	**{
400	(k.decode("utf-8") if isinstance(k, bytes) else k): (
401	v.decode("utf-8") if isinstance(v, bytes) else v
402	)
403	for k, v in data.items()
404	},
405	"requeued_from": message_id,
406	},
407	maxlen=STREAM_MAX_LEN,
408	approximate=True,
409	)
410
411	# Decode new_message_id for logging
412	if isinstance(new_message_id, bytes):	5✔
413	new_message_id = new_message_id.decode("utf-8")	5✔
414
415	logger.warning(	5✔
416	"Reconciliation: requeued stuck %s priority message %s -> %s",
417	priority_label,
418	message_id,
419	new_message_id,
420	)
421
422	def _find_task(self, group_name: str, task_name: str) -> Task \| None:	5✔
423	"""Find a task by group and task name in registered task groups.
424
425	Args:
426	group_name: The task group name.
427	task_name: The task name.
428
429	Returns:
430	The Task if found, None otherwise.
431	"""
432	for task_group in self._task_manager.task_groups:	5✔
433	if task_group.name == group_name:	5✔
434	for task in task_group.tasks:	5✔
435	if task.name == task_name:	5✔
436	return task	5✔
437	return None	5✔

Morry98 / fastapi-task-manager / 22850221594

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