# Making Workers Production Ready ## The Worker Has Been Loaded… Now What? In the [previous](https://blog.lucrolearning.com/inside-the-worker-building-a-resilient-ai-execution-platform) article we followed a job from the Queue Manager into a newly created worker process. By this point our worker has: * Started a new Python process * Created its own asyncio event loop * Connected to MongoDB * Dynamically loaded the configured analysis pipeline At this stage the worker is finally ready to execute the actual business logic. It might be tempting to simply instantiate the pipeline and call: ```python await pipeline.run() ``` That would certainly work. But an execution platform needs much more than simply running code. The worker needs a way to: * Communicate progress * Manage external resources * Terminate gracefully * Avoid resource leaks * Remain independent from global application state These concerns are just as important as the analysis itself. In this article we'll explore how those responsibilities are handled. ![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/8af4b4cd-684a-4d48-932d-5e8839c4ef13.webp align="center") ## Dependency Injection Instead of Global State One design decision that often goes unnoticed is how the analysis pipeline receives its dependencies. Rather than importing global objects, everything required by the pipeline is injected during construction. ```python constructor_args = { "survey_id": payload["survey_id"], "question_id": payload["questionId"], "logger": logger, "register_thread": register_thread, "register_process": register_process, } ``` The worker knows how to construct the pipeline, but it doesn't dictate how the pipeline should behave. This approach has several advantages. ### Loose Coupling The pipeline doesn't need to know where the logger came from. It simply receives one. The same applies to thread registration, subprocess registration, configuration, or database services. ### Easier Testing Because dependencies are injected rather than imported globally, unit tests can substitute lightweight mock implementations. Instead of starting real worker processes, tests can inject fake loggers and in-memory services. ### Reusable Pipelines Nothing inside the pipeline depends on FastAPI, Queue Manager or multiprocessing. The pipeline could be executed from: * A worker process * CLI/Terminal based execution * Another orchestration system * Notebook without modification. This separation keeps the orchestration layer independent from the business logic. ## Reporting Progress AI workloads can easily take several minutes. Some may process thousands of comments. Others may call hundreds of external APIs. Without progress reporting the client has no visibility into what is happening. Instead of waiting silently, the worker continuously publishes milestones back to MongoDB. ```python async def progress_cb(message, percentage): await db_client.post_milestone(message, percentage) ``` The pipeline receives this callback during execution. Whenever meaningful progress occurs, it simply calls ```python await progress_cb("Generating embeddings", 35) ``` or ```python await progress_cb("Clustering comments", 70) ``` The pipeline never needs to know where those updates are stored. It simply reports progress. The worker takes responsibility for persistence. This is another example of separating execution from business logic. ## Why Callbacks Instead of Database Calls? The answer is abstraction. Imagine replacing MongoDB tomorrow with Redis, PostgreSQL or an event bus. If every pipeline contained database code, every implementation would need to change. Instead, the pipeline exposes intent: > "Progress has reached 60%." The worker decides how that information should be persisted. Keeping responsibilities separated makes the pipeline significantly easier to evolve. ## Graceful Shutdown Long-running AI jobs must also be interruptible: A deployment may require workers to stop , a watchdog may detect a stalled process or an administrator may terminate a runaway job. If we simply kill the process immediately, that can leave: * Partially written data * Orphaned subprocesses * Unfinished threads * Inconsistent state Instead the worker listens for operating system signals. ```python signal.signal(SIGTERM, handle_sigterm) ``` When the operating system sends `SIGTERM`, the handler doesn't terminate immediately. Instead it requests a graceful shutdown. ```python stop_event.set() ``` The pipeline periodically checks this event. If shutdown has been requested it can complete its current operation, release resources and exit safely. This gives the worker an opportunity to leave the system in a consistent state before terminating. ![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/917636ec-d4b2-4982-b4dc-bec2aeeb8135.webp align="center") ## Cleanup Matters One of the easiest ways to destabilise production systems is forgetting about resources created during execution. Analysis pipelines often create: * Background threads * Subprocesses * External tools Those resources don't automatically disappear when your analysis completes. To avoid leaks the worker keeps track of everything it creates. Whenever the pipeline starts a thread it registers it. Whenever it launches a subprocess it registers that as well. When execution finishes the worker performs one final cleanup pass. ```python cleanup_subtasks() ``` Internally this function: * Terminates child processes * Waits for graceful shutdown * Joins background threads * Clears internal registries Only after cleanup completes does the worker exit. This guarantees that every job leaves the system exactly as it found it. ## Putting it all together By now our worker has become much more than a background process. It is a complete execution environment. ```text Worker Starts │ ▼ Load Pipeline │ ▼ Inject Dependencies │ ▼ Execute Analysis │ ▼ Report Progress │ ▼ Handle Shutdown Requests │ ▼ Cleanup Resources │ ▼ Exit ``` Every worker is isolated. Every worker owns its lifecycle. Every worker can be terminated independently. Most importantly, the orchestration platform remains completely unaware of what the pipeline actually does. It simply provides a safe environment in which arbitrary analysis workloads can execute reliably. ## Conclusion At this point we've completed the execution platform. A client submits work. FastAPI persists the request. The Queue Manager schedules execution. An isolated worker loads the configured pipeline, executes it safely, reports progress, and cleans up every resource before exiting. What we haven't explored yet is **the pipeline itself**. How do we process tens of thousands of comments without exceeding LLM token limits? How do we execute embedding generation concurrently while respecting API rate limits? How do we recover from failures without repeating expensive computation? In the next—and final—article we'll step inside the AI pipeline itself and explore the engineering patterns behind scalable LLM-powered thematic analysis.