103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Executable model tests for the proposed bounded GEL drain architecture."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
MAX_ROWS = 4096
|
|
|
|
|
|
class Model:
|
|
def __init__(self, path: Path):
|
|
self.db = sqlite3.connect(path)
|
|
self.db.executescript(
|
|
"PRAGMA journal_mode=MEMORY; PRAGMA synchronous=OFF;"
|
|
"CREATE TABLE outbox(seq INTEGER PRIMARY KEY AUTOINCREMENT,"
|
|
" action TEXT NOT NULL, payload TEXT NOT NULL, coalesce_key TEXT UNIQUE);"
|
|
"CREATE TABLE metrics(key TEXT PRIMARY KEY,value INTEGER NOT NULL);"
|
|
"INSERT INTO metrics VALUES('dropped',0);"
|
|
)
|
|
self.worker_enqueued = False
|
|
self.max_work_specs = 0
|
|
self.max_outbox = 0
|
|
|
|
def enqueue(self, action: str, payload: dict, coalescible: bool) -> None:
|
|
encoded = json.dumps(payload, sort_keys=True)
|
|
key = action if coalescible else None
|
|
if key is not None:
|
|
self.db.execute(
|
|
"INSERT INTO outbox(action,payload,coalesce_key) VALUES(?,?,?) "
|
|
"ON CONFLICT(coalesce_key) DO UPDATE SET payload=excluded.payload",
|
|
(action, encoded, key),
|
|
)
|
|
else:
|
|
count = self.db.execute("SELECT count(*) FROM outbox").fetchone()[0]
|
|
if count >= MAX_ROWS:
|
|
self.db.execute("UPDATE metrics SET value=value+1 WHERE key='dropped'")
|
|
self.db.commit()
|
|
return
|
|
self.db.execute(
|
|
"INSERT INTO outbox(action,payload,coalesce_key) VALUES(?,?,NULL)",
|
|
(action, encoded),
|
|
)
|
|
self.db.commit()
|
|
self.worker_enqueued = True # enqueueUniqueWork(GEL_DRAIN, KEEP)
|
|
self.max_work_specs = max(self.max_work_specs, int(self.worker_enqueued))
|
|
self.max_outbox = max(self.max_outbox,
|
|
self.db.execute("SELECT count(*) FROM outbox").fetchone()[0])
|
|
|
|
def drain(self, limit: int, crash_after_handle: bool = False) -> list[int]:
|
|
handled = []
|
|
selected = self.db.execute(
|
|
"SELECT seq FROM outbox ORDER BY seq LIMIT ?", (limit,)
|
|
).fetchall()
|
|
for (seq,) in selected:
|
|
handled.append(seq)
|
|
if crash_after_handle:
|
|
return handled # no ack: event remains for at-least-once retry
|
|
self.db.execute("DELETE FROM outbox WHERE seq=?", (seq,))
|
|
self.db.commit()
|
|
if self.db.execute("SELECT count(*) FROM outbox").fetchone()[0] == 0:
|
|
self.worker_enqueued = False
|
|
return handled
|
|
|
|
def count(self) -> int:
|
|
return self.db.execute("SELECT count(*) FROM outbox").fetchone()[0]
|
|
|
|
|
|
def main() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
m = Model(Path(tmp) / "outbox.db")
|
|
for i in range(10_000):
|
|
m.enqueue("GATHERING_ALARM_INTENT", {"tick": i}, coalescible=True)
|
|
assert m.count() == 1 and m.max_work_specs == 1
|
|
payload = json.loads(m.db.execute("SELECT payload FROM outbox").fetchone()[0])
|
|
assert payload == {"tick": 9999}
|
|
m.drain(100)
|
|
assert m.count() == 0
|
|
|
|
seen = []
|
|
for i in range(10_000):
|
|
m.enqueue("DURABLE_EVENT", {"seq": i}, coalescible=False)
|
|
if i % 32 == 31:
|
|
seen.extend(m.drain(32))
|
|
seen.extend(m.drain(MAX_ROWS))
|
|
assert m.count() == 0 and m.max_outbox <= 32 and m.max_work_specs == 1
|
|
assert seen == sorted(seen)
|
|
|
|
m.enqueue("DURABLE_EVENT", {"seq": "crash"}, coalescible=False)
|
|
first = m.drain(1, crash_after_handle=True)
|
|
assert m.count() == 1
|
|
second = m.drain(1)
|
|
assert first == second and m.count() == 0
|
|
m.db.close()
|
|
print("PASS: 10k coalescing, 10k FIFO, O(1) WorkSpec, bounded queue, crash retry")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|