# How Pixelle-Video Handles Task Cleanup: A Deep Dive into the Async TaskManager

> Discover how Pixelle-Video handles task cleanup. Learn about its automatic cleanup loop, retention windows, and resource leak prevention.

- Repository: [AIDC-AI/Pixelle-Video](https://github.com/AIDC-AI/Pixelle-Video)
- Tags: deep-dive
- Published: 2026-04-23

---

**Pixelle-Video automatically cleans up finished tasks through a periodic `_cleanup_loop` coroutine that removes completed, failed, or cancelled tasks after a configurable retention window, ensuring no resource leaks when the server shuts down.**

The open-source **Pixelle-Video** project (AIDC-AI/Pixelle-Video) provides a robust background task system for AI video generation. Understanding how it handles task cleanup is critical for production deployments where memory management and resource constraints matter. This article examines the complete cleanup architecture from configuration to shutdown.

## Core Cleanup Architecture

Pixelle-Video's cleanup system centers on three interlocking components defined in [`api/tasks/manager.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py):

| Component | Purpose | Key Method |
|-----------|---------|------------|
| **Task lifecycle management** | Create, track, and execute async tasks | `create_task()`, `execute_task()` |
| **Periodic cleanup loop** | Continuously purge stale tasks | `_cleanup_loop()` |
| **Old task removal** | Apply retention policy to finished tasks | `_cleanup_old_tasks()` |

The `TaskManager` class orchestrates these operations through a singleton pattern exposed as `task_manager` throughout the application.

## Starting and Stopping the Cleanup Loop

The cleanup lifecycle begins when the FastAPI server starts. In [`api/app.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/app.py), the lifespan context manager initializes the system:

```python

# api/app.py (lines 68-77)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize the task manager

    await task_manager.start()
    yield
    # Shutdown: cancel all pending tasks and cleanup

    await task_manager.stop()

```

The `start()` method in [`api/tasks/manager.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py) (lines 46-71) creates the persistent cleanup task:

```python

# api/tasks/manager.py – start() implementation

def start(self):
    """Start the background cleanup loop."""
    self._cleanup_task = asyncio.create_task(
        self._cleanup_loop(),
        name="task_cleanup_loop"
    )

```

When the server shuts down, `stop()` (lines 66-71) ensures graceful termination:

- Cancels the `_cleanup_loop` task
- Cancels all pending `Future` objects via `future.cancel()`
- Clears the internal `_tasks` and `_task_futures` dictionaries

This guarantees no dangling references or leaked asyncio tasks remain.

## The Cleanup Loop: `_cleanup_loop`

The heart of the system is `_cleanup_loop` (lines 232-242), an infinite coroutine that drives periodic cleanup:

```python

# api/tasks/manager.py – _cleanup_loop (lines 232-242)

async def _cleanup_loop(self):
    """Background task that periodically cleans up old tasks."""
    while True:
        try:
            await asyncio.sleep(self._cleanup_interval)
            await self._cleanup_old_tasks()
        except asyncio.CancelledError:
            # Expected on shutdown

            break
        except Exception as e:
            logger.error(f"Error in cleanup loop: {e}")

```

Key characteristics:

- **Configurable interval**: controlled by `task_cleanup_interval` (default 3600 seconds / 1 hour)
- **Exception resilience**: logs errors but continues running; only `CancelledError` breaks the loop
- **Non-blocking**: uses `asyncio.sleep` to yield control between runs

## Applying Retention Policy: `_cleanup_old_tasks`

The actual removal logic resides in `_cleanup_old_tasks` (lines 445-461):

```python

# api/tasks/manager.py – _cleanup_old_tasks (lines 445-461)

async def _cleanup_old_tasks(self):
    """Remove tasks that have exceeded the retention time."""
    cutoff = datetime.utcnow() - timedelta(seconds=self._retention_time)
    
    expired_tasks = [
        task_id for task_id, task in self._tasks.items()
        if task.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED)
        and task.completed_at and task.completed_at < cutoff
    ]
    
    for task_id in expired_tasks:
        del self._tasks[task_id]
        if task_id in self._task_futures:
            self._task_futures[task_id].cancel()
            del self._task_futures[task_id]
    
    if expired_tasks:
        logger.info(f"Cleaned up {len(expired_tasks)} expired tasks")

```

This method implements a **three-state cleanup eligibility**:

| Status | Eligible for Cleanup? |
|--------|----------------------|
| `COMPLETED` | ✅ Yes |
| `FAILED` | ✅ Yes |
| `CANCELLED` | ✅ Yes |
| `PENDING` or `RUNNING` | ❌ No |

The retention calculation uses `completed_at` timestamp compared against `task_retention_time` (default 86400 seconds / 24 hours).

## Configuration Settings

Cleanup behavior is controlled in [`api/config.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/config.py) (lines 33-37):

```python

# api/config.py – cleanup configuration (lines 33-37)

class APIConfig(BaseSettings):
    # How often to run the cleanup loop (seconds)

    task_cleanup_interval: int = 3600  # 1 hour

    
    # How long to keep finished task results (seconds)

    task_retention_time: int = 86400  # 24 hours

```

These environment-configurable values allow deployment-specific tuning:

- **High-volume services**: Reduce `task_cleanup_interval` to minutes
- **Compliance requirements**: Adjust `task_retention_time` for audit trails
- **Memory-constrained environments**: Lower retention to free RAM faster

## Complete Cleanup Flow Example

Here's how task cleanup operates from creation to removal:

```python
from api.tasks import task_manager, TaskType
import asyncio

async def demonstrate_cleanup():
    # 1. Create a video generation task

    task = task_manager.create_task(
        task_type=TaskType.VIDEO_GENERATION,
        request_params={"prompt": "A robot dancing in the rain"}
    )
    print(f"Created task: {task.task_id}")
    
    # 2. Execute the work (simulated here)

    await task_manager.execute_task(
        task_id=task.task_id,
        coro_func=asyncio.sleep,  # stand-in for actual video generation

        delay=0.1
    )
    
    # 3. Task is now COMPLETED with completed_at timestamp

    
    # 4. Cleanup loop (running every hour) will eventually call:

    #    _cleanup_old_tasks() → checks if completed_at < now - 24h

    #    → deletes task if expired

    
    # 5. On server shutdown, lifespan calls task_manager.stop():

    #    → cancels cleanup loop

    #    → cancels all futures

    #    → clears all dictionaries

# Run with: asyncio.run(demonstrate_cleanup())

```

## Summary

- **Start/stop lifecycle**: `task_manager.start()` launches `_cleanup_loop`; `task_manager.stop()` cancels it and clears all state
- **Periodic execution**: `_cleanup_loop` sleeps for `task_cleanup_interval` (default 1 hour) between cleanup runs
- **Retention policy**: `_cleanup_old_tasks` removes tasks with `COMPLETED`, `FAILED`, or `CANCELLED` status when `completed_at` exceeds `task_retention_time` (default 24 hours)
- **Graceful shutdown**: All pending futures are cancelled and dictionaries cleared to prevent memory leaks
- **Configuration**: Cleanup timing is controlled via `APIConfig` in [`api/config.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/config.py)

## Frequently Asked Questions

### How often does Pixelle-Video clean up finished tasks?

By default, the cleanup loop runs every **3600 seconds (1 hour)**. This is controlled by the `task_cleanup_interval` setting in [`api/config.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/config.py). You can reduce this to minutes for high-traffic deployments or increase it for lower resource usage.

### What determines when a task becomes eligible for cleanup?

A task must meet two conditions: its status must be `COMPLETED`, `FAILED`, or `CANCELLED` (not `PENDING` or `RUNNING`), and its `completed_at` timestamp must be older than the retention window. The default `task_retention_time` is **86400 seconds (24 hours)**.

### What happens to running tasks when the server shuts down?

When the FastAPI server shuts down, the `lifespan` context manager in [`api/app.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/app.py) calls `task_manager.stop()`. This cancels the cleanup loop, cancels all pending `Future` objects via `future.cancel()`, and clears the internal `_tasks` and `_task_futures` dictionaries to ensure no resource leaks.

### Can I customize the retention policy for different task types?

The current implementation in [`api/tasks/manager.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py) uses global `task_cleanup_interval` and `task_retention_time` values from `APIConfig`. There is no per-task-type retention in the base implementation. To implement custom retention, you would need to modify `_cleanup_old_tasks` to check `task.task_type` against a type-specific retention mapping.