feat(daemon): give submissions a durable receipt
Keep each accepted input in line across retries and restarts. Bound the queue and keep image payloads out of display metadata.
177 unmodified linesExpand all17817918018118218318218318418518618748 unmodified lines177 unmodified linesExpand all end), Check = self(), {Reader, ReaderMonitor} = spawn_monitor(fun() -> State = sys:get_state(Pid), %% session_state.State currently has 20 fields: history is field 6. state = element(1, State), 21 = tuple_size(State), %% history is field 6 in session_state.State. state = element(1, State), Check ! {checked, element(7, State) =:= none, contains(State, <<"REPLAY_SENTINEL:">>)} end), {Unloaded, Reachable} = receive48 unmodified lines151 unmodified linesExpand all1521531541551561561571581591601611621631641651661671681 unmodified line17017117217316616716817417517617717817918018118218318418518618718818919019119219319419519619719841 unmodified lines151 unmodified linesExpand all }) let assert Ok(_) = conversation.initialise(ledger) conversation.load_entries(ledger, "timestamped") |> should.equal( Ok([transcript.Entry(types.User("legacy"), None, None, None)]), Ok([ transcript.Entry( types.User("legacy"), None, None, None, Some(transcript.SourceRef("timestamped", 1)), ), ]), )
let new_inputs = [types.User("current"), types.Assistant("answer")] let assert Ok(timestamp) =1 unmodified line let assert True = timestamp > 1_000_000_000_000 let assert Ok(entries) = conversation.load_entries(ledger, "timestamped") entries |> should.equal([ transcript.Entry(types.User("legacy"), None, None, None), transcript.Entry(types.User("current"), Some(timestamp), None, None), transcript.Entry(types.Assistant("answer"), Some(timestamp), None, None), transcript.Entry( types.User("legacy"), None, None, None, Some(transcript.SourceRef("timestamped", 1)), ), transcript.Entry( types.User("current"), Some(timestamp), None, None, Some(transcript.SourceRef("timestamped", 2)), ), transcript.Entry( types.Assistant("answer"), Some(timestamp), None, None, Some(transcript.SourceRef("timestamped", 3)), ), ]) conversation.load(ledger, "timestamped") |> should.equal(Ok([types.User("legacy"), ..new_inputs]))
41 unmodified lines98 unmodified lines99100101102103104105106107108109110111254 unmodified linesExpand all3663673683693703713723733743753763773723733743783793803813823833843856 unmodified lines392393394395389396397398399400358 unmodified linesExpand all759760761762756757763764765766767768127 unmodified linesExpand all98 unmodified lines
def command(self, name, **args): return self.post(f"/sessions/{self.sid}/commands", {"name": name, **args})
def activate(self, name, **args): return self.post( f"/sessions/{self.sid}/events", {"type": "skill", "name": name, **args} )
def turn(self, prompt, session=None, *, expected=1): sid = session or self.sid before = len(self.provider.requests) self.app.prompt(sid, prompt).close()254 unmodified linesExpand all ], "modelCallable": True, "userTurn": True, "page": False, "skill": True, } ], ) self.assertTrue( {"/model", "/context", "/compact"} <= {item["name"] for item in catalog} ) before = len(self.provider.requests) self.assertEqual( self.command("/demo", arguments="one two", clientId="fixture-client"), {"submitted": True}, self.assertTrue( self.activate("/demo", arguments="one two", clientId="fixture-client")[ "ok" ] ) self.app.idle(self.sid) self.assertEqual(len(self.provider.requests), before + 1) activation = json.dumps(self.provider.requests[-1]["request"]["input"])6 unmodified lines ) self.assertEqual(self.get(f"/sessions/{self.sid}/commands"), catalog) self.skill.write_text(SKILL) before = len(self.provider.requests) self.command("/demo", arguments="slash argument") self.activate("/demo", arguments="slash argument") self.app.idle(self.sid) self.assertEqual(len(self.provider.requests), before + 1) user = next( item["content"]358 unmodified linesExpand all self.assertFalse( next(item["enabled"] for item in disabled if item["name"] == "skills") ) self.rejected( f"/sessions/{self.sid}/commands", {"name": "/demo", "arguments": "must not run"}, f"/sessions/{self.sid}/events", {"type": "skill", "name": "/demo", "arguments": "must not run"}, ) pid = self.app.connection["pid"] (request,) = self.turn("extension disabled") self.assertIn("<available_skills>", request["instructions"])127 unmodified linesExpand all44 unmodified lines454647484950515253545556575859606170 unmodified lines132133134135136137138139140141142143144145146147713 unmodified linesExpand all86186286386486586686786886987087187287387464 unmodified lines44 unmodified linesimport jsonimport osfrom pathlib import Pathimport shutilimport secretsimport signalimport subprocessimport sysimport tempfileimport threadingimport timeimport urllib.errorimport uuid
ROOT = Path(__file__).resolve().parents[2]sys.path.insert(0, str(ROOT / "test"))import scratch # noqa: E40270 unmodified lines# Guards the provider registry and the shared provider server._config_lock = threading.RLock()
def operation_id(timestamp=None): """Generate one UUIDv7 action identity, optionally at a fixture time.""" milliseconds = int(time.time() * 1000) if timestamp is None else timestamp bits = (milliseconds << 80) | (7 << 76) | secrets.randbits(76) bits = (bits & ~(3 << 62)) | (2 << 62) return str(uuid.UUID(int=bits))
def exclusive(target): """Run the test, or every test of the class, on a daemon of its own that boots after the fixture's prepare step and is discarded afterwards: it may change global settings or restart the daemon."""713 unmodified linesExpand all def _fail(self, message): self.daemon._fail(message)
def api(self, path, body=None, *, method=None): if ( isinstance(body, dict) and method in (None, "POST") and (path == "/sessions" or path.endswith("/events")) ): body = {"operationId": operation_id(), **body} return self.daemon.api(path, body, method=method)
def cli(self, *args): return self.daemon.cli(*args)64 unmodified lines161 unmodified linesExpand all162163164165166167168169170171172173174175176177178106 unmodified linesExpand all161 unmodified linesExpand all if event.get("source") == "image scaled" ] self.assertEqual(len(notes), 1, notes) self.assertIn(NOTE, notes[0]) original = [ event for event in app.events(session) if event.get("type") == "user" and event.get("text") == "look" ] self.assertEqual(len(original), 1) self.assertEqual(original[0]["image"]["width"], 1200) self.assertEqual(original[0]["image"]["height"], 10) self.assertEqual(original[0]["image"]["bytes"], len(WIDE))
source = hashlib.sha256(WIDE_DATA.encode()).hexdigest() database = f"file:{app.home / 'albedo.sqlite'}?mode=ro"
106 unmodified linesExpand all5 unmodified lines6789101011121314375 unmodified linesExpand all39039139239339439439539639739821 unmodified lines42042142242342442542442542642742842943043143243043343443543643743843944044143644244344444544636 unmodified lines4834844854864814824834844854864874884894904914924934944954969 unmodified lines506507508509504510511512513514515516517518519520521473 unmodified linesExpand all5 unmodified linesimport timeimport unittestimport urllib.error
from harness import Albedo, Provider, exclusive, python, textfrom harness import Albedo, Provider, exclusive, operation_id, python, text
def contents(item): content = item.get("content", "")375 unmodified linesExpand all {"content": None}, {"content": 4}, {"content": "must not submit", "clientId": False}, {"type": "continue", "content": False}, {"type": "resume", "image": {}}, {"type": "continue", "image": {}}, { "content": "must not submit", "image": { "mimeType": "image/png",21 unmodified lines session = app.session() fixture = json.loads( (Path(__file__).parents[1] / "fixtures/mutation_responses.json").read_text() )["submission"] for kind in (None, "user", "continue", "resume"): payload = {"content": "valid submission " + str(kind)} for kind in (None, "user", "continue"): payload = { "operationId": operation_id(), "content": "valid submission " + str(kind), } if kind is not None: payload["type"] = kind with app.api(f"/sessions/{session}/events", payload) as response: self.assertEqual(response.status, fixture["status"]) self.assertEqual(json.load(response), fixture["body"]) self.assertEqual( json.load(response), {**fixture["body"], "operationId": payload["operationId"]}, ) app.idle(session) users = [ event["text"] for event in app.events(session) if event["type"] == "user" ] self.assertEqual(users, ["valid submission None", "valid submission user"]) self.assertGreaterEqual(len(self.provider.requests), 4) self.assertGreaterEqual(len(self.provider.requests), 3)
def test_busy_messages_join_next_model_request_in_order(self): app = self.app_for("responses") session = app.session()36 unmodified lines self.assertEqual(Path((old / "cwd-probe").read_text()).resolve(), old.resolve()) old.rename(moved) pid = app.connection["pid"] before = app.events(session) with self.assertRaises(urllib.error.HTTPError) as caught: app.prompt(session, "workspace probe after move").close() self.assertEqual(caught.exception.code, 409) failure = json.load(caught.exception) self.assertEqual(failure["code"], "workspace_missing") self.assertEqual(failure["workspace"], str(old)) operation = operation_id() request = {"operationId": operation, "content": "workspace probe after move"} with app.api(f"/sessions/{session}/events", request) as response: self.assertEqual(response.status, 202) receipt = self.read(app, f"/operations/{operation}") self.assertEqual(receipt["deliveryStatus"], "pending") for invalid in ("relative", str(old), str(moved / "cwd-probe")): with self.assertRaises(urllib.error.HTTPError) as caught: app.api( f"/sessions/{session}/workspace", {"workspace": invalid}9 unmodified lines ], str(moved), ) self.assertEqual(json.loads((app.home / "daemon.json").read_text())["pid"], pid) self.send(app, session, "workspace probe after move") deadline = time.monotonic() + 35 while time.monotonic() < deadline: receipt = self.read(app, f"/operations/{operation}") if receipt["deliveryStatus"] == "committed": break time.sleep(0.05) self.assertEqual(receipt["deliveryStatus"], "committed") app.idle(session) self.assertEqual( Path((moved / "cwd-probe").read_text()).resolve(), moved.resolve() ) self.assertFalse(old.exists())473 unmodified linesExpand all123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339"""Admission and consumption failures must leave recoverable, atomic decisions."""
import base64import jsonimport randomimport structimport zlibimport sqlite3import timeimport unittestimport urllib.error
from harness import Albedo, Provider, exclusive, operation_id, textfrom image_limits_test import png
SEVEN_DAYS_MS = 7 * 24 * 60 * 60 * 1000
@exclusiveclass OperationLifecycleTests(unittest.TestCase): def setUp(self): self.provider = Provider(lambda request: text("done")) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
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 delivered(self, operation): deadline = time.monotonic() + 35 while time.monotonic() < deadline: receipt = self.receipt(operation) if receipt["deliveryStatus"] == "committed": return receipt time.sleep(0.05) self.fail(f"operation was not committed: {receipt}")
def test_creation_failure_rolls_back_session_and_receipt(self): operation = operation_id() request = { "operationId": operation, "workspace": str(self.app.workspace), "provider": self.app.profile, } with self.database() as db: before = db.execute("SELECT count(*) FROM sessions").fetchone()[0] db.execute( "CREATE TRIGGER fail_creation BEFORE INSERT ON sessions " "BEGIN SELECT RAISE(ABORT, 'fixture'); END" ) with self.assertRaises(urllib.error.HTTPError): self.post("/sessions", request) with self.database() as db: self.assertEqual( db.execute("SELECT count(*) FROM sessions").fetchone()[0], before ) self.assertEqual( db.execute( "SELECT count(*) FROM operations WHERE id=?", (operation,) ).fetchone()[0], 0, ) db.execute("DROP TRIGGER fail_creation") created = self.post("/sessions", request) self.assertEqual(self.post("/sessions", request), created) self.assertEqual(self.receipt(operation)["status"], "accepted")
def test_admission_failure_can_retry_original_id(self): session = self.app.session() operation = operation_id() request = {"operationId": operation, "content": "durable input"} with self.database() as db: db.execute( "CREATE TRIGGER fail_admission BEFORE INSERT ON pending_inputs " "BEGIN SELECT RAISE(ABORT, 'fixture'); END" ) with self.assertRaises(urllib.error.HTTPError): self.post(f"/sessions/{session}/events", request) with self.database() as db: self.assertEqual( db.execute( "SELECT count(*) FROM operations WHERE id=?", (operation,) ).fetchone()[0], 0, ) self.assertEqual( db.execute( "SELECT count(*) FROM transcript WHERE session=?", (session,) ).fetchone()[0], 0, ) db.execute("DROP TRIGGER fail_admission") accepted = self.post(f"/sessions/{session}/events", request) self.delivered(operation) self.assertEqual(self.post(f"/sessions/{session}/events", request), accepted) 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_consumption_failure_keeps_pending_input_until_storage_recovers(self): session = self.app.session() operation = operation_id() image = png(2, 2) request = { "operationId": operation, "content": "recover after storage repair", "image": { "mimeType": "image/png", "data": base64.b64encode(image).decode(), "width": 2, "height": 2, "bytes": len(image), }, } with self.database() as db: db.execute( "CREATE TRIGGER fail_consumption BEFORE INSERT ON submission_events " "BEGIN SELECT RAISE(ABORT, 'fixture'); END" ) accepted = self.post(f"/sessions/{session}/events", request) receipt = self.receipt(operation) self.assertEqual(receipt["status"], "accepted") self.assertEqual(receipt["deliveryStatus"], "pending") with self.database() as db: self.assertEqual( db.execute( "SELECT count(*) FROM pending_inputs WHERE operation_id=?", (operation,), ).fetchone()[0], 1, ) self.assertEqual( db.execute( "SELECT count(*) FROM transcript WHERE session=?", (session,) ).fetchone()[0], 0, ) self.assertEqual(db.execute("SELECT count(*) FROM images").fetchone()[0], 0) self.assertEqual( db.execute("SELECT count(*) FROM submission_events").fetchone()[0], 0 ) self.app.restart(crash=True) self.assertEqual(self.post(f"/sessions/{session}/events", request), accepted) self.assertEqual(self.receipt(operation)["deliveryStatus"], "pending") with self.database() as db: db.execute("DROP TRIGGER fail_consumption") self.delivered(operation) self.app.idle(session) self.app.restart(crash=True) self.assertEqual(self.receipt(operation)["deliveryStatus"], "committed") with self.assertRaises(urllib.error.HTTPError) as conflict: self.post( f"/sessions/{session}/events", {**request, "content": "changed after restart"}, ) self.assertEqual(conflict.exception.code, 409) self.assertEqual(json.load(conflict.exception)["code"], "operation_conflict") 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_repeated_image_commits_grow_only_compact_metadata(self): session = self.app.session() width, height = 256, 256
def chunk(kind, data): body = kind + data return ( struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body)) )
pixels = random.Random(1).randbytes(width * height * 3) rows = b"".join( b"\x00" + pixels[start : start + width * 3] for start in range(0, len(pixels), width * 3) ) image = ( b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)) + chunk(b"IDAT", zlib.compress(rows)) + chunk(b"IEND", b"") ) attachment = { "mimeType": "image/png", "data": base64.b64encode(image).decode(), "width": width, "height": height, "bytes": len(image), }
def durable_bytes(): with self.database() as db: return sum( db.execute( f"SELECT coalesce(sum(length({column})),0) FROM {table}" ).fetchone()[0] for table, column in ( ("images", "data"), ("transcript", "payload"), ("submission_events", "payload"), ("continuation_markers", "payload"), ("pending_inputs", "payload"), ("operations", "response"), ) )
operations = [] echoes = [] sizes = [] for index in range(3): operation = operation_id() echo = f"image-echo-{index}" operations.append(operation) echoes.append(echo) self.post( f"/sessions/{session}/events", { "operationId": operation, "clientId": echo, "content": "same image", "image": attachment, }, ) self.delivered(operation) self.app.idle(session) sizes.append(durable_bytes()) self.assertLess(sizes[2] - sizes[0], len(image)) self.app.restart(crash=True) users = [ event for event in self.app.events(session) if event.get("type") == "user" ] self.assertEqual([event["operationId"] for event in users], operations) self.assertEqual([event["clientId"] for event in users], echoes) self.assertEqual( [event["image"] for event in users], [{key: value for key, value in attachment.items() if key != "data"}] * 3, )
def test_expired_and_future_ids_cannot_create_work(self): now = int(time.time() * 1000) for timestamp, status, code in ( (now - SEVEN_DAYS_MS - 10000, 410, "operation_expired"), (now + 6 * 60 * 1000, 400, "operation_future"), ): operation = operation_id(timestamp) request = {"operationId": operation, "workspace": str(self.app.workspace)} with self.assertRaises(urllib.error.HTTPError) as failure: self.post("/sessions", request) self.assertEqual(failure.exception.code, status) self.assertEqual(json.load(failure.exception)["code"], code) with self.database() as db: self.assertEqual( db.execute( "SELECT count(*) FROM operations WHERE id=?", (operation,) ).fetchone()[0], 0, )
def test_rejected_creation_replays_after_workspace_is_repaired(self): operation = operation_id() workspace = self.app.workspace / "missing" request = { "operationId": operation, "workspace": str(workspace), "provider": self.app.profile, } with self.assertRaises(urllib.error.HTTPError) as original: self.post("/sessions", request) error = json.load(original.exception) self.assertEqual(self.receipt(operation)["status"], "rejected") workspace.mkdir() self.app.restart(crash=True) with self.assertRaises(urllib.error.HTTPError) as repeated: self.post("/sessions", request) self.assertEqual(repeated.exception.code, original.exception.code) self.assertEqual(json.load(repeated.exception), error) with self.database() as db: self.assertEqual( db.execute("SELECT count(*) FROM sessions").fetchone()[0], 0 )
def test_failed_interrupt_does_not_discard_pending_input(self): session = self.app.session() operation = operation_id() request = { "operationId": operation, "content": "keep until cancellation commits", } with self.database() as db: db.execute( "CREATE TRIGGER hold_consumption BEFORE INSERT ON transcript " "WHEN NEW.row_class='user' BEGIN SELECT RAISE(ABORT, 'fixture'); END" ) self.post(f"/sessions/{session}/events", request) with self.database() as db: db.execute( "CREATE TRIGGER fail_cancellation BEFORE UPDATE ON operations " "WHEN NEW.delivery='cancelled' BEGIN SELECT RAISE(ABORT, 'fixture'); END" ) result = self.post(f"/sessions/{session}/interrupt", {}) self.assertFalse(result["interrupted"]) self.assertEqual(self.receipt(operation)["deliveryStatus"], "pending") with self.database() as db: db.execute("DROP TRIGGER fail_cancellation") result = self.post(f"/sessions/{session}/interrupt", {}) self.assertTrue(result["interrupted"]) self.app.restart(crash=True) self.assertEqual(self.receipt(operation)["deliveryStatus"], "cancelled") with self.database() as db: self.assertEqual( db.execute( "SELECT count(*) FROM pending_inputs WHERE operation_id=?", (operation,), ).fetchone()[0], 0, )123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227"""Receipts must survive deletion and preserve aged pending work during pruning."""
import jsonimport sqlite3import timeimport unittestimport urllib.error
from harness import Albedo, Provider, exclusive, operation_id, text
RETENTION_MS = 7 * 24 * 60 * 60 * 1000
@exclusiveclass 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, )7 unmodified lines891011121213141516138 unmodified linesExpand all7 unmodified lines "submission": { "status": 202, "body": { "ok": true, "queued": false "queued": false, "operationId": "01900000-0000-7000-8000-000000000001" } }, "interruption": { "status": 200,138 unmodified linesExpand all1234567891011121314151617-module(albedo_operations).-export([validate_id/2, decode_display/1]).
validate_id(Id, Now) -> case re:run(Id, <<"^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$">>, [{capture, none}]) of match -> <<First:8/binary, "-", Second:4/binary, _/binary>> = Id, Timestamp = binary_to_integer(<<First/binary, Second/binary>>, 16), if Timestamp > Now + 300000 -> {error, <<"operation_future">>}; Timestamp < Now - 604800000 -> {error, <<"operation_expired">>}; true -> {ok, nil} end; _ -> {error, <<"operation_invalid">>} end.
decode_display(Payload) -> binary_to_term(Payload, [safe]).