Implement FastStream task queue

[spwoodcock]

Is your feature request related to a problem? Please describe.

• No matter how much we optimise async processing, there is a limit to how many simultaneous requests can be handled by the FastAPI server + BackgroundTasks.
• Under high load we would need a task queue.
• We also need to scale under load, and a tried and tested metric to scale with is task queue length.

Describe the solution you'd like

• FastStream is a more modern alternative to Celery, managing event streams (in our case, a task queue).
• It's a merger of two message queue processing libraries (Propan & FastKafka) and is rapidly growing in support.
• There is easy integration into FastAPI.
• We should probably use a Redis broker and results backend, for the minimum resource requirements.
• Uses Pydantic for input and results serialisation, simple syntax, dependency injection (like FastAPI) and automatic documentation

Describe alternatives you've considered

• Celery (but it doesn't support asyncio).
• Simpliy optimising async processing as much as possible, plus BackgroundTasks (current implementation).
• arq which is a pretty neat upgrade over redis queue (rq). It uses a novel pessimistic execution model, where tasks are only removed from the queue once they have completed. As a consequence all tasks must be idempotent (can be run multiple times without issue).
• PostgresSQL LISTEN/NOTIFY event queue:
  • Postgres is typically overlooked as an event queue, but can function quite well for this purpose (as long as scale is not 1000's of simultaneous users).
  • In the spirit of keeping things simple, and reducing the knowledge overhead, resources required, and maintenance burden of adding another service (e.g. Redis) to the stack, this would be the best approach.
  • A great example of this: https://www.javaadvent.com/2022/12/using-postgres-as-a-message-queue.html
  • There is a Python implementation that could be used called Procrastinate, however I am uncertain about it's longevity, as the sole maintainer seems to be short of time.

Why FastStream was chosen over the above

• Easy integration into FastAPI.
• Junior developer friendly:
  • A Postgres based solution would be optimal, but adds extra complexity.
  • aqr would also be good, but requires task idempotecy, which is hard to enforce for devs.
  • FastStream hides a lot of complexity.
• Built in Pydantic validation is very nice.

[spwoodcock]

• I was starting to think we could get away without this by improving the task creation efficiency.
• However sometimes project areas can be large, with lots of tasks, so we need the task queue for two reasons:
  • To reduce the database concurrent connections during task data creation (odk token etc).
  • Allow for the automatic creation of basemaps for each task on project creation.
  • Prevent timeouts during project creation.
• Instead of using the project log file and polling, we can monitor the task status from the faststream redis queue, perhaps via websockets.

Note after this, we would probably want to decrease the connections param in docker-compose db setup, plus the SQLAlchemy create_engine, pool_size and max_overflow params.

[spwoodcock]

Ideally we need an endpoint to re-run the task creation if any tasks failed, probably by checking if there is an odk_token present for the task. If yes, attempt the generate_task_files function again.

[spwoodcock]

Further updates!

So the detail about requiring many connections during project creation is no longer true. Adding entities has greatly improved efficiency.

The only thing we really need this for is basemap generation. For that we have two alternatives:

• A separate microservice for basemap generation.
• Link into the use of electric-sql real time updates (from mapper frontend stack). The generation could run on a thread and update the frontend once done.

The second approach lends itself to adding a notifications table and perhaps having real time notifications when project creation is done & when basemaps are done.

I think this should be closed and replaced with either option above.