507 lines
16 KiB
Python
507 lines
16 KiB
Python
"""Live state and queued work following a renamed flow key (sections 1.2, 1.3
|
|
and 3 of docs/private/id_refac_rename.md). Memory always, Redis with
|
|
FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""
|
|
|
|
import os
|
|
import threading
|
|
import time
|
|
import uuid
|
|
|
|
import pytest
|
|
import redis
|
|
|
|
from fluksio.flow import state as state_module
|
|
from fluksio.flow.namespace import HISTORY, NAMES, Rename
|
|
from fluksio.flow.queue import GROUP, MemoryWorkQueue, RedisWorkQueue, WorkItem
|
|
from fluksio.flow.state import MemoryState, RedisState
|
|
|
|
HEAT = Rename("heat", "warm")
|
|
BACK = Rename("warm", "heat")
|
|
NOTHING = dict.fromkeys(("queue_items", "delayed", "parked", "done_markers"), 0)
|
|
|
|
|
|
def _server() -> tuple[str, int]:
|
|
host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
|
|
if not host:
|
|
pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
|
|
return host, int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379"))
|
|
|
|
|
|
def _drop(client, ns: str) -> None:
|
|
keys = list(client.scan_iter(match=f"{ns}:*"))
|
|
if keys:
|
|
client.delete(*keys)
|
|
|
|
|
|
def _redis_queue(ns: str, consumer: str) -> RedisWorkQueue:
|
|
host, port = _server()
|
|
return RedisWorkQueue(host, port, namespace=ns, consumer=consumer)
|
|
|
|
|
|
@pytest.fixture
|
|
def redis_queue():
|
|
queue = _redis_queue(f"test-rename-{uuid.uuid4().hex}", "engine-a")
|
|
try:
|
|
yield queue
|
|
finally:
|
|
_drop(queue._redis, queue._ns)
|
|
queue.close()
|
|
|
|
|
|
@pytest.fixture(params=["memory", "redis"])
|
|
def queue(request):
|
|
if request.param == "memory":
|
|
yield MemoryWorkQueue()
|
|
else:
|
|
yield request.getfixturevalue("redis_queue")
|
|
|
|
|
|
@pytest.fixture(params=["memory", "redis"])
|
|
def state(request):
|
|
if request.param == "memory":
|
|
yield MemoryState()
|
|
return
|
|
host, port = _server()
|
|
state = RedisState(host, port, namespace=f"test-rename-{uuid.uuid4().hex}", ttl=600)
|
|
try:
|
|
yield state
|
|
finally:
|
|
_drop(state._client, state._namespace)
|
|
state.close()
|
|
|
|
|
|
def item(node: str, kind: str = "cascade", **fields) -> WorkItem:
|
|
return WorkItem(kind=kind, node=node, flow=node.split(".")[0], **fields)
|
|
|
|
|
|
def _fill(queue) -> WorkItem:
|
|
"""Work in every place a queue keeps it, naming heat or not. Returns the
|
|
item left claimed."""
|
|
queue.add(item("heat.acked"))
|
|
queue.ack(*queue.claim(1, 10))
|
|
queue.add(item("heat.valve", outputs={"heat.temp": 21}))
|
|
queue.add(item("hall.lamp", outputs={"heat.target": 21, "hall.on": True}))
|
|
queue.add(item("heating.valve"))
|
|
queue.add(WorkItem(kind="run", node="", flow="heat", run_id="r1"))
|
|
queue.add_delayed(item("heat.pump", kind="flush"), 100.0)
|
|
queue.move_due(150.0)
|
|
(claimed,) = queue.claim(1, 10, due_only=True)
|
|
queue.mark_done(claimed.entry_id, "heat.pump")
|
|
queue.mark_done("other", "heating.valve")
|
|
queue.add_delayed(item("heat.limit", guard_key="generation", guard_value="3"), 9e9)
|
|
queue.add_delayed(item("heating.limit"), 9e9 + 1)
|
|
queue.park("heat", item("heat.p"))
|
|
queue.park("hall", item("hall.q", outputs={"heat.x": 1}))
|
|
queue.dead_letter(item("heat.dead"), "3 deliveries")
|
|
return claimed
|
|
|
|
|
|
def _dump(queue) -> object:
|
|
"""Everything a queue holds, apart from how long it has been idle."""
|
|
if isinstance(queue, MemoryWorkQueue):
|
|
return (
|
|
list(queue._items),
|
|
list(queue._due),
|
|
list(queue._delayed),
|
|
dict(queue._parked),
|
|
list(queue._done),
|
|
list(queue._held),
|
|
)
|
|
r = queue._redis
|
|
dump = {}
|
|
for key in sorted(r.scan_iter(match=f"{queue._ns}:*")):
|
|
kind = r.type(key)
|
|
if kind == "stream":
|
|
info = r.xinfo_stream(key)
|
|
groups = r.xinfo_groups(key)
|
|
dump[key] = (
|
|
r.xrange(key),
|
|
[info[k] for k in ("last-generated-id", "entries-added")],
|
|
[
|
|
{k: g[k] for k in ("last-delivered-id", "entries-read", "lag")}
|
|
for g in groups
|
|
],
|
|
[
|
|
(p["message_id"], p["consumer"], p["times_delivered"])
|
|
for p in r.xpending_range(key, GROUP, "-", "+", 100)
|
|
]
|
|
if groups
|
|
else [],
|
|
)
|
|
elif kind == "zset":
|
|
dump[key] = r.zrange(key, 0, -1, withscores=True)
|
|
elif kind == "list":
|
|
dump[key] = r.lrange(key, 0, -1)
|
|
elif kind == "set":
|
|
dump[key] = sorted(r.smembers(key))
|
|
else:
|
|
dump[key] = (r.get(key), r.ttl(key) > 3500)
|
|
return dump
|
|
|
|
|
|
def test_queued_work_moves_to_the_new_names(queue):
|
|
claimed = _fill(queue)
|
|
before = queue.live_items()
|
|
# Memory hands a claimed item out and forgets it.
|
|
in_stream = 3 if isinstance(queue, RedisWorkQueue) else 2
|
|
assert queue.count(HEAT) == {
|
|
"queue_items": in_stream,
|
|
"delayed": 1,
|
|
"parked": 2,
|
|
"done_markers": 1,
|
|
}
|
|
|
|
queue.rename(HEAT)
|
|
|
|
def order(one):
|
|
return one.entry_id, one.node, one.enqueued_at
|
|
|
|
# Ids, deadlines, guards and everything else but the names stay.
|
|
assert sorted(queue.live_items(), key=order) == sorted(
|
|
[HEAT.item(one) or one for one in before], key=order
|
|
)
|
|
assert queue.count(HEAT) == NOTHING
|
|
assert queue.was_done(claimed.entry_id, "warm.pump")
|
|
assert not queue.was_done(claimed.entry_id, "heat.pump")
|
|
assert queue.was_done("other", "heating.valve")
|
|
assert queue.next_due() == 9e9
|
|
assert queue.move_due(9e9) == 1
|
|
assert [one.node for one in queue.claim(10, 10)] == [
|
|
"warm.limit",
|
|
"warm.valve",
|
|
"hall.lamp",
|
|
"heating.valve",
|
|
"",
|
|
]
|
|
assert queue.unpark("heat") == []
|
|
assert [one.node for one in queue.unpark("warm")] == ["warm.p"]
|
|
assert [one.outputs for one in queue.unpark("hall")] == [{"warm.x": 1}]
|
|
if isinstance(queue, RedisWorkQueue):
|
|
# History stays as it was.
|
|
assert [d["node"] for d in queue.dead_letters()] == ["heat.dead"]
|
|
assert queue._redis.xrange(queue._stream)[0][1]["node"] == "heat.acked"
|
|
|
|
|
|
def test_a_second_run_changes_nothing_and_renaming_back_restores_the_queue(queue):
|
|
_fill(queue)
|
|
queue.hold()
|
|
queue.add(item("heat.held"))
|
|
before = _dump(queue)
|
|
|
|
queue.rename(HEAT)
|
|
once = _dump(queue)
|
|
queue.rename(HEAT)
|
|
|
|
assert once != before
|
|
assert _dump(queue) == once
|
|
queue.rename(BACK)
|
|
assert _dump(queue) == before
|
|
|
|
|
|
def test_held_work_is_released_in_order_behind_older_work(queue):
|
|
queue.add(item("hall.a"))
|
|
queue.hold()
|
|
queue.add(item("heat.b"))
|
|
queue.add(item("hall.c"))
|
|
|
|
assert queue.held
|
|
assert queue.count(HEAT)["queue_items"] == 1
|
|
assert {one.node for one in queue.live_items()} == {"hall.a", "heat.b", "hall.c"}
|
|
queue.rename(HEAT)
|
|
queue.release()
|
|
queue.add(item("hall.d"))
|
|
|
|
assert not queue.held
|
|
assert [one.node for one in queue.claim(10, 10)] == [
|
|
"hall.a",
|
|
"warm.b",
|
|
"hall.c",
|
|
"hall.d",
|
|
]
|
|
|
|
|
|
def test_a_hold_waits_for_an_add_already_writing_to_the_stream(
|
|
redis_queue, monkeypatch
|
|
):
|
|
"""Adds do not wait for each other, but a hold waits for them, so no
|
|
entry lands in a stream after the rebuild has read it."""
|
|
writing, finish = threading.Event(), threading.Event()
|
|
xadd = redis_queue._redis.xadd
|
|
|
|
def slow(*args, **kwargs):
|
|
writing.set()
|
|
finish.wait(5)
|
|
return xadd(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(redis_queue._redis, "xadd", slow)
|
|
adding = threading.Thread(target=redis_queue.add, args=(item("heat.a"),))
|
|
adding.start()
|
|
assert writing.wait(5)
|
|
holding = threading.Thread(target=redis_queue.hold)
|
|
holding.start()
|
|
holding.join(0.2)
|
|
assert holding.is_alive()
|
|
|
|
finish.set()
|
|
holding.join(5)
|
|
adding.join(5)
|
|
assert redis_queue.count(HEAT)["queue_items"] == 1
|
|
assert redis_queue.held
|
|
redis_queue.release()
|
|
|
|
|
|
def test_a_guarded_timer_still_finds_its_guard(queue):
|
|
state = MemoryState()
|
|
state["__node__:heat.limit:generation"] = 3
|
|
queue.add_delayed(item("heat.limit", guard_key="generation", guard_value="3"), 1.0)
|
|
|
|
queue.rename(HEAT)
|
|
state.rename_keys(HEAT)
|
|
queue.move_due(1.0)
|
|
|
|
(due,) = queue.claim(1, 10)
|
|
assert (due.node, due.guard_key, due.guard_value) == (
|
|
"warm.limit",
|
|
"generation",
|
|
"3",
|
|
)
|
|
# What the execution service compares it with (`Node.recall`).
|
|
assert str(state.get(f"__node__:{due.node}:{due.guard_key}")) == due.guard_value
|
|
|
|
|
|
def test_a_pending_item_keeps_its_id_owner_and_count_and_is_not_repeated(
|
|
redis_queue,
|
|
):
|
|
redis_queue.add(item("heat.valve"))
|
|
(claimed,) = redis_queue.claim(1, 10)
|
|
# A second delivery, as a reclaim after a crash would make it.
|
|
redis_queue._redis.xclaim(
|
|
redis_queue._stream, GROUP, "engine-a", 0, [claimed.entry_id]
|
|
)
|
|
redis_queue.mark_done(claimed.entry_id, "heat.valve")
|
|
|
|
redis_queue.rename(HEAT)
|
|
|
|
(row,) = redis_queue._redis.xpending_range(redis_queue._stream, GROUP, "-", "+", 10)
|
|
assert (row["message_id"], row["consumer"], row["times_delivered"]) == (
|
|
claimed.entry_id,
|
|
"engine-a",
|
|
2,
|
|
)
|
|
other = _redis_queue(redis_queue._ns, "engine-b")
|
|
try:
|
|
(back,) = other.reclaim_stale(0)
|
|
# What the pipeline asks before it runs a side effect again.
|
|
assert other.was_done(back.entry_id, back.node)
|
|
finally:
|
|
other.close()
|
|
assert (back.entry_id, back.node, back.deliveries) == (
|
|
claimed.entry_id,
|
|
"warm.valve",
|
|
3,
|
|
)
|
|
marker = redis_queue._done_key(claimed.entry_id, "warm.valve")
|
|
assert 3500 < redis_queue._redis.ttl(marker) <= 3600
|
|
|
|
|
|
def test_a_crash_before_a_stream_is_swapped_in_and_a_rerun_converge(
|
|
redis_queue, monkeypatch
|
|
):
|
|
_fill(redis_queue)
|
|
before = _dump(redis_queue)
|
|
|
|
def crash(*args):
|
|
raise redis.ConnectionError
|
|
|
|
with monkeypatch.context() as patch:
|
|
patch.setattr(redis_queue._redis, "rename", crash)
|
|
with pytest.raises(redis.ConnectionError):
|
|
redis_queue.rename(HEAT)
|
|
after = _dump(redis_queue)
|
|
assert after.pop(f"{redis_queue._due_stream}:rename")
|
|
assert after == before
|
|
|
|
redis_queue.rename(HEAT)
|
|
assert redis_queue.count(HEAT) == NOTHING
|
|
redis_queue.rename(BACK)
|
|
assert _dump(redis_queue) == before
|
|
|
|
|
|
def test_work_held_while_the_held_list_is_rewritten_survives(redis_queue, monkeypatch):
|
|
redis_queue.hold()
|
|
redis_queue.add(item("heat.a"))
|
|
redis_queue.add(item("hall.b"))
|
|
lrange = redis_queue._redis.lrange
|
|
|
|
def racing(key, *args):
|
|
raw = lrange(key, *args)
|
|
if key == redis_queue._held_key and len(raw) == 2:
|
|
redis_queue.add(item("heat.c"))
|
|
return raw
|
|
|
|
with monkeypatch.context() as patch:
|
|
patch.setattr(redis_queue._redis, "lrange", racing)
|
|
redis_queue.rename(HEAT)
|
|
redis_queue.release()
|
|
|
|
assert [one.node for one in redis_queue.claim(10, 10)] == [
|
|
"warm.a",
|
|
"hall.b",
|
|
"heat.c",
|
|
]
|
|
|
|
|
|
def test_a_held_list_left_by_a_stopped_process_is_released_at_startup(
|
|
redis_queue,
|
|
):
|
|
redis_queue.hold()
|
|
for i in range(150):
|
|
redis_queue.add(item(f"hall.n{i}"))
|
|
|
|
fresh = _redis_queue(redis_queue._ns, "engine-b")
|
|
try:
|
|
assert not fresh.held
|
|
assert fresh.release_held() == 150
|
|
assert fresh.release_held() == 0
|
|
claimed = fresh.claim(200, 10)
|
|
finally:
|
|
fresh.close()
|
|
assert [one.node for one in claimed] == [f"hall.n{i}" for i in range(150)]
|
|
|
|
|
|
def _fill_state(state) -> None:
|
|
state.record(
|
|
{"heat.temp": 21.5, "hall.temp": 19.0, "heating.temp": 30.0},
|
|
{"__ts__:heat.temp": 1.0, "__ts__:hall.temp": 1.0},
|
|
["__version__:heat.temp", "__version__:hall.temp"],
|
|
ts=1.0,
|
|
limits={"heat.temp": 5},
|
|
)
|
|
for i in range(2, 8):
|
|
state.append_history({"heat.temp": float(i)}, float(i), {"heat.temp": 5})
|
|
state.update(
|
|
{
|
|
"__last_seen__:heat.valve:hall.temp": 1,
|
|
"__last_seen__:hall.lamp:heat.temp": 2,
|
|
"__in_ts__:heat.valve:heat.temp": 3.0,
|
|
"__held__:heat.out": {"heat.temp": 1},
|
|
"__flush__:heat.valve": 4.0,
|
|
# A node's own state may mention any name: only its key moves.
|
|
"__node__:heat.limit:wait:heat.temp": {"last": "heat.temp"},
|
|
"__node__:hall.limit:wait:heat.temp": 5,
|
|
}
|
|
)
|
|
|
|
|
|
def _state_dump(state) -> object:
|
|
keys = sorted(state.keys())
|
|
names = sorted(state.message_names())
|
|
return (
|
|
keys,
|
|
{k: state.get(k) for k in keys if not k.startswith((HISTORY, NAMES))},
|
|
names,
|
|
{name: state.history(name) for name in names},
|
|
)
|
|
|
|
|
|
def test_state_moves_to_the_new_names_with_its_type_ttl_and_capacity(state):
|
|
_fill_state(state)
|
|
|
|
assert state.count(HEAT) == 10
|
|
assert state.rename_keys(HEAT) == 10
|
|
|
|
assert state.count(HEAT) == 0
|
|
assert sorted(state.message_names()) == ["hall.temp", "heating.temp", "warm.temp"]
|
|
assert not state.exists("heat.temp")
|
|
assert state.get_multi(
|
|
[
|
|
"warm.temp",
|
|
"__version__:warm.temp",
|
|
"__last_seen__:warm.valve:hall.temp",
|
|
"__last_seen__:hall.lamp:warm.temp",
|
|
"__in_ts__:warm.valve:warm.temp",
|
|
"__held__:warm.out",
|
|
"__flush__:warm.valve",
|
|
"__node__:warm.limit:wait:heat.temp",
|
|
"__node__:hall.limit:wait:heat.temp",
|
|
]
|
|
) == {
|
|
"warm.temp": 21.5,
|
|
"__version__:warm.temp": 1,
|
|
"__last_seen__:warm.valve:hall.temp": 1,
|
|
"__last_seen__:hall.lamp:warm.temp": 2,
|
|
"__in_ts__:warm.valve:warm.temp": 3.0,
|
|
"__held__:warm.out": {"heat.temp": 1},
|
|
"__flush__:warm.valve": 4.0,
|
|
"__node__:warm.limit:wait:heat.temp": {"last": "heat.temp"},
|
|
"__node__:hall.limit:wait:heat.temp": 5,
|
|
}
|
|
assert state.history("warm.temp") == [(float(i), float(i)) for i in range(3, 8)]
|
|
if isinstance(state, MemoryState):
|
|
assert state._history["warm.temp"].maxlen == 5
|
|
else:
|
|
history = state._key(f"{HISTORY}:warm.temp")
|
|
assert state._client.type(history) == b"list"
|
|
for key in (
|
|
history,
|
|
state._key("warm.temp"),
|
|
state._key("__flush__:warm.valve"),
|
|
):
|
|
assert 590 < state._client.ttl(key) <= 600
|
|
|
|
|
|
def test_renaming_state_twice_changes_nothing_and_back_restores_it(state):
|
|
_fill_state(state)
|
|
before = _state_dump(state)
|
|
|
|
state.rename_keys(HEAT)
|
|
once = _state_dump(state)
|
|
|
|
assert state.rename_keys(HEAT) == 0
|
|
assert _state_dump(state) == once
|
|
state.rename_keys(BACK)
|
|
assert _state_dump(state) == before
|
|
|
|
|
|
def test_a_state_rename_stopped_after_a_batch_completes_on_a_rerun(state, monkeypatch):
|
|
monkeypatch.setattr(state_module, "RENAME_BATCH", 3)
|
|
_fill_state(state)
|
|
before = _state_dump(state)
|
|
|
|
def stop(_step):
|
|
raise ConnectionError
|
|
|
|
with pytest.raises(ConnectionError):
|
|
state.rename_keys(HEAT, progress=stop)
|
|
state.rename_keys(HEAT)
|
|
|
|
assert state.count(HEAT) == 0
|
|
assert sorted(state.message_names()) == ["hall.temp", "heating.temp", "warm.temp"]
|
|
state.rename_keys(BACK)
|
|
assert _state_dump(state) == before
|
|
|
|
|
|
def test_state_is_never_moved_onto_a_key_that_exists(state):
|
|
state["heat.temp"] = 1
|
|
state["warm.temp"] = 2
|
|
|
|
with pytest.raises(RuntimeError, match="exists"):
|
|
state.rename_keys(HEAT)
|
|
assert state.get("warm.temp") == 2
|
|
|
|
|
|
def test_another_engine_reading_the_queue_is_seen(redis_queue):
|
|
assert redis_queue.server_version() >= (7, 0)
|
|
redis_queue.claim(1, 10)
|
|
assert redis_queue.other_consumers(10_000) == []
|
|
|
|
other = _redis_queue(redis_queue._ns, "engine-b")
|
|
try:
|
|
other.claim(1, 10)
|
|
assert redis_queue.other_consumers(10_000) == ["engine-b"]
|
|
# Gone quiet for longer than asked about: a stopped engine.
|
|
time.sleep(0.05)
|
|
assert redis_queue.other_consumers(10) == []
|
|
finally:
|
|
other.close()
|