Memory-queue ids carry a per-instance boot token, so a restart without Redis no longer overwrites earlier run rows

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:

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()