How to do work queuing?

Introduction

One of the important things to consider for this CI/CD tool is how to execute the different jobs.
CI/CD jobs are long running and can take up to many hours depending on the complexity of the code
base, language, tooling, etc. This means that we cannot execute such tasks during the lifetime of a
request.

This would require having a job queue that multiple background workers read from to execute the
different jobs. After we are done with a job we can save its status to a database for later
querying. The user can poll the server to see the status of the CI/CD job.

For the actual implementation of such a job queue we have a few options. But we first need to know
what requirements we need the work queue to fulfill.

What do we need in a work queue?

There are a few things we need in a work queue:

1. At its most basic form a work queue needs to be able to store messages, and for those messages to be able to be queried by multiple workers.
2. Suppose a worker claims a job from the queue, then it crashes, or the message never reaches it because of a network partition. If we removed the message after the worker request, then that job is gone.

This means we cannot do anything to the job until we are sure of the delivery of the message. In
other words, we need an at-least-once delivery guarantee.

1. Suppose now that a worker receives the message and confirms the receipt, then crashes while executing the job. There are two potential issues in such a scenario:
   • If we just deliver the job without persisting it, that job is lost when the worker crashes, we have no way of handling the failure.
   • Even if we persisted the job, we need a specific retry mechanism.
2. We cannot keep retrying forever, we need a way to know when we retried a job too much. This is done by tracking the number of retries for a given job. In case we want to debug a potential issue in our system that might be causing job failure, we can save the job to a Dead Letter Queue.

For the specific case of a CI/CD pipeline, we don't need retries. CI/CD jobs run for long duration
and a job failure doesn't always translate to a system failure - it could be that the job is not
defined correctly. This is also what popular CI/CD tools do: GitHub, GitLab, etc. This means that we
don't need point 3 and 4.

Dedicated message queues

Now that we know what we need from our queue - or our implementation of one - we can start with the
first solution for a work queue, which is to use a dedicated message queue such as RabbitMQ or Kafka.

A message queue is difficult to get right, and using a dedicated tool for it allows us to move the
complexity of such a tool to the infrastructure, while delivering the guarantees we need. RabbitMQ
and Kafka have dedicated brokers that handle things such as persistence, acknowledgements, retries,
message tracking, and re-delivery.

They might, however, need some tuning to get the features needed - for example, RabbitMQ requires
enabling persistence to allow recovery after crashes, and you also need to configure consumers of
messages to acknowledge messages when they are done.

Kafka can process up to 1 million jobs/second and RabbitMQ around 10K/s.

Redis

Another common way of doing work queues - used by tools like Celery - is to use Redis. Redis is a
key-value store, which means it is not purposefully built to handle message queuing. However it has
a few data structures that allow implementing a message queue that can satisfy the requirements
listed above. We have two options: Lists and Streams.

A common way of implementing message queues in Redis is to use normal and sorted lists. The main
list holds the jobs. After claiming a job, the worker moves it from the main list to a sorted list
acting as the processing list - we'll get into why it needs to be sorted shortly. Once the job is
done, the worker removes it from the sorted list and either deletes it or writes the final result to
a database.

Since the worker explicitly moves the job between lists, we get an at-least-once delivery guarantee -
the worker has confirmed receipt. But if the worker fails mid-job, the job stays stuck in the
processing list. To handle this we need a background process that scans the processing list and
re-queues jobs that are past a time threshold. This is why the list needs to be sorted - it makes it
easier to retrieve the oldest stuck items.

We also don't want to rerun a buggy job forever, so we track failures using a retry field on the job.
Past a certain threshold, the job moves to a dead letter queue for later debugging.

The stream in Redis is also another way of implementing a queue. How it works is similar to Kafka in
that it is a log-based data structure that logs any incoming messages. The stream is a bit more
complicated than plain lists, however, and is more suitable when having a lot of workers that need to
be coordinated in different ways. Since it relies on a log-based mechanism for storage, it requires
compaction which needs to be done manually.

Redis queues can process up to 20K jobs/second.

Postgres & MySQL

In relational databases a job queue is simply another table in your database. This means that you
track many of the states of a given job using fields in your table schema (status), and you just
need to implement the logic around it:

1. Job claiming changes status to processing.
2. Finishing a job changes the status to done/failed depending on the result.
3. Only jobs that are queued/unprocessed will be picked up by idle workers.
4. A background job rechecks for timed out jobs and changes the status to queued/unprocessed/created in case a worker fails before completing a job.

The main challenge with relational databases is competing workers. If both worker A and worker B do
a select to get another job to execute and both get the same job (say with id 1), they will both
claim the job and start processing it.

You might ask can't that also be an issue for Redis? Well... not really, Redis can only execute one
operation at a time so no competing workers.

To solve this we can use SELECT ... FOR UPDATE which would select a row and lock it so that no
other transaction (run by a given worker) can read it. The worker would then update the row to
processing on the same transaction where it claimed it. The lock is then released when the
transaction commits.

There is one issue however: if another worker tries to look for a job to process in another
transaction, it would get blocked on the locked row - this is because relational databases block on
locks by default. In order to avoid this we can use SKIP LOCKED which would skip any locked rows.
So the full statement SELECT ... FOR UPDATE SKIP LOCKED would select a new unlocked row and apply a
lock on the row for the worker to update it.

For the actual implementation a few things are worth noting:

1. Locks can degrade the performance of the database, so it is best to keep transactions as short-lived as possible to release locks - this means you shouldn't process the jobs in the transactions.
2. Consider adding an index to speed up the querying of jobs.

Work queues based on Postgres process up to 12K requests/second. I don't have any information about MySQL.

Discussion

While many of the existing solutions are well suited for implementing a work queue, which one to
reach for depends on your specific use case, and on what infrastructure you already have or how much
you are willing to invest in new infrastructure.

For my current goal with KCX, which is the tool being a single deployable unit, none of the existing
solutions are suitable. They all require adding another component to the system in order to have a
queue. This is why I chose to use SQLite, which would require just an extra volume mounted when
deployed on a Kubernetes cluster.

Final implementation

While being an SQL database, SQLite has the same concurrency model as Redis, at least to some
degree. In its default mode, readers block writers and writers block each other.

This is fine for KCX's use case. SQLite lets me:

• Have a single deployable unit without an external dependency
• Keep write volume low, since CI/CD jobs are naturally long running

However it is still worthwhile to apply some optimizations to improve performance:

1. Enable Write-Ahead Logging (WAL) mode so that writers don't block readers. Writers still block each other, however.
2. In order for a writer to write to the database it must acquire a file (database) level lock. Other writers will receive a busy error when attempting to write in such a situation. In order to avoid the error, we must set a busy_timeout which would allow the writers to wait a bit before throwing an error. Another solution is to allow a single database connection. In order for writers to not block readers - in WAL mode - we need a separate connection for reading.
3. Another situation in which we could have a busy error is if a transaction promotes from a simple read transaction to a write transaction due to the execution of a write operation. The busy_timeout option can solve this but it is better to avoid it altogether. To avoid the error we must start a transaction as a write transaction using BEGIN IMMEDIATE.

I wasn't able to find a reliable source for how much SQLite can process, but one blog post reports
requests in the order of tens/second.

I will be considering moving to another queue system in the future to improve performance, but
SQLite suits my purposes for now.

References

• Postgres Queue Implementation
• MySQL SKIP LOCKED
• Kafka and RabbitMQ Performance Comparison
• SQLite-based Work Queue Guide By River
• Redis Performance Measurement
• Postgres Performance
• SQLite Queue Implementation