How to properly launch a background worker sharing state and event loop with Starlette?

[abdhx]

Hello team 👋

Problem

I want to launch a background worker in Starlette that shares the same state and event loop as the main application. My goal is to process messages both from the Starlette API and a worker queue using the same logic and shared state. Launching a separate process is not an option, as I need to share state and logic. (And also due to company microservices guidelines...)

I tried launching an async worker inside the application's lifespan event, but encountered many issues related to handling the worker lifecycle, especially stopping everything cleanly when the worker crashes.

Minimal Example

from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import JSONResponse
import asyncio

async def homepage(request):
    state = request.app.state.shared_state
    return JSONResponse({"value": state["value"]})

async def worker(app):
    try:
        while True:
            await asyncio.sleep(1)
            app.state.shared_state["value"] += 1
            print("Worker incremented value")
    except asyncio.CancelledError:
        print("Worker stopped cleanly")
        raise
    except Exception as exc:
        print(f"Worker crashed: {exc}")
        # I need to shutdown the app here
        raise

async def lifespan(app):
    app.state.shared_state = {"value": 0}
    app.state.worker_task = asyncio.create_task(worker(app))
    yield

app = Starlette(routes=[Route("/", homepage)], lifespan=lifespan)

Question

What is the recommended approach for running a background task like this in Starlette, ensuring that it shares the same event loop and state as the main application? How should the worker lifecycle (especially crash handling and stopping everything cleanly) be managed?

A different approach would be launching the worker first and letting it manages starlette and not the opposite?

Any advice or examples or suggestions of new approaches would be appreciated!

Thank you.

[nggit]

A passer-by, not a team.

First of all, the 'worker' I see in your code is not a real worker, just a background task. Which lives on the same single process as the server.

So if you want to gracefully shutdown the server just send a SIGINT/SIGTERM signal to the current pid.

import os
import signal


async def worker(app):
    try:
        while True:
            await asyncio.sleep(1)
            app.state.shared_state["value"] += 1
            print("Worker incremented value")
            # raise ValueError  # simulates a crash
    except asyncio.CancelledError:
        print("Worker stopped cleanly")
    except Exception as exc:
        print(f"Worker crashed: {exc}")
        # shutdown
        os.kill(os.getpid(), signal.SIGINT)


async def lifespan(app):
    app.state.shared_state = {"value": 0}
    app.state.worker_task = asyncio.create_task(worker(app))
    try:
        yield
    finally:
        app.state.worker_task.cancel()  # always stop 'worker'
        await app.state.worker_task

[abdhx]

I was thinking about changing the approach completely 🤔

Make the worker manage the server and not the opposite.. what do you think? I like gathering random people feedback

[nggit]

What is the reason? The Lifespan protocol exists to address needs like this.

[abdhx]

@nggit
I can accept that the server goes down but not the worker.

I was thinking about making the worker the "main" task. So there will be no need for hacks (to shutdown the server when the worker is down...)

[nggit]

One possible way is to run Uvicorn programatically.

But that's tricky if you want the worker to stay alive in main. uvicorn.run() is not a coroutine but blocking. You will still need to create a worker (as thread) first, then run uvicorn.run(). It will also be complicated in sharing the event loop.

[adriangb]

I recommend you: run unicorn programmatically from an asynchronous context. You have to create a Server and run that yourself. You can use asapi or check the source code for inspiration. Also use TaskGroups.