diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index 7c0ef72..6837616 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -18,6 +18,7 @@ from __future__ import annotations import heapq import logging +import secrets import threading import time from abc import ABC, abstractmethod @@ -272,12 +273,15 @@ class MemoryWorkQueue(WorkQueue): self._in_flight = 0 self._counter = 0 self._seq = 0 + # The ids become `FlowRun.id`, a primary key. Counting from zero alone, + # a restart would overwrite the previous boot's rows. + self._boot = secrets.token_hex(4) self._lock = threading.Lock() self._wake = threading.Condition(self._lock) def _next_id(self) -> str: self._seq += 1 - return f"mem-{self._seq}" + return f"mem-{self._boot}-{self._seq}" def add(self, item: WorkItem) -> None: with self._wake: diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index 71cb25e..6922a01 100644 --- a/backend/tests/flow/test_queue.py +++ b/backend/tests/flow/test_queue.py @@ -25,6 +25,18 @@ def test_items_come_back_in_the_order_they_went_in(): assert all(item.entry_id for item in claimed) +def test_two_queues_never_hand_out_the_same_id(): + """A restart without Redis must not reuse the last boot's run ids.""" + first, second = MemoryWorkQueue(), MemoryWorkQueue() + for queue in (first, second): + queue.add(WorkItem(kind="cascade", node="f.n", flow="f")) + + ids = [queue.claim(1, 10)[0].entry_id for queue in (first, second)] + + assert ids[0] != ids[1] + assert all(len(entry_id) <= 64 for entry_id in ids) + + def test_claiming_an_empty_queue_waits_and_gives_up(): queue = MemoryWorkQueue() started = time.monotonic()