"""Receipts must survive deletion and preserve aged pending work during pruning.""" import json import sqlite3 import time import unittest import urllib.error from harness import Albedo, Provider, exclusive, operation_id, text RETENTION_MS = 7 * 24 * 60 * 60 * 1000 @exclusive class OperationRetentionTests(unittest.TestCase): def setUp(self): self.provider = Provider(lambda request: text("done")) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses", prepare=self.prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) def prepare(self, app): app.env["ALBEDO_SCHEDULE_TICK_MS"] = "50" def database(self): return sqlite3.connect(self.app.home / "albedo.sqlite", timeout=10) def post(self, path, request): with self.app.api(path, request) as response: return json.load(response) def receipt(self, operation): with self.app.api(f"/operations/{operation}") as response: return json.load(response) def wait_delivery(self, operation, expected): deadline = time.monotonic() + 35 while time.monotonic() < deadline: receipt = self.receipt(operation) if receipt["deliveryStatus"] == expected: return receipt time.sleep(0.05) self.fail(f"operation did not become {expected}: {receipt}") def test_terminal_receipt_expiry_prevents_reexecution(self): operation = operation_id() request = { "operationId": operation, "workspace": str(self.app.workspace), "provider": self.app.profile, } created = self.post("/sessions", request) old_time = int(time.time() * 1000) - RETENTION_MS - 10000 aged_operation = operation_id(old_time) with self.database() as db: db.execute( "UPDATE operations SET id=?,created_at=?,terminal_at=? WHERE id=?", (aged_operation, old_time, old_time, operation), ) deadline = time.monotonic() + 10 while time.monotonic() < deadline: try: self.receipt(aged_operation) except urllib.error.HTTPError as failure: self.assertEqual(failure.code, 410) self.assertEqual(json.load(failure)["code"], "operation_expired") break time.sleep(0.05) else: self.fail("terminal receipt did not expire") request["operationId"] = aged_operation with self.assertRaises(urllib.error.HTTPError) as failure: self.post("/sessions", request) self.assertEqual(failure.exception.code, 410) with self.database() as db: self.assertEqual( db.execute("SELECT id FROM sessions").fetchall(), [(created["id"],)] ) def test_aged_pending_receipt_remains_recoverable_after_restart(self): session = self.app.session() operation = operation_id() request = {"operationId": operation, "content": "old pending input"} with self.database() as db: db.execute( "CREATE TRIGGER hold_pending BEFORE INSERT ON transcript " "WHEN NEW.row_class='user' BEGIN SELECT RAISE(ABORT,'fixture'); END" ) accepted = self.post(f"/sessions/{session}/events", request) self.assertEqual(self.receipt(operation)["deliveryStatus"], "pending") old_time = int(time.time() * 1000) - RETENTION_MS - 10000 aged_operation = operation_id(old_time) def age_pending(app): with self.database() as db: payload = db.execute( "SELECT payload FROM pending_inputs WHERE operation_id=?", (operation,), ).fetchone()[0] payload = payload.replace(operation.encode(), aged_operation.encode()) db.execute( "UPDATE operations SET id=?,created_at=? WHERE id=?", (aged_operation, old_time, operation), ) db.execute( "UPDATE pending_inputs SET operation_id=?,payload=?,accepted_at=? " "WHERE operation_id=?", (aged_operation, payload, old_time, operation), ) self.app.restart(crash=True, prepare=age_pending) request["operationId"] = aged_operation accepted["operationId"] = aged_operation # Admission response is persisted; the fixture also updates its action ID. with self.database() as db: response = db.execute( "SELECT response FROM operations WHERE id=?", (aged_operation,) ).fetchone()[0] response = response.replace(operation, aged_operation) db.execute( "UPDATE operations SET response=? WHERE id=?", (response, aged_operation), ) self.assertEqual(self.receipt(aged_operation)["deliveryStatus"], "pending") self.assertEqual(self.post(f"/sessions/{session}/events", request), accepted) with self.database() as db: db.execute("DROP TRIGGER hold_pending") self.wait_delivery(aged_operation, "committed") self.app.idle(session) self.app.restart(crash=True) self.assertEqual(self.receipt(aged_operation)["deliveryStatus"], "committed") with self.database() as db: self.assertEqual( db.execute( "SELECT count(*) FROM transcript WHERE session=? AND row_class='user'", (session,), ).fetchone()[0], 1, ) def test_deleted_session_keeps_creation_and_cancelled_input_receipts(self): creation = operation_id() create_request = { "operationId": creation, "workspace": str(self.app.workspace), "provider": self.app.profile, } created = self.post("/sessions", create_request) session = created["id"] with self.database() as db: db.execute( "CREATE TRIGGER hold_pending BEFORE INSERT ON transcript " "WHEN NEW.row_class='user' BEGIN SELECT RAISE(ABORT,'fixture'); END" ) operation = operation_id() request = {"operationId": operation, "content": "cancel when deleted"} accepted = self.post(f"/sessions/{session}/events", request) self.app.idle(session) with self.app.api(f"/sessions/{session}", method="DELETE") as response: json.load(response) self.assertEqual(self.receipt(operation)["deliveryStatus"], "cancelled") self.app.restart(crash=True) self.assertEqual(self.post("/sessions", create_request), created) self.assertEqual(self.receipt(creation)["target"], session) self.assertEqual(self.post(f"/sessions/{session}/events", request), accepted) self.assertEqual(self.receipt(operation)["deliveryStatus"], "cancelled") with self.database() as db: self.assertEqual( db.execute("SELECT count(*) FROM sessions").fetchone()[0], 0 ) self.assertEqual( db.execute("SELECT count(*) FROM pending_inputs").fetchone()[0], 0 ) def test_continuation_marker_survives_receipt_pruning_and_restart(self): session = self.app.session() self.app.prompt(session, "start").close() self.app.idle(session) operation = operation_id() request = {"operationId": operation, "type": "continue"} self.post(f"/sessions/{session}/events", request) self.wait_delivery(operation, "committed") self.app.idle(session) old_time = int(time.time() * 1000) - RETENTION_MS - 10000 with self.database() as db: db.execute( "UPDATE operations SET terminal_at=? WHERE id=?", (old_time, operation), ) deadline = time.monotonic() + 10 while time.monotonic() < deadline: try: self.receipt(operation) except urllib.error.HTTPError as failure: self.assertEqual(failure.code, 404) break time.sleep(0.05) else: self.fail("terminal receipt did not expire") self.app.restart(crash=True) with self.database() as db: marker = db.execute( "SELECT session,seq FROM continuation_markers WHERE operation_id=?", (operation,), ).fetchall() self.assertEqual(len(marker), 1) self.assertEqual(marker[0][0], session) self.assertGreater(marker[0][1], 0) self.assertEqual( db.execute( "SELECT count(*) FROM transcript WHERE session=? AND row_class='user'", (session,), ).fetchone()[0], 1, ) self.assertEqual(len(self.provider.requests), 2) with self.app.api(f"/sessions/{session}", method="DELETE") as response: json.load(response) with self.database() as db: self.assertEqual( db.execute( "SELECT count(*) FROM continuation_markers WHERE operation_id=?", (operation,), ).fetchone()[0], 0, )