From 1d9291539c9b761a97f8213dd794a0d6d5a0c7ce Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 06:36:49 -0400 Subject: [PATCH 1/2] feat(receiver): bound asynchronous signed delivery until receipt --- async_scheduled_sink.py | 95 ++++++++++++++ docs/demos/async-scheduled.log | 13 ++ docs/demos/async-scheduled.timing | 6 + docs/scheduled-receiver.md | 39 ++++++ pyproject.toml | 2 +- scripts/demo-async-scheduled.py | 63 ++++++++++ test_async_scheduled_sink.py | 198 ++++++++++++++++++++++++++++++ 7 files changed, 415 insertions(+), 1 deletion(-) create mode 100644 async_scheduled_sink.py create mode 100644 docs/demos/async-scheduled.log create mode 100644 docs/demos/async-scheduled.timing create mode 100644 scripts/demo-async-scheduled.py create mode 100644 test_async_scheduled_sink.py diff --git a/async_scheduled_sink.py b/async_scheduled_sink.py new file mode 100644 index 0000000..3f0d52d --- /dev/null +++ b/async_scheduled_sink.py @@ -0,0 +1,95 @@ +"""Bounded background delivery, with receipt only after signed publication. + +An operator supplies one immutable scope and trusted callbacks. Incoming data +may select only its UTC slot. Threads remain in the owning service's cgroup; +this adapter cannot keep a child alive after that service is stopped. +""" +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from contextvars import copy_context +from dataclasses import asdict +import json +import threading + +from scheduled_receiver import ExecutionScope, receive, run_id +from slot_receiver import SlotLedger + + +class DeliveryPending(ConnectionError): + """No terminal receipt yet; leave the durable emitter row undelivered.""" + + +class AsyncScheduledSink: + """A sink for complete admitted bindings, not generic emitter envelopes. + + deliver() never waits for a child: it raises DeliveryPending until receive() + has persisted and published the outcome. A failed worker raises its actual + error on the next delivery; another delivery then re-enters the durable + receiver, which permits publication retries but never child replay. + + Capacity counts queued, running AND completed-but-unacknowledged futures. + A thread pool's unbounded queue therefore never becomes an unbounded job + queue. The owner must call close() during orderly shutdown. Abrupt death + leaves claimed slots UNKNOWN; reconcile them, never erase the ledger. + """ + + def __init__(self, ledger: SlotLedger, scope: ExecutionScope, port_factory, + execute, *, workers: int = 1, capacity: int = 4): + scope.validate() + if (type(workers) is not int or type(capacity) is not int or + not 1 <= workers <= capacity <= 64): + raise ValueError("require 1 <= workers <= capacity <= 64") + if not callable(port_factory) or not callable(execute): + raise ValueError("trusted port factory and executor required") + self.ledger, self.scope = ledger, scope + self.port_factory, self.execute = port_factory, execute + self.capacity = capacity + self._lock = threading.Lock() + self._pending = {} + self._closed = False + self._pool = ThreadPoolExecutor(max_workers=workers, + thread_name_prefix="scheduled-receiver") + + def delivery_scope(self): + return {"receiver": "async-scheduled-v1", **asdict(self.scope)} + + def deliver(self, delivery: dict) -> None: + if type(delivery) is not dict: + raise ValueError("complete admitted binding required") + slot = delivery.get("scheduled_at") + identity = run_id(self.scope, slot) + binding = {**asdict(self.scope), "scheduled_at": slot} + if (json.dumps(delivery, sort_keys=True, allow_nan=False) != + json.dumps(binding, sort_keys=True, allow_nan=False)): + raise ValueError("delivery differs from pinned scope") + with self._lock: + if self._closed: + raise ConnectionError("scheduled receiver is closed") + future = self._pending.get(identity) + if future is None: + if len(self._pending) >= self.capacity: + raise DeliveryPending("scheduled receiver capacity reached") + context = copy_context() + future = self._pool.submit(context.run, self._receive, slot, binding) + self._pending[identity] = future + # Submission is not a durable receipt, even if a fast worker + # happens to finish before this call exits. + raise DeliveryPending("scheduled delivery submitted; receipt pending") + if not future.done(): + raise DeliveryPending("scheduled delivery still pending") + try: + future.result() # terminal publication verified, or real error + finally: + del self._pending[identity] + + def _receive(self, slot, binding): + port = self.port_factory(slot) + return receive(self.ledger, self.scope, slot, binding, port, + lambda: self.execute(slot)) + + def close(self): + """Stop admission and drain workers; call outside the reactor tick.""" + with self._lock: + self._closed = True + self._pool.shutdown(wait=True) diff --git a/docs/demos/async-scheduled.log b/docs/demos/async-scheduled.log new file mode 100644 index 0000000..83ba297 --- /dev/null +++ b/docs/demos/async-scheduled.log @@ -0,0 +1,13 @@ +Script started on 2026-10-15 06:37:02-04:00 [COMMAND="stty -onlcr && .venv/bin/python scripts/demo-async-scheduled.py" ] +Source demo: real worker thread and SQLite; signing callbacks are fixtures. +Existing receiver positive control: True +Async handoff available: True +First delivery: PENDING, no receipt +Duplicate delivery: PENDING, no receipt +Worker blocked while caller continued: True +Execution callbacks: 1 ; publications: 0 +Later delivery: RECEIPT after terminal publication +Published actual outcome: skipped-by-hold +Final execution callbacks: 1 + +Script done on 2026-10-15 06:37:02-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/async-scheduled.timing b/docs/demos/async-scheduled.timing new file mode 100644 index 0000000..7e77af2 --- /dev/null +++ b/docs/demos/async-scheduled.timing @@ -0,0 +1,6 @@ +0.035160 117 +0.000015 30 +0.020770 36 +0.009772 125 +0.010488 93 +0.000004 29 diff --git a/docs/scheduled-receiver.md b/docs/scheduled-receiver.md index 20e1dda..0bed821 100644 --- a/docs/scheduled-receiver.md +++ b/docs/scheduled-receiver.md @@ -1,5 +1,44 @@ # Signed scheduled delivery bridge +## Bounded asynchronous handoff + +`async_scheduled_sink.AsyncScheduledSink` runs the existing signed receiver +on a bounded thread pool. The operator supplies one immutable `ExecutionScope`, +the persistent ledger, a trusted `port_factory(scheduled_at)`, and a trusted +`execute(scheduled_at)` callback. Delivery data must repeat the exact scope and +integer UTC slot; it cannot supply commands or choose a different identity. +The callback must enforce the admitted command's runtime and output limits. + +`deliver(binding)` raises `DeliveryPending` after submission and while work is +running. It returns only on a later call after the receiver verifies terminal +publication. Existing emitter delivery therefore keeps its durable outbox row +undelivered and retries with its normal backoff. Pending counts as a delivery +failure in existing metrics; it does not mean the child failed. A terminal hold +or failed child can have a successful delivery receipt for its actual outcome. +The emitter's retry cadence bounds receipt latency independently of completion. + +Capacity includes queued, running, and completed work awaiting receipt. Duplicate +in-flight deliveries reuse one future. Worker exceptions reach the next delivery +unchanged; later retries use the persistent slot ledger, so publication may retry +but a claimed child cannot run again. Context variables are copied on submission, +including any strict-admission context set by the caller. + +The owner calls `close()` outside the reactor tick to stop admission and drain +workers. This can wait for the full callback duration. Abrupt termination leaves +claims UNKNOWN. Threads and children remain inside the owning service's cgroup; +this adapter cannot provide survival across that service's exit. + +This source adapter accepts complete bindings. A generic emitter envelope does +not contain that binding: production configuration still needs a separately +validated schedule/envelope adapter and trusted fixed-command executor. No sink +factory or production configuration enables this class automatically. Do not +acknowledge an upstream delivery merely because a future was submitted. + +Reproduce the blocked-child/receipt/capacity controls with +`python -m pytest -q test_async_scheduled_sink.py`. These use real threads, +SQLite claims and the actual emitter outbox, with synthetic signing callbacks. +They do not establish installed signing, schedule coverage or natural firings. + `scheduled_receiver.receive` connects an already admitted delivery to the persistent `SlotLedger` and a trusted `SignedRunPort`. It is a library entry point: importing it starts no listener, timer, scheduler or child process. diff --git a/pyproject.toml b/pyproject.toml index c114451..3f76081 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,7 @@ chaski-once = "once:main" [tool.setuptools] py-modules = [ - "alertmanager_sink", "change_stream", "chaski", "command_sink", + "alertmanager_sink", "async_scheduled_sink", "change_stream", "chaski", "command_sink", "consumer_lease", "demo", "emitter", "external_event", "failure_family_adapter", "graph_sink", "incremental", "once", "scheduled_receiver", "shuttle_process_port", "slot_receiver", ] diff --git a/scripts/demo-async-scheduled.py b/scripts/demo-async-scheduled.py new file mode 100644 index 0000000..be84de5 --- /dev/null +++ b/scripts/demo-async-scheduled.py @@ -0,0 +1,63 @@ +"""Real threads and ledger; synthetic signing callbacks and no external I/O.""" +from dataclasses import asdict +import importlib.util +from pathlib import Path +import sys +import tempfile +import threading + +source = (Path(sys.argv[2]).resolve() if len(sys.argv) == 3 and sys.argv[1] == '--source' + else Path(__file__).resolve().parents[1]) +sys.path.insert(0, str(source)) +from scheduled_receiver import ExecutionScope, receive # noqa: E402 +from slot_receiver import SlotLedger # noqa: E402 + +print('Source demo: real worker thread and SQLite; signing callbacks are fixtures.') +print('Existing receiver positive control:', callable(receive)) +present = importlib.util.find_spec('async_scheduled_sink') is not None +print('Async handoff available:', present) +if not present: + sys.exit(0) +from async_scheduled_sink import AsyncScheduledSink, DeliveryPending # noqa: E402 + +with tempfile.TemporaryDirectory() as root: + entered, release = threading.Event(), threading.Event() + calls, publications = [], [] + class Port: + def admit(self, scope, slot): pass + def started(self, run, scope): pass + def finished(self, run, scope, outcome): pass + def publish(self, run, scope, outcome): publications.append(outcome) + def execute(slot): + calls.append(slot) + entered.set() + assert release.wait(5) + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'operator', 'urn:example:workflow') + delivery = {**asdict(scope), 'scheduled_at': 600} + sink = AsyncScheduledSink(SlotLedger(Path(root)/'slots.sqlite'), scope, + lambda slot: Port(), execute, capacity=1) + try: + for label in ('First delivery', 'Duplicate delivery'): + try: + sink.deliver(delivery) + except DeliveryPending: + print(label + ': PENDING, no receipt') + else: + raise AssertionError('acknowledged blocked child') + assert entered.wait(5) + print('Worker blocked while caller continued:', not release.is_set()) + print('Execution callbacks:', len(calls), '; publications:', len(publications)) + release.set() + # Demo waits for completion explicitly; deliver never waits on workers. + with sink._lock: + future = next(iter(sink._pending.values())) + future.result(timeout=5) + sink.deliver(delivery) + print('Later delivery: RECEIPT after terminal publication') + print('Published actual outcome:', publications[0]['outcome']) + print('Final execution callbacks:', len(calls)) + finally: + release.set() + sink.close() diff --git a/test_async_scheduled_sink.py b/test_async_scheduled_sink.py new file mode 100644 index 0000000..cc53a1a --- /dev/null +++ b/test_async_scheduled_sink.py @@ -0,0 +1,198 @@ +"""Real threads, durable claims and emitter receipts; no production execution.""" +from contextvars import ContextVar +from dataclasses import asdict +import json +import threading + +import pytest + +from async_scheduled_sink import AsyncScheduledSink, DeliveryPending +from emitter import connect, deliver_pending +from scheduled_receiver import ExecutionScope +from slot_receiver import SlotLedger, SlotUnknown + + +@pytest.fixture +def rig(tmp_path): + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'operator', 'urn:example:workflow') + entered, release, published = (threading.Event() for _ in range(3)) + calls, phases = [], [] + fail_publish = [False] + + class Port: + def admit(self, scope, slot): + phases.append('admit') + + def started(self, run, scope): + phases.append('started') + + def finished(self, run, scope, outcome): + phases.append('finished') + + def publish(self, run, scope, outcome): + phases.append('publish') + if fail_publish[0]: + raise ConnectionError('publication unavailable') + published.set() + + def execute(slot): + calls.append(slot) + entered.set() + assert release.wait(5), 'test did not release blocked worker' + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + + ledger = SlotLedger(tmp_path/'slots.sqlite') + sink = AsyncScheduledSink(ledger, scope, lambda slot: Port(), execute, + workers=1, capacity=1) + yield sink, scope, entered, release, published, calls, phases, fail_publish, ledger + release.set() + sink.close() + + +def binding(scope, slot=600): + return {**asdict(scope), 'scheduled_at': slot} + + +def drain_worker(sink): + # Wait for actual worker completion without timing guesses; only the test + # waits, never deliver() or the reactor. + with sink._lock: + futures = list(sink._pending.values()) + for future in futures: + try: + future.result(timeout=5) + except ConnectionError: + pass + + +def test_emitter_stays_undelivered_while_child_blocks_then_receipts(rig, tmp_path): + sink, scope, entered, release, published, calls, phases, _, _ = rig + conn = connect(tmp_path/'emitter.sqlite') + try: + conn.execute('INSERT INTO outbox(event_id,emitter,item,generation,payload,created) ' + 'VALUES(?,?,?,?,?,?)', + ('event1', 'schedule', scope.job, 1, json.dumps(binding(scope)), 0)) + assert deliver_pending(conn, sink, 0, 'schedule') == (0, 1) + assert entered.wait(5) + assert conn.execute('SELECT delivered_at FROM outbox').fetchone() == (None,) + # A real running child remains blocked, yet delivery can be called + # again immediately and cannot launch another child. + assert deliver_pending(conn, sink, 30, 'schedule') == (0, 1) + assert calls == [600] and not published.is_set() + release.set() + drain_worker(sink) + assert published.is_set() + assert deliver_pending(conn, sink, 90, 'schedule') == (1, 0) + assert conn.execute('SELECT delivered_at FROM outbox').fetchone() == (90,) + assert phases == ['admit', 'started', 'finished', 'publish'] + finally: + conn.close() + + +def test_capacity_includes_finished_unacknowledged_work_and_reuses_slot(rig): + sink, scope, entered, release, _, calls, _, _, _ = rig + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + assert entered.wait(5) + with pytest.raises(DeliveryPending, match='capacity'): + sink.deliver(binding(scope, 1200)) + release.set() + drain_worker(sink) + with pytest.raises(DeliveryPending, match='capacity'): + sink.deliver(binding(scope, 1200)) + sink.deliver(binding(scope)) + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope, 1200)) + drain_worker(sink) + sink.deliver(binding(scope, 1200)) + assert calls == [600, 1200] + + +def test_publication_retry_and_receiver_recreation_never_repeat_child(rig): + sink, scope, entered, release, _, calls, _, fail_publish, ledger = rig + fail_publish[0] = True + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + assert entered.wait(5) + release.set() + drain_worker(sink) + with pytest.raises(ConnectionError, match='publication unavailable'): + sink.deliver(binding(scope)) + fail_publish[0] = False + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + drain_worker(sink) + sink.deliver(binding(scope)) + sink.close() + replacement = AsyncScheduledSink(ledger, scope, sink.port_factory, + lambda slot: pytest.fail('replayed child')) + try: + with pytest.raises(DeliveryPending): + replacement.deliver(binding(scope)) + drain_worker(replacement) + replacement.deliver(binding(scope)) + assert calls == [600] + finally: + replacement.close() + + +def test_claimed_unknown_slot_stays_unknown_without_child(rig): + sink, scope, _, _, _, calls, _, _, ledger = rig + from slot_receiver import slot_id + with pytest.raises(RuntimeError): + ledger.run(slot_id(scope.job, scope.revision, 600), binding(scope), + lambda: (_ for _ in ()).throw(RuntimeError('crashed after claim'))) + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + with pytest.raises(SlotUnknown): + drain_worker(sink) + with pytest.raises(SlotUnknown): + sink.deliver(binding(scope)) + assert calls == [] + + +@pytest.mark.parametrize('change', [{'host': 'urn:example:other'}, + {'scheduled_at': True}, {'command': 'anything'}, + {'scheduled_at': -1}]) +def test_bad_binding_refuses_before_admission(rig, change): + sink, scope, _, _, _, calls, phases, _, _ = rig + with pytest.raises(ValueError): + sink.deliver({**binding(scope), **change}) + assert not sink._pending and calls == [] and phases == [] + + +def test_context_is_copied_at_submit_and_restored_in_caller(rig): + sink, scope, _, _, _, _, _, _, _ = rig + flag = ContextVar('strict_admission', default=False) + observed = [] + def execute(slot): + observed.append(flag.get()) + flag.set(False) + return {'outcome': 'succeeded', 'exit_code': 0} + sink.execute = execute + token = flag.set(True) + try: + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + drain_worker(sink) + sink.deliver(binding(scope)) + assert observed == [True] and flag.get() is True + finally: + flag.reset(token) + + +def test_close_refuses_new_work(rig): + sink, scope, _, _, _, calls, phases, _, _ = rig + sink.close() + with pytest.raises(ConnectionError, match='closed'): + sink.deliver(binding(scope)) + assert calls == [] and phases == [] + + +@pytest.mark.parametrize('workers,capacity', [(0, 1), (True, 1), (2, 1), (1, 65)]) +def test_unbounded_or_invalid_capacity_refuses(rig, workers, capacity): + sink, scope, _, _, _, _, _, _, ledger = rig + with pytest.raises(ValueError): + AsyncScheduledSink(ledger, scope, sink.port_factory, sink.execute, + workers=workers, capacity=capacity) From 0b4a0eb56800020bc21846927d1c21514b179493 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 07:20:14 -0400 Subject: [PATCH 2/2] feat(receiver): verify native scheduled emitter envelopes --- docs/demos/scheduled-event.log | 12 +++ docs/demos/scheduled-event.timing | 6 ++ docs/scheduled-receiver.md | 43 +++++++- pyproject.toml | 2 +- scheduled_event_sink.py | 75 ++++++++++++++ scripts/demo-scheduled-event.py | 75 ++++++++++++++ test_scheduled_event_sink.py | 162 ++++++++++++++++++++++++++++++ 7 files changed, 369 insertions(+), 6 deletions(-) create mode 100644 docs/demos/scheduled-event.log create mode 100644 docs/demos/scheduled-event.timing create mode 100644 scheduled_event_sink.py create mode 100644 scripts/demo-scheduled-event.py create mode 100644 test_scheduled_event_sink.py diff --git a/docs/demos/scheduled-event.log b/docs/demos/scheduled-event.log new file mode 100644 index 0000000..c650d33 --- /dev/null +++ b/docs/demos/scheduled-event.log @@ -0,0 +1,12 @@ +Script started on 2026-10-15 07:19:09-04:00 [COMMAND="stty -onlcr && .venv/bin/python scripts/demo-scheduled-event.py" ] +Source demo: actual emitter/outbox, worker and ledger; signing is a fixture. +Existing async receiver positive control: True +Native scheduled envelope adapter available: True +Observed due declaration; durable outbox rows: 1 +Two delivery attempts; child still blocked: True +Execution callbacks: 1 ; published: 0 ; receipt: None +Published actual outcome: skipped-by-hold +Terminal receipt: 690.0 ; execution callbacks: 1 +Outbox pinned emitter and recipient: schedule operator + +Script done on 2026-10-15 07:19:09-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/scheduled-event.timing b/docs/demos/scheduled-event.timing new file mode 100644 index 0000000..d7f590f --- /dev/null +++ b/docs/demos/scheduled-event.timing @@ -0,0 +1,6 @@ +0.054368 124 +0.000013 50 +0.063978 49 +0.022699 103 +0.016719 91 +0.000049 55 diff --git a/docs/scheduled-receiver.md b/docs/scheduled-receiver.md index 0bed821..b5fa094 100644 --- a/docs/scheduled-receiver.md +++ b/docs/scheduled-receiver.md @@ -28,17 +28,50 @@ workers. This can wait for the full callback duration. Abrupt termination leaves claims UNKNOWN. Threads and children remain inside the owning service's cgroup; this adapter cannot provide survival across that service's exit. -This source adapter accepts complete bindings. A generic emitter envelope does -not contain that binding: production configuration still needs a separately -validated schedule/envelope adapter and trusted fixed-command executor. No sink -factory or production configuration enables this class automatically. Do not -acknowledge an upstream delivery merely because a future was submitted. +The asynchronous receiver accepts complete bindings. `ScheduledEventSink` below +converts native emitter envelopes into that binding. Production configuration +still needs a validated catalogue projection and trusted fixed-command executor. +No sink factory or production configuration enables either class automatically. +Do not acknowledge an upstream delivery merely because a future was submitted. Reproduce the blocked-child/receipt/capacity controls with `python -m pytest -q test_async_scheduled_sink.py`. These use real threads, SQLite claims and the actual emitter outbox, with synthetic signing callbacks. They do not establish installed signing, schedule coverage or natural firings. +### Native emitter envelopes + +Wrap the asynchronous receiver with +`scheduled_event_sink.ScheduledEventSink(receiver, emitter, recipient)`. The +operator pins the emitter label, human routing recipient and receiver scope. +The adapter accepts the existing emitter's `scheduled` envelope only when its +item and routing match those pins. It decodes the slot from the canonical +`scheduled-v1-` event identifier and re-encodes the complete expected scope to +verify exact equality. Alternate encodings, foreign scopes, duplicate or extra +JSON fields, missing fields, boolean slots and unbounded timestamps refuse +before the asynchronous receiver is called. Delivery timestamps and emitter +generation cannot replace the frozen scheduled slot. + +The adapter exposes a stable delivery scope containing the emitter, recipient +and inner receiver scope. Existing outbox delivery freezes this scope before +its first call, preserving the original binding across retries. A scheduling +adapter must emit `run_id(scope, scheduled_at)` as its deterministic `event_id` +and use `event: scheduled`; it must not set `generation_event_ids`, which +replaces that identifier with a hash. Set `baseline_emits` when an already-due +slot should emit during bootstrap. These options require the reviewed schedule +projection; this envelope adapter cannot infer a schedule or admit execution. + +`test_scheduled_event_sink.py` uses actual `observe` and `deliver_pending` with +the durable outbox, real worker threads and slot ledger. Signing callbacks are +synthetic. It verifies one blocked execution, no early receipt and terminal +publication before acknowledgment, alongside rejected scopes and malformed +events. All execution admission remains the trusted signed port's job. + +Replay `python scripts/demo-scheduled-event.py`, or download the output-only +[terminal recording](demos/scheduled-event.log) and [timing](demos/scheduled-event.timing) +for `scriptreplay`. The demo prints actual outbox and receipt state with +synthetic signing callbacks; it does not contact a service or run a real job. + `scheduled_receiver.receive` connects an already admitted delivery to the persistent `SlotLedger` and a trusted `SignedRunPort`. It is a library entry point: importing it starts no listener, timer, scheduler or child process. diff --git a/pyproject.toml b/pyproject.toml index 3f76081..6de0bc1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ chaski-once = "once:main" py-modules = [ "alertmanager_sink", "async_scheduled_sink", "change_stream", "chaski", "command_sink", "consumer_lease", "demo", "emitter", "external_event", "failure_family_adapter", - "graph_sink", "incremental", "once", "scheduled_receiver", "shuttle_process_port", "slot_receiver", + "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_receiver", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scheduled_event_sink.py b/scheduled_event_sink.py new file mode 100644 index 0000000..7dc22b6 --- /dev/null +++ b/scheduled_event_sink.py @@ -0,0 +1,75 @@ +"""Translate a pinned emitter envelope into the existing scheduled binding. + +Only the canonical event identifier carries the frozen scope and UTC slot. +Delivery time, generation and evidence can never choose a command or slot. +""" +from __future__ import annotations + +import base64 +import json +import math +import re + +from scheduled_receiver import run_id + + +class ScheduledEventSink: + def __init__(self, receiver, emitter: str, recipient: str): + if (type(emitter) is not str or + not re.fullmatch(r'[A-Za-z][A-Za-z0-9_.-]{0,127}', emitter) or + type(recipient) is not str or not 0 < len(recipient) <= 256): + raise ValueError('pinned emitter and recipient required') + receiver.scope.validate() + self.receiver, self.scope = receiver, receiver.scope + self.emitter, self.recipient = emitter, recipient + + def delivery_scope(self): + return {'receiver': 'scheduled-event-v1', 'emitter': self.emitter, + 'recipient': self.recipient, 'binding': self.receiver.delivery_scope()} + + def deliver(self, event: dict) -> None: + required = {'event_id', 'event', 'emitter', 'item', 'generation', + 'recipient', 'evidence', 'observed_at'} + if (type(event) is not dict or not required <= event.keys() or + event.keys() - required - {'work_kind'}): + raise ValueError('complete supported emitter envelope required') + if (event['event'] != 'scheduled' or event['emitter'] != self.emitter or + event['item'] != self.scope.job or event['recipient'] != self.recipient): + raise ValueError('event differs from pinned receiver routing') + generation, observed = event['generation'], event['observed_at'] + if (type(generation) is not int or not 0 <= generation < 2**63 or + type(observed) not in (int, float) or + not 0 <= observed <= 253402300799 or not math.isfinite(observed)): + raise ValueError('invalid emitter metadata') + evidence = event['evidence'] + if evidence is not None and (type(evidence) is not str or len(evidence) > 8192): + raise ValueError('bounded textual evidence required') + if 'work_kind' in event and (type(event['work_kind']) is not str or + not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,127}', event['work_kind'])): + raise ValueError('invalid work kind') + identity = event['event_id'] + if (type(identity) is not str or len(identity) > 24000 or + not identity.startswith('scheduled-v1-')): + raise ValueError('bounded canonical scheduled identifier required') + try: + encoded = identity[len('scheduled-v1-'):] + raw = base64.b64decode(encoded + '=' * (-len(encoded) % 4), + altchars=b'-_', validate=True) + binding = json.loads(raw) + if type(binding) is not dict: + raise ValueError('scheduled binding must be an object') + slot = binding.get('scheduled_at') + if type(slot) is not int or not 0 <= slot <= 253402300799: + raise ValueError('bounded integer UTC slot required') + # Exact re-encoding refuses a foreign scope, duplicate JSON keys, + # extra fields, booleans in integer positions, padding and alternate + # JSON spellings. The observed delivery timestamp is not an anchor. + if run_id(self.scope, slot) != identity: + raise ValueError('scheduled identifier differs from pinned scope') + except (ValueError, UnicodeError, RecursionError) as exc: + raise ValueError('invalid canonical scheduled identifier') from exc + del binding['version'] + self.receiver.deliver(binding) + + def close(self): + self.receiver.close() diff --git a/scripts/demo-scheduled-event.py b/scripts/demo-scheduled-event.py new file mode 100644 index 0000000..064fed0 --- /dev/null +++ b/scripts/demo-scheduled-event.py @@ -0,0 +1,75 @@ +"""Actual emitter/outbox, thread and ledger; synthetic signing, no network.""" +import importlib.util +import json +import logging +from pathlib import Path +import sys +import tempfile +import threading + +source = (Path(sys.argv[2]).resolve() if len(sys.argv) == 3 and sys.argv[1] == '--source' + else Path(__file__).resolve().parents[1]) +sys.path.insert(0, str(source)) +from async_scheduled_sink import AsyncScheduledSink # noqa: E402 +from emitter import Emitter, connect, deliver_pending, observe # noqa: E402 +from scheduled_receiver import ExecutionScope, run_id # noqa: E402 +from slot_receiver import SlotLedger # noqa: E402 + +# Pending delivery is expected here. Show the measured outbox/receipt state +# below instead of logging the entire reversible fixture identifier twice. +logging.getLogger('chaski.emitter').setLevel(logging.ERROR) + +print('Source demo: actual emitter/outbox, worker and ledger; signing is a fixture.') +print('Existing async receiver positive control:', callable(AsyncScheduledSink.deliver)) +present = importlib.util.find_spec('scheduled_event_sink') is not None +print('Native scheduled envelope adapter available:', present) +if not present: + sys.exit(0) +from scheduled_event_sink import ScheduledEventSink # noqa: E402 + +with tempfile.TemporaryDirectory() as root: + root = Path(root) + entered, release = threading.Event(), threading.Event() + calls, publications = [], [] + class Port: + def admit(self, scope, slot): pass + def started(self, run, scope): pass + def finished(self, run, scope, outcome): pass + def publish(self, run, scope, outcome): publications.append(outcome) + def execute(slot): + calls.append(slot) + entered.set() + assert release.wait(5) + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + definition = Emitter('schedule', ['unused'], 60, 'operator', + event='scheduled', baseline_emits=True) + conn = connect(root/'emitter.sqlite') + inner = AsyncScheduledSink(SlotLedger(root/'slots.sqlite'), scope, + lambda slot: Port(), execute) + sink = ScheduledEventSink(inner, 'schedule', 'operator') + try: + queued = observe(conn, definition, [{'item': scope.job, 'verdict': 'UNBLOCKED', + 'event_id': run_id(scope, 600), 'owner': 'operator'}], 600) + print('Observed due declaration; durable outbox rows:', len(queued)) + assert deliver_pending(conn, sink, 600, 'schedule') == (0, 1) + assert entered.wait(5) + assert deliver_pending(conn, sink, 630, 'schedule') == (0, 1) + receipt = conn.execute('SELECT delivered_at FROM outbox').fetchone()[0] + print('Two delivery attempts; child still blocked:', not release.is_set()) + print('Execution callbacks:', len(calls), '; published:', len(publications), '; receipt:', receipt) + release.set() + with inner._lock: + future = next(iter(inner._pending.values())) + future.result(timeout=5) + assert deliver_pending(conn, sink, 690, 'schedule') == (1, 0) + receipt = conn.execute('SELECT delivered_at FROM outbox').fetchone()[0] + print('Published actual outcome:', publications[0]['outcome']) + print('Terminal receipt:', receipt, '; execution callbacks:', len(calls)) + frozen = json.loads(conn.execute('SELECT delivery_scope FROM outbox').fetchone()[0]) + print('Outbox pinned emitter and recipient:', frozen['emitter'], frozen['recipient']) + finally: + release.set() + sink.close() + conn.close() diff --git a/test_scheduled_event_sink.py b/test_scheduled_event_sink.py new file mode 100644 index 0000000..ec2b4c7 --- /dev/null +++ b/test_scheduled_event_sink.py @@ -0,0 +1,162 @@ +"""Actual observe/outbox delivery into the async receiver; signing is a fixture.""" +import base64 +from dataclasses import asdict, replace +import json +import threading + +import pytest + +from async_scheduled_sink import AsyncScheduledSink +from emitter import Emitter, connect, deliver_pending, observe +from scheduled_event_sink import ScheduledEventSink +from scheduled_receiver import ExecutionScope, run_id +from slot_receiver import SlotLedger + + +@pytest.fixture +def emitted(tmp_path): + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + definition = Emitter('schedule', ['unused'], 60, 'operator', + event='scheduled', baseline_emits=True) + conn = connect(tmp_path/'emitter.sqlite') + assert observe(conn, definition, [{'item': scope.job, 'verdict': 'UNBLOCKED', + 'event_id': run_id(scope, 600), 'owner': 'operator'}], 0) + event = json.loads(conn.execute('SELECT payload FROM outbox').fetchone()[0]) + yield scope, conn, event + conn.close() + + +class RecordingReceiver: + def __init__(self, scope): + self.scope, self.calls = scope, [] + + def delivery_scope(self): + return {'scope': asdict(self.scope)} + + def deliver(self, binding): + self.calls.append(binding) + + +def encode(value): + raw = json.dumps(value, sort_keys=True, separators=(',', ':')).encode() + return 'scheduled-v1-' + base64.urlsafe_b64encode(raw).decode().rstrip('=') + + +def test_real_emitter_outbox_waits_for_signed_receiver_and_pins_scope(emitted, tmp_path): + scope, conn, event = emitted + entered, release, published = (threading.Event() for _ in range(3)) + calls = [] + + class Port: + def admit(self, scope, slot): pass + def started(self, run, scope): pass + def finished(self, run, scope, outcome): pass + def publish(self, run, scope, outcome): published.set() + + def execute(slot): + calls.append(slot) + entered.set() + assert release.wait(5) + return {'outcome': 'succeeded', 'exit_code': 0} + + inner = AsyncScheduledSink(SlotLedger(tmp_path/'slots.sqlite'), scope, + lambda slot: Port(), execute) + sink = ScheduledEventSink(inner, 'schedule', 'operator') + try: + assert deliver_pending(conn, sink, 0, 'schedule') == (0, 1) + assert entered.wait(5) + frozen, receipt = conn.execute('SELECT delivery_scope,delivered_at FROM outbox').fetchone() + assert json.loads(frozen) == sink.delivery_scope() and receipt is None + assert deliver_pending(conn, sink, 30, 'schedule') == (0, 1) + assert calls == [600] and not published.is_set() + release.set() + with inner._lock: + future = next(iter(inner._pending.values())) + future.result(timeout=5) + assert deliver_pending(conn, sink, 90, 'schedule') == (1, 0) + assert published.is_set() and calls == [600] + assert conn.execute('SELECT delivered_at FROM outbox').fetchone() == (90,) + assert event['event_id'] == run_id(scope, 600) + finally: + release.set() + sink.close() + + +def test_delivery_time_and_generation_cannot_change_frozen_slot(emitted): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + sink = ScheduledEventSink(receiver, 'schedule', 'operator') + event.update(observed_at=9876543210, generation=123) + sink.deliver(event) + assert receiver.calls == [{**asdict(scope), 'scheduled_at': 600}] + + +@pytest.mark.parametrize('field,value', [('event', 'unblocked'), ('emitter', 'other'), + ('item', 'urn:example:foreign'), ('recipient', 'other'), ('command', 'anything'), + ('generation', True), ('generation', -1), ('observed_at', True), + ('observed_at', float('nan')), ('observed_at', float('inf')), + ('evidence', 'x'*8193), ('work_kind', []), ('event_id', 'scheduled-v1-!'), + ('event_id', 'scheduled-v1-'+'a'*24000)]) +def test_misdirected_or_malformed_envelope_never_reaches_receiver(emitted, field, value): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + sink = ScheduledEventSink(receiver, 'schedule', 'operator') + event[field] = value + with pytest.raises(ValueError): + sink.deliver(event) + assert receiver.calls == [] + + +@pytest.mark.parametrize('field', ['job', 'revision', 'host', 'manifest_sha256', 'agent', 'workflow']) +def test_each_foreign_scope_field_refuses_before_receiver(emitted, field): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + other = 'b'*64 if field == 'manifest_sha256' else 'urn:example:foreign' + event['event_id'] = run_id(replace(scope, **{field: other}), 600) + with pytest.raises(ValueError): + ScheduledEventSink(receiver, 'schedule', 'operator').deliver(event) + assert receiver.calls == [] + + +@pytest.mark.parametrize('change', ['duplicate', 'padding', 'version_bool', 'slot_bool', + 'slot_huge', 'extra', 'array', 'missing', 'nested']) +def test_noncanonical_or_unbounded_run_binding_refuses(emitted, change): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + binding = {**asdict(scope), 'scheduled_at': 600, 'version': 1} + if change == 'duplicate': + raw = json.dumps(binding, sort_keys=True, separators=(',', ':')) + raw = raw.replace('"scheduled_at":600', '"scheduled_at":600,"scheduled_at":600') + event['event_id'] = 'scheduled-v1-'+base64.urlsafe_b64encode(raw.encode()).decode().rstrip('=') + elif change == 'padding': + event['event_id'] += '=' + elif change == 'nested': + raw = '['*1500 + '0' + ']'*1500 + event['event_id'] = 'scheduled-v1-'+base64.urlsafe_b64encode(raw.encode()).decode().rstrip('=') + else: + if change == 'version_bool': + binding['version'] = True + elif change == 'slot_bool': + binding['scheduled_at'] = True + elif change == 'slot_huge': + binding['scheduled_at'] = 10**100 + elif change == 'extra': + binding['command'] = 'anything' + elif change == 'array': + binding = [binding] + elif change == 'missing': + del binding['version'] + event['event_id'] = encode(binding) + with pytest.raises(ValueError): + ScheduledEventSink(receiver, 'schedule', 'operator').deliver(event) + assert receiver.calls == [] + + +def test_missing_frozen_fields_are_not_filled_from_current_configuration(emitted): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + del event['event_id'] + with pytest.raises(ValueError): + ScheduledEventSink(receiver, 'schedule', 'operator').deliver(event) + assert receiver.calls == []