Fastpath graceful shutdown not reliable on failure path

[LDiazN]

When the fastpath process gets a sigterm, it stops receiving new measurements and waits for workers to finish all remaining work in the queue.

This approach works well for the usual use case: Everything is working properly and we want to reboot/restart/stop the Fastpath

However, when there's an issue with measurement processing (Clickhouse is unresponsive, for example) this approach presents some issues:

1. Measurement burning: If workers are unable to process the measurement, they will error on every measurement they get from the queue, burning them forever. Measurements won't be stored in the DB, s3 or disk.
2. sigkill escalation: If worker processes are too slow to process measurements on the queue, docker will kill them with a sigkill signal, interrupting the graceful shutdown and losing measurements stored in the in-memory queue

It's worth noting that since these measurements already left the ooniprobe service, they won't be backed up to s3 if they are lost from the queue

Impact

At the moment the maximum queue size is 500, nearly 10s worth of measurements at our current rate. So every time we're in this situation we can expect to lose ~10s worth of measurements

[LDiazN]

An additional challenge with this problem is that we have no way to signal workers that they should stop, other than the queue itself.

On shutdown, we could try to drain the queue and save as many measurements as possible, but some of them will still get burned by the failing workers. Only after the queue is empty we can send the None value to tell workers to stop

One approach could be to change the queue into a priority queue to give more priority to shutdown messages

[LDiazN]

While working on this problem, I found another issue

If we try to take out the content of the queue on shutdown, the process that tries to drain the queue blocks on the first queue.get and the shutdown times out, getting a sigkill from docker and unable to backup the queue content to disk

I made the following minimal example to reproduce this issue, but it works fine. It seems related to something else unique to our setup:

from queue import Empty
import multiprocessing as mp
import os
import signal
import time

queue = mp.Queue()

def worker_fn(queue: mp.Queue):
    item = queue.get(block=True)
    print(f"[{os.getpid()}] Got item: {item}, starting to work....")
    time.sleep(3000)


def main():
    workers = [mp.Process(target=worker_fn, args=(queue,)) for n in range(3)]
    for worker in workers:
        worker.start()

    fork_pid = os.fork()
    if fork_pid == 0:
        # Child process: fill the queue then busy-wait (emulates gunicorn worker)
        N_ELEMENTS = 10
        print(f"[{os.getpid()}] Filler: filling queue with {N_ELEMENTS} items...")
        for i in range(N_ELEMENTS):
            queue.put(i)
        print(f"[{os.getpid()}] Filler: done filling, busy-waiting...")
        while True:
            pass  # busy-wait

    print(f"[{os.getpid()}] Main: waiting for filler to fill the queue...")
    time.sleep(2)

    print(f"[{os.getpid()}] Main: killing forked filler (pid={fork_pid})...")
    os.kill(fork_pid, signal.SIGKILL)
    os.waitpid(fork_pid, 0)
    print(f"[{os.getpid()}] Main: forked filler killed.")

    print("Waiting for workers to pick up some items...")
    time.sleep(5)

    print("Recovering elements from the queue that were not processed...")
    remaining = []
    while True:
        try:
            remaining.append(queue.get(block=False))
        except Empty:
            break

    print(f"Recovered {len(remaining)} elements from the queue!")
    print("Killing workers...")
    for worker in workers:
        worker.terminate()
    print("Workers killed!")


if __name__ == "__main__":
    main()