So far, we've built an execution engine that starts alongside our FastAPI application and runs independently of incoming HTTP requests. We have also seen the role of a queue manager that schedules and dispatches worker processes
The next question is straightforward:
How does a pending job actually become a running worker?
That responsibility belongs to the Listener Loop.
Conceptually, the Listener Loop acts as a scheduler. It continuously monitors the job queue, checks whether the system has available capacity, and dispatches new work when resources become available.
Unlike message brokers such as RabbitMQ or Kafka, MongoDB doesn't push notifications when new documents are inserted. Instead, our scheduler periodically polls the queue for pending work.
The implementation is surprisingly compact.
async def _listener_loop(self):
while True:
await self._reap_finished()
running = await self.db.count_running_jobs()
capacity = max(
0,
MAX_CONCURRENT_JOBS - running
)
if capacity > 0:
for _ in range(capacity):
doc = await self.db.claim_next_pending_job(...)
if not doc:
break
await self._spawn_worker(...)
await asyncio.sleep(POLL_INTERVAL_SECONDS)
Despite its size, this loop performs nearly all of the scheduling decisions in the platform.
Let's break it down.
A Scheduler, Not a Consumer
It's tempting to think of the Listener Loop as a "consumer" that simply pulls jobs from a queue.
In reality, it's doing considerably more than that.
Every iteration of the loop answers three questions:
How many jobs are already running?
How many more jobs can this instance execute?
Which pending jobs should be promoted to running?
Only after answering all three does it create new worker processes.
This is why I prefer describing it as a scheduler rather than a queue consumer.
It isn't just reading jobs.
It's making execution decisions.
Listener Loop Scheduling Cycle
Step 1: Reclaim Finished Workers
Every scheduling cycle begins by checking whether any worker processes have completed.
await self._reap_finished()
Why first?
Imagine three workers finish while the Listener Loop is sleeping.
From MongoDB's perspective, those jobs may already be marked as completed.
However, from the operating system's perspective, the associated processes still exist until the parent process acknowledges their termination. Reaping completed workers keeps the Queue Manager's in-memory process registry accurate and frees capacity for new work.
We'll explore process reaping in much greater detail later in this article, but for now it's enough to think of it as housekeeping before new scheduling decisions are made.
Step 2: Measure Current Capacity
The next step is determining how much work the system can safely accept.
running = await self.db.count_running_jobs()
capacity = max(
0,
MAX_CONCURRENT_JOBS - running
)
Rather than blindly launching workers whenever pending jobs exist, the scheduler first checks how many analyses are already executing.
Suppose our configuration allows a maximum of four concurrent workers.
MAX_CONCURRENT_JOBS = 4
If three jobs are currently running, the scheduler computes:
capacity = 4 - 3 = 1
Only one additional worker may be started. If four jobs are already running:
capacity = 0
No new workers are launched until capacity becomes available.
This simple calculation provides a hard upper bound on system resource usage.
Why Limit Concurrency?
LLM-powered analysis is both CPU and network intensive. Every worker consumes memory, maintains its own database client, performs external API calls, and may generate embeddings or process large datasets. Allowing unlimited workers would quickly exhaust available resources and degrade overall system performance. By enforcing a configurable concurrency limit, the scheduler ensures that throughput increases in a controlled and predictable manner rather than overwhelming the host machine.
Step 3: Claiming Work Atomically
Finding a pending job sounds trivial.
At first, you might imagine something like this:
job = db.find_one({
"status": "pending"
})
Unfortunately, this introduces a subtle race condition.
Imagine two Queue Managers running simultaneously.
Queue Manager A
↓
Find Pending Job #123
↓
Worker Spawn
At exactly the same moment:
Queue Manager B
↓
Find Pending Job #123
↓
Worker Spawn
Both schedulers have now selected the same document.
The result?
Two workers execute the same analysis.
Duplicate processing.
Conflicting updates.
Undefined behaviour.
This is one of the most common pitfalls in distributed scheduling systems. Instead of simply reading a pending document, our implementation performs an atomic claim.
Conceptually, it behaves like this:
Find first pending job
AND
Immediately update status
↓
running
Because both operations occur as a single database transaction (or atomic update operation), only one scheduler can successfully claim a given job. Every other scheduler simply moves on to the next available document. This small implementation detail is what makes the scheduler safe to scale horizontally across multiple application instances.
Key Insight
A queue isn't just about storing work. It's about ensuring that every piece of work is executed exactly once.
Filling Available Capacity
Once a job has been successfully claimed, the scheduler repeats the process until either:
for _ in range(capacity):
doc = await self.db.claim_next_pending_job(...)
if not doc:
break
await self._spawn_worker(...)
Notice that the scheduler doesn't claim every pending job in one pass.
Instead, it only claims enough work to fill the currently available capacity.
Suppose we have:
Pending Jobs: 15
Running Jobs: 2
Maximum Workers: 6
The scheduler calculates:
capacity = 4
Only four jobs are claimed.
The remaining eleven stay safely in the queue until the next scheduling cycle.
This approach naturally applies back-pressure to the system. The queue can grow as demand increases, while worker creation remains bounded by the configured concurrency limit. Rather than overwhelming the machine during traffic spikes, excess work simply waits its turn.
This is one of the defining characteristics of resilient asynchronous systems.
Sleeping Without Blocking
After scheduling any available work, the Listener Loop pauses briefly before repeating.
await asyncio.sleep(
POLL_INTERVAL_SECONDS
)
Because the scheduler is implemented using asyncio, this sleep doesn't block the event loop.
While the Listener is waiting, FastAPI continues serving HTTP requests, the Watchdog continues monitoring running jobs, and worker processes continue executing analyses.
The scheduler simply yields control until it's time for the next polling cycle.
Choosing the polling interval is a trade-off.
A shorter interval reduces the delay before new jobs begin executing but increases database activity.
A longer interval reduces database load at the cost of slightly higher scheduling latency.
In practice, a polling interval of a few seconds is more than sufficient for long-running AI workloads measured in minutes rather than milliseconds.
The Result
By the end of each iteration, the Listener Loop has transformed persisted job records into actively executing worker processes while respecting the system's configured capacity.
It doesn't perform any analysis itself.
It doesn't understand embeddings, prompts, or LLMs.
Its sole responsibility is deciding when work should begin.
That separation keeps the scheduler simple, predictable, and easy to reason about—qualities that become increasingly valuable as systems grow in complexity.
