# Building the Execution Engine: Queue Manager, Workers and Process Orchestration > *Part 3 of the "Beyond FastAPI" series.* In the [previous article](https://hashnode.com/edit/cmrnf7vk900000ahtatwo5i7b), we built a thin orchestration layer using FastAPI. The API validates incoming requests, persists them as pending jobs, and immediately returns a Job ID to the client. At this point, our application is responsive, but it's also incomplete. Although jobs are now safely stored in MongoDB, nothing is actually executing them. Something still needs to answer a number of important questions. * Which job should run next? * How many jobs can execute concurrently? * How do we prevent two workers from processing the same job? * What happens when a worker crashes? * How do we detect jobs that become stuck? * How do we clean up finished processes? These responsibilities don't belong inside the API. They belong to a dedicated scheduling component. In our architecture, that component is the **Queue Manager**. * * * ### The Queue Manager Is Not the Queue One misconception developers often have when reading architectures like this is assuming that the Queue Manager *is* the queue. It isn't. The queue already exists. It's our MongoDB collection. Although it can be any other SQL based database as following. ```plaintext ┌──────────────────────────┐ │ MongoDB Jobs Collection │ ├──────────────────────────┤ │ pending │. │ │ pending │. │ │ running │. │ │ completed │. │ │ failed │. │ └──────────────────────────┘ ``` Every analysis request submitted through FastAPI is stored as a document (or row) inside this collection. Each document represents the current state of a job. The Queue Manager doesn't own this data. Instead, it continuously observes it and makes scheduling decisions based on the current state of the system. This distinction is subtle but extremely important. So MongoDB as the **source of truth**. The Queue Manager is simply the **scheduler** responsible for moving jobs through their lifecycle. * * * ![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/6459f262-2cd5-4580-8403-9e7bdd75caaa.jpg align="center") ### Separating Scheduling from Execution One of the design goals of this platform was ensuring that no single component performs too many responsibilities. Let's compare two possible implementations. ### Option 1 ```plaintext FastAPI ↓ Find Pending Job ↓ Execute Analysis ↓ Store Result ``` This is simple. Unfortunately, it's also tightly coupled. The API now owns: * HTTP communication * Scheduling * Concurrency * Worker lifecycle * Analysis execution Every new feature increases the complexity of the web application. Instead, our architecture separates these concerns. ```plaintext FastAPI │ ▼ MongoDB Job Queue │ ▼ Queue Manager ┌───────┴────────┐ ▼ ▼ Listener Loop Watchdog Loop │ ▼ Worker Process ``` Each component now has a single responsibility. FastAPI accepts work. MongoDB stores work. Queue Manager schedules work. Workers execute work. Because each component focuses on one job, the overall system becomes much easier to understand, test and evolve. * * * ## Anatomy of the Queue Manager Let's look at the class itself. ```python class QueueManager: def __init__( self, DBFactory ): self.db = DBFactory() self._db_impl_factory_for_worker = DBFactory self._listener_task = None self._watchdog_task = None self._running_processes = {} ``` Although the implementation is relatively small, every member variable exists for a specific reason. ### Database Interface ```python self.db ``` The scheduler never communicates directly with MongoDB. Instead, it depends on the same abstraction used throughout the application. This keeps scheduling logic independent of the underlying persistence implementation. * * * ### Worker Database Factory ```python self._db_impl_factory_for_worker ``` This is one of the more interesting design decisions in the project. Instead of sharing a single database instance with worker processes, we inject a **factory function** capable of creating fresh database implementations. Why? Because worker processes execute in separate operating system processes. Sharing an already connected database client across processes can lead to unpredictable behaviour depending on the driver and operating system. Instead, every worker creates its own independent database connection. ```plaintext Parent Process │ ▼ Queue Manager │ ▼ Factory Function │ ┌────┴────┐ ▼ ▼ Worker 1 Worker 2 │ │ ▼ ▼ MongoDB MongoDB Connection Connection ``` This keeps process boundaries clean while avoiding shared state between workers. > **A Note on Database Connections** At first glance, creating a separate database client for each worker process might seem expensive. However, workers are **not spawned without bounds**. The Queue Manager enforces a configurable concurrency limit, ensuring that only a fixed number of worker processes can run simultaneously. Each worker establishes its own process-local MongoDB client; following MongoDB's recommendation of one client per process and exits once its job completes. Because the number of concurrent workers is tightly controlled, the maximum number of database clients is also bounded and predictable. We'll explore how this concurrency limit is enforced in the next section when we dive into the Listener Loop and scheduling logic. * * * ### Background Tasks The Queue Manager owns two asynchronous tasks. ```python _listener_task _watchdog_task ``` These are long-running background coroutines that execute independently of incoming HTTP requests. Each solves a different problem: The **Listener Loop** continuously searches for new work. The **Watchdog Loop** monitors already running work. Keeping these responsibilities separate makes the scheduling logic significantly easier to reason about. We'll dive into both loops next. * * * ### Tracking Running Processes Finally, the Queue Manager maintains an in-memory registry. ```python _running_processes ``` Conceptually, it looks like this. ```plaintext { "job-123": Process(pid=4128), "job-456": Process(pid=4132), "job-789": Process(pid=4141) } ``` Notice what this dictionary is **not**. It isn't the source of truth for job state. *MongoDB already provides that*. Instead, this registry exists purely so the Queue Manager can interact with operating system processes. For example: * Determining whether a worker is still alive * Terminating workers during shutdown * Reclaiming finished processes * Detecting crashed workers Keeping process metadata separate from persisted job state keeps the architecture both simple and resilient. * * * ### Starting the Execution Engine Once FastAPI starts, it invokes the Queue Manager's startup routine. ```python async def start(self): self._listener_task = asyncio.create_task( self._listener_loop() ) self._watchdog_task = asyncio.create_task( self._watchdog_loop() ) ``` Notice something interesting. Neither loop blocks the application. Instead, both are started using: ```python asyncio.create_task(...) ``` This allows them to execute concurrently alongside the FastAPI event loop. ```plaintext Event Loop ┌─────────┬──────────┐ ▼ ▼ ▼ HTTP Server Listener Watchdog ``` Although all three components share the same event loop, they remain completely independent. * The HTTP server continues accepting requests. * The Listener discovers new jobs. * The Watchdog monitors existing workers. No component blocks another. This is precisely the type of workload `asyncio` was designed for—multiple long-running, I/O-bound tasks cooperating within a single thread. ![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/bd25a0f0-ca4a-454c-9618-8968d5711858.jpg align="center") * * * ### Why `asyncio` Instead of Threads? A common question is why the scheduler itself isn't implemented using threads. The answer lies in understanding what the Queue Manager actually does. It isn't performing CPU-intensive computation. Instead, it's mostly waiting. * Waiting for database operations. * Waiting for polling intervals. * Waiting for workers to finish. * Waiting for timeout checks. These are classic **I/O-bound** operations. Using `asyncio` allows a single thread to efficiently coordinate hundreds or even thousands of these waiting operations without the overhead of managing multiple operating system threads. The workers, on the other hand, are a completely different story. They execute expensive LLM pipelines, consume significant CPU and memory resources, and need strong isolation from one another. That's why the workers are implemented using **separate processes**, not asynchronous coroutines. This separation allows each technology to solve the problem it is best suited for: * **asyncio** coordinates work. * **Processes** execute work. As we'll see next, the Listener Loop brings these two worlds together by continuously transforming pending jobs into isolated worker processes.