diff --git a/docs/demos/fixed-outcome-executor.log b/docs/demos/fixed-outcome-executor.log new file mode 100644 index 0000000..300e28a --- /dev/null +++ b/docs/demos/fixed-outcome-executor.log @@ -0,0 +1,10 @@ +Script started on 2026-10-15 08:37:01-04:00 [COMMAND="stty -onlcr && .venv/bin/python -I scripts/demo-fixed-outcome-executor.py" ] +Receiver identity positive control: True +Fixed outcome executor available: True +Child: {'outcome': 'succeeded', 'exit_code': 0} +Child: {'outcome': 'skipped-by-hold', 'exit_code': 0} +Child: {'outcome': 'failed', 'exit_code': 7} +Exit-zero child without result: UNKNOWN; preserve claim, no retry. +No production schedule, signing key, graph or cron was used. + +Script done on 2026-10-15 08:37:01-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/fixed-outcome-executor.timing b/docs/demos/fixed-outcome-executor.timing new file mode 100644 index 0000000..9775484 --- /dev/null +++ b/docs/demos/fixed-outcome-executor.timing @@ -0,0 +1,6 @@ +0.030625 80 +0.026840 48 +0.023159 54 +0.016917 45 +0.020396 67 +0.000159 61 diff --git a/docs/demos/scheduled-projection.log b/docs/demos/scheduled-projection.log new file mode 100644 index 0000000..90b1b9a --- /dev/null +++ b/docs/demos/scheduled-projection.log @@ -0,0 +1,10 @@ +Script started on 2026-10-15 07:48:58-04:00 [COMMAND="stty -onlcr && .venv/bin/python scripts/demo-scheduled-projection.py" ] +Receiver identity positive control: True +Pinned schedule projection available: True +Effective 1250; first eligible aligned slot: 1800 +Three observations; two current-slot events: 2 +Next deadline: 3000 +Missing required identity: REFUSED; retained outbox rows: 2 +Execution callbacks: 0; no sink or production schedule enabled. + +Script done on 2026-10-15 07:48:58-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/scheduled-projection.timing b/docs/demos/scheduled-projection.timing new file mode 100644 index 0000000..519fafe --- /dev/null +++ b/docs/demos/scheduled-projection.timing @@ -0,0 +1,5 @@ +0.032912 41 +0.000014 43 +0.001719 50 +0.070044 67 +0.000009 124 diff --git a/docs/fixed-outcome-executor.md b/docs/fixed-outcome-executor.md new file mode 100644 index 0000000..46c69ae --- /dev/null +++ b/docs/fixed-outcome-executor.md @@ -0,0 +1,39 @@ +# Fixed command outcome adapter + +`FixedOutcomeExecutor` is a synchronous callback for the scheduled receiver. A +trusted caller constructs it with fixed absolute argv, an existing directory, +an explicit environment, a result descriptor variable and bounded runtime/output. +Delivery metadata cannot choose those values. The slot is validated as identity +metadata and is never interpolated into the command. The adapter does not inherit +the caller's ambient environment or invoke a shell. + +The child receives a writable anonymous descriptor in the configured variable. +It must write one bounded JSON object with `schema_version: 1`, `outcome`, and +`exit_code`; a bounded optional `reason` is accepted and discarded. Supported +outcomes are `succeeded`, `failed`, and `skipped-by-hold`. Failure requires a +nonzero exit, and the other outcomes require zero. The declared exit must match +the actual process exit. Duplicate/extra keys, malformed JSON, missing results, +signals, output overflow, timeout and unconfirmed cleanup raise `ExecutionUnknown`. +Neither exit zero nor a log message is sufficient evidence of success. + +The caller passes this callback to `scheduled_receiver.receive`, after the +receiver has durably claimed and admitted the slot. A valid hold remains distinct +from success. Lost publication replays the retained outcome without executing the +child again. Unknown execution preserves the claim for operator disposition; +it must never cause an automatic second execution. + +Combined stdout/stderr is drained and discarded under a byte limit. Results have +a separate 4096-byte limit. A new process group permits cleanup of ordinary +descendants, but is not kernel containment: a process that creates a new session +can escape that group. Host execution still needs its approved cgroup/supervisor, +immutable artifact and manifest admission, credential policy and timeout policy. +The adapter does not provide parent-session survival or artifact custody. + +Outcome metadata is unsigned. The surrounding signed receiver/port must retain +the documented distinction between signed lifecycle state and unsigned metadata. +This module neither enables a schedule nor installs a production executor. + +Run `python scripts/demo-fixed-outcome-executor.py` for isolated child-process +examples. The retained recording shows a real descriptor result, an explicit +hold, a nonzero failure, and an exit-zero child whose missing descriptor result +is refused. Tests also check timeout cleanup and durable no-repeat behavior. diff --git a/docs/scheduled-projection.md b/docs/scheduled-projection.md new file mode 100644 index 0000000..65680c1 --- /dev/null +++ b/docs/scheduled-projection.md @@ -0,0 +1,45 @@ +# Pinned fixed-interval projection + +`scheduled_projection.ScheduledProjection` consumes the versioned declaration +catalogue produced by the graph reader. Its constructor pins one execution +scope, catalogue graph, manifest target IRI, owner, anchor, interval, timezone +and effective UTC time from trusted local admission. The manifest digest stays +inside `ExecutionScope`; a target IRI is not a command or a digest substitute. + +Evaluation requires the declared identity to be an explicit verified membership +control. A missing, duplicate, malformed or changed declaration raises; it never +means an empty healthy catalogue. The caller must treat failure as UNKNOWN and +retain its durable deadlines, verdicts and delivery outbox. Disabled declarations +remain explicit BLOCKED records. Their next check is null, so change-feed routing +and bounded catalogue reconciliation must observe subsequent enablement. + +The result uses the existing incremental adapter envelope: one scoped record +and its `next_checks` deadline. An already-due eligible slot gets the canonical +signed-run identifier, `UNBLOCKED` and the pinned owner. Configure its emitter +with `event: scheduled`, `baseline_emits: true`, and no generation-based identity +rewriting. Repeated observations deduplicate in the actual emitter. A later +slot advances even while earlier receipts remain pending; success acceptance +and missed-run alerts are separate responsibilities. + +Only the current aligned slot is projected, never historical catch-up work. +The first permitted slot is the anchor-aligned occurrence at or after the +explicit effective time. UTC arithmetic preserves fixed elapsed seconds through +DST; it does not implement local calendar recurrence. The caller supplies a +trusted current clock and must preserve UNKNOWN when that clock is uncertain. +There is no persistent clock watermark or backwards-clock policy in this module. + +This is a library, not a configured adapter process or service. It performs no +query, write, child execution, receipt verification, graph profile installation +or cron change. The configured catalogue transport, change subscription, sink +factory, bounded executor, deadline-success verifier and activation remain +separate integration work. It does not clear the child-survival or runtime gates. + +`python scripts/demo-scheduled-projection.py` demonstrates actual emitter outbox +deduplication and refusal without executing anything. The fixture's runtime +admission fields remain UNKNOWN. `test_scheduled_projection.py` covers activation +boundaries, missing identity retention, pinned scope refusal, disabled state, +time bounds and a real DST transition. + +Replay the output-only [recording](demos/scheduled-projection.log) with its +[timing](demos/scheduled-projection.timing) using `scriptreplay`. On base0b4a0eb5 +the same demo reports the projection absent and its receiver control present. diff --git a/fixed_outcome_executor.py b/fixed_outcome_executor.py new file mode 100644 index 0000000..de6d168 --- /dev/null +++ b/fixed_outcome_executor.py @@ -0,0 +1,159 @@ +"""Fixed trusted command with a bounded, unsigned outcome descriptor. + +Manifest/artifact admission and durable slot claims belong to the signed receiver. +Nothing in a delivery can select argv, environment, directory or a descriptor name. +""" +from __future__ import annotations + +import json +import math +import os +from pathlib import Path +import re +import selectors +import signal +import subprocess +import time +from types import MappingProxyType + +MAX_RESULT = 4096 + + +class ExecutionUnknown(RuntimeError): + """No confirmed outcome; preserve the receiver claim and never rerun it.""" + + +def _pairs(pairs): + out = {} + for key, value in pairs: + if key in out: + raise ValueError('duplicate result field') + out[key] = value + return out + + +def _constant(_): + raise ValueError('non-finite JSON result') + + +def decode_result(raw: bytes, actual_code: int) -> dict: + """Validate the child channel against the actual process exit, not its logs.""" + if (type(raw) is not bytes or not 0 < len(raw) <= MAX_RESULT or + type(actual_code) is not int or not 0 <= actual_code <= 255): + raise ExecutionUnknown('bounded result and actual normal process exit required') + try: + value = json.loads(raw, object_pairs_hook=_pairs, parse_constant=_constant) + required = {'schema_version', 'outcome', 'exit_code'} + if (type(value) is not dict or not required <= value.keys() or + value.keys() - required - {'reason'} or + type(value['schema_version']) is not int or value['schema_version'] != 1 or + type(value['exit_code']) is not int or value['exit_code'] != actual_code or + value['outcome'] not in ('succeeded', 'failed', 'skipped-by-hold') or + (value['outcome'] == 'failed') != (actual_code != 0) or + ('reason' in value and (type(value['reason']) is not str or + not 0 < len(value['reason']) <= 256))): + raise ValueError('inconsistent process result') + return {'outcome': value['outcome'], 'exit_code': actual_code} + except (ValueError, UnicodeError, RecursionError): + raise ExecutionUnknown('invalid or inconsistent child result') from None + + +class FixedOutcomeExecutor: + def __init__(self, argv, *, cwd, env: dict, result_fd_env: str, + timeout_s: float, output_limit: int = 1048576): + if (type(argv) not in (list, tuple) or not 0 < len(argv) <= 64 or + any(type(a) is not str or not a or '\0' in a or len(a) > 8192 for a in argv) or + sum(map(len, argv)) > 131072 or not os.path.isabs(argv[0])): + raise ValueError('bounded fixed argv with an absolute executable required') + cwd = os.fspath(cwd) + if type(cwd) is not str or not os.path.isabs(cwd) or not Path(cwd).is_dir(): + raise ValueError('existing absolute trusted working directory required') + if (type(result_fd_env) is not str or + not re.fullmatch(r'[A-Z][A-Z0-9_]{0,124}_FD', result_fd_env) or + type(env) is not dict or result_fd_env in env or len(env) > 256 or + any(type(k) is not str or not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]{0,127}', k) or + type(v) is not str or '\0' in v or len(v) > 32768 for k, v in env.items())): + raise ValueError('bounded explicit environment and unused result descriptor name required') + if sum(len(k)+len(v) for k, v in env.items()) > 131072: + raise ValueError('environment exceeds bound') + if (type(timeout_s) not in (int, float) or not 0 < timeout_s <= 86400 or + not math.isfinite(timeout_s) or type(output_limit) is not int or + not 0 < output_limit <= 8388608): + raise ValueError('bounded timeout and output limit required') + self._argv, self._cwd = tuple(argv), cwd + self._env = MappingProxyType(dict(env)) + self._fd_env, self._timeout, self._output_limit = result_fd_env, timeout_s, output_limit + + def __call__(self, scheduled_at: int) -> dict: + if type(scheduled_at) is not int or not 0 <= scheduled_at <= 253402300799: + raise ValueError('bounded integer UTC slot required') + # The slot is identity metadata, never an argument interpolation source. + read_fd, write_fd = os.pipe() + proc = None + selector = selectors.DefaultSelector() + deadline = time.monotonic() + self._timeout + try: + proc = subprocess.Popen(self._argv, cwd=self._cwd, + env={**self._env, self._fd_env: str(write_fd)}, stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=False, + pass_fds=(write_fd,), start_new_session=True) + os.close(write_fd) + write_fd = None + for fd, kind in ((read_fd, 'result'), (proc.stdout.fileno(), 'output')): + os.set_blocking(fd, False) + selector.register(fd, selectors.EVENT_READ, kind) + result, output = bytearray(), 0 + while selector.get_map(): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ExecutionUnknown('command or inherited descriptors exceeded timeout') + for key, _ in selector.select(min(remaining, 0.1)): + try: + chunk = os.read(key.fd, 65536) + except BlockingIOError: + continue + if not chunk: + selector.unregister(key.fd) + elif key.data == 'result': + result.extend(chunk) + if len(result) > MAX_RESULT: + raise ExecutionUnknown('child result exceeded limit') + else: + output += len(chunk) + if output > self._output_limit: + raise ExecutionUnknown('combined child output exceeded limit') + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ExecutionUnknown('command exceeded timeout') + code = proc.wait(timeout=remaining) + # A normal exit is not completion if ordinary descendants remain. + try: + os.killpg(proc.pid, 0) + except ProcessLookupError: + pass + else: + raise ExecutionUnknown('command left descendants in its process group') + return decode_result(bytes(result), code) + except (OSError, subprocess.SubprocessError): + raise ExecutionUnknown('command invocation or completion unavailable') from None + finally: + selector.close() + os.close(read_fd) + if write_fd is not None: + os.close(write_fd) + if proc is not None: + cleanup_failed = False + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + except OSError: + cleanup_failed = True + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + cleanup_failed = True + finally: + proc.stdout.close() + if cleanup_failed: + raise ExecutionUnknown('child cleanup unconfirmed') from None diff --git a/pyproject.toml b/pyproject.toml index 6de0bc1..9af3df7 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_event_sink", "scheduled_receiver", "shuttle_process_port", "slot_receiver", + "fixed_outcome_executor", "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_projection", "scheduled_receiver", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scheduled_projection.py b/scheduled_projection.py new file mode 100644 index 0000000..8832ca3 --- /dev/null +++ b/scheduled_projection.py @@ -0,0 +1,106 @@ +"""Pure projection of one pinned declaration; no reads, writes or execution. + +The caller supplies a fresh, bounded catalogue from the declaration reader. +Errors must remain UNKNOWN in the caller, preserving its durable queue. This +module neither caches a failed catalogue nor grants executable admission. +""" +from __future__ import annotations + +from dataclasses import dataclass +import math +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from scheduled_receiver import ExecutionScope, run_id + +MAX_TIME = 253402300799 + + +def _text(value): + return type(value) is str and 0 < len(value) <= 2048 + + +@dataclass(frozen=True) +class ScheduledProjection: + scope: ExecutionScope + graph: str + target: str + owner: str + anchor_s: int + interval_s: int + timezone: str + effective_at: int + + def __post_init__(self): + self.scope.validate() + if not all(_text(v) for v in (self.graph, self.target, self.owner, self.timezone)): + raise ValueError('bounded pinned graph, target, owner and timezone required') + if len(self.owner) > 256: + raise ValueError('owner must fit the pinned receiver recipient') + for value in (self.anchor_s, self.effective_at): + if type(value) is not int or not 0 <= value <= MAX_TIME: + raise ValueError('bounded integer UTC anchors required') + if type(self.interval_s) is not int or not 0 < self.interval_s <= MAX_TIME: + raise ValueError('bounded positive fixed interval required') + try: + ZoneInfo(self.timezone) + except (ValueError, ZoneInfoNotFoundError) as exc: + raise ValueError('valid pinned IANA timezone required') from exc + + def evaluate(self, catalogue: dict, now: float) -> dict: + """One current eligible slot, never a catch-up execution list. + + UTC arithmetic intentionally preserves fixed elapsed seconds through + DST. Calendar recurrence is not supported. effective_at is trusted + local activation policy, not a delivery timestamp or graph command. + """ + if (type(now) not in (int, float) or not 0 <= now <= MAX_TIME or + not math.isfinite(now)): + raise ValueError('bounded current UTC time required') + root = {'schema_version', 'source', 'graph', 'membership', 'jobs', + 'executable_admission', 'signed_run_status', 'firing_eligibility'} + if (type(catalogue) is not dict or set(catalogue) != root or + type(catalogue['schema_version']) is not int or + catalogue['schema_version'] != 1 or catalogue['source'] != 'declarations_only' or + catalogue['graph'] != self.graph or any(catalogue[k] != 'UNKNOWN' for k in + ('executable_admission', 'signed_run_status', 'firing_eligibility'))): + raise ValueError('pinned versioned declaration catalogue required') + membership, jobs = catalogue['membership'], catalogue['jobs'] + if (type(membership) is not dict or set(membership) != {'status', 'required'} or + membership['status'] != 'required_identities_verified' or + type(membership['required']) is not list or + not 0 < len(membership['required']) <= 256 or + not all(_text(v) for v in membership['required']) or + type(jobs) is not list or not 0 < len(jobs) <= 256): + raise ValueError('bounded catalogue and positive membership controls required') + required = membership['required'] + identities = [j.get('iri') if type(j) is dict else None for j in jobs] + if (not all(_text(v) for v in identities) or len(set(identities)) != len(identities) or + len(set(required)) != len(required) or self.scope.job not in required or + not set(required) <= set(identities)): + raise ValueError('missing or duplicate required job identity') + job = jobs[identities.index(self.scope.job)] + fields = {'iri', 'name', 'revision', 'owner', 'enabled', 'host', 'target', + 'event', 'schedule', 'interval_s', 'anchor_s', 'timezone'} + pins = {'revision': self.scope.revision, 'host': self.scope.host, + 'target': self.target, 'owner': self.owner, 'interval_s': self.interval_s, + 'anchor_s': self.anchor_s, 'timezone': self.timezone} + if (set(job) != fields or type(job['enabled']) is not bool or + type(job['anchor_s']) is not int or type(job['interval_s']) is not int or + not all(_text(job[k]) for k in ('name', 'event', 'schedule')) or + any(job[k] != v for k, v in pins.items())): + raise ValueError('declaration differs from pinned schedule and execution scope') + first = self.anchor_s + max(0, (self.effective_at - self.anchor_s + + self.interval_s - 1) // self.interval_s) * self.interval_s + record = {'item': self.scope.job, 'owner': self.owner, 'verdict': 'BLOCKED'} + next_check = first if first <= MAX_TIME else None + if not job['enabled']: + next_check = None + elif now >= first: + slot = self.anchor_s + (math.floor(now) - self.anchor_s) // self.interval_s * self.interval_s + record.update(verdict='UNBLOCKED', event_id=run_id(self.scope, slot), + evidence='Current eligible UTC slot; historical slots are not replayed.') + next_check = slot + self.interval_s + if next_check > MAX_TIME: + next_check = None + return {'version': 1, 'scope': [self.scope.job], 'records': [record], + 'next_checks': {self.scope.job: next_check}} diff --git a/scripts/demo-fixed-outcome-executor.py b/scripts/demo-fixed-outcome-executor.py new file mode 100644 index 0000000..418f1ea --- /dev/null +++ b/scripts/demo-fixed-outcome-executor.py @@ -0,0 +1,35 @@ +"""Local temporary child processes only; no signing, network or production job.""" +import importlib.util +import json +from pathlib import Path +import sys +import tempfile + +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, run_id # noqa: E402 + +scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') +print('Receiver identity positive control:', run_id(scope, 600).startswith('scheduled-v1-')) +present = importlib.util.find_spec('fixed_outcome_executor') is not None +print('Fixed outcome executor available:', present) +if not present: + sys.exit(0) +from fixed_outcome_executor import ExecutionUnknown, FixedOutcomeExecutor # noqa: E402 + +with tempfile.TemporaryDirectory() as root: + for outcome, code in [('succeeded', 0), ('skipped-by-hold', 0), ('failed', 7), (None, 0)]: + packet = json.dumps(dict(schema_version=1, outcome=outcome, exit_code=code)).encode() + program = ('import os,sys; ' + f'os.write(int(os.environ["RESULT_FD"]), {packet!r}); sys.exit({code})') + if outcome is None: + program = 'pass' + call = FixedOutcomeExecutor([sys.executable, '-c', program], cwd=root, env={}, + result_fd_env='RESULT_FD', timeout_s=2) + try: + print('Child:', call(600)) + except ExecutionUnknown: + print('Exit-zero child without result: UNKNOWN; preserve claim, no retry.') +print('No production schedule, signing key, graph or cron was used.') diff --git a/scripts/demo-scheduled-projection.py b/scripts/demo-scheduled-projection.py new file mode 100644 index 0000000..987db04 --- /dev/null +++ b/scripts/demo-scheduled-projection.py @@ -0,0 +1,52 @@ +"""Pure pinned projection and actual emitter outbox; no network or execution.""" +import importlib.util +from pathlib import Path +import sys +import tempfile + +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 emitter import Emitter, connect, observe # noqa: E402 +from scheduled_receiver import ExecutionScope, run_id # noqa: E402 + +scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') +print('Receiver identity positive control:', run_id(scope, 600).startswith('scheduled-v1-')) +present = importlib.util.find_spec('scheduled_projection') is not None +print('Pinned schedule projection available:', present) +if not present: + sys.exit(0) +from scheduled_projection import ScheduledProjection # noqa: E402 + +p = ScheduledProjection(scope, 'urn:example:catalogue', 'urn:example:manifest', + 'operator', 600, 600, 'UTC', 1250) +catalogue = {'schema_version': 1, 'source': 'declarations_only', 'graph': p.graph, + 'executable_admission': 'UNKNOWN', 'signed_run_status': 'UNKNOWN', + 'firing_eligibility': 'UNKNOWN', + 'membership': {'status': 'required_identities_verified', 'required': [scope.job]}, + 'jobs': [{'iri': scope.job, 'name': 'fixture', 'revision': scope.revision, + 'owner': p.owner, 'enabled': True, 'host': scope.host, 'target': p.target, + 'event': 'urn:example:event', 'schedule': 'urn:example:schedule', + 'interval_s': 600, 'anchor_s': 600, 'timezone': 'UTC'}]} +print('Effective 1250; first eligible aligned slot:', p.evaluate(catalogue, 1250)['next_checks'][scope.job]) +with tempfile.TemporaryDirectory() as root: + conn = connect(Path(root)/'outbox.sqlite') + e = Emitter('fixture-schedule', ['unused'], 60, p.owner, + event='scheduled', baseline_emits=True) + try: + for now in (1801, 1802, 2401): + out = p.evaluate(catalogue, now) + observe(conn, e, out['records'], now, scope=set(out['scope'])) + print('Three observations; two current-slot events:', conn.execute( + 'SELECT COUNT(*) FROM outbox').fetchone()[0]) + print('Next deadline:', out['next_checks'][scope.job]) + catalogue['jobs'] = [] + try: + p.evaluate(catalogue, 3001) + except ValueError: + print('Missing required identity: REFUSED; retained outbox rows:', conn.execute( + 'SELECT COUNT(*) FROM outbox').fetchone()[0]) + print('Execution callbacks: 0; no sink or production schedule enabled.') + finally: + conn.close() diff --git a/test_fixed_outcome_executor.py b/test_fixed_outcome_executor.py new file mode 100644 index 0000000..d30dfba --- /dev/null +++ b/test_fixed_outcome_executor.py @@ -0,0 +1,164 @@ +from itertools import permutations +import json +import sys +import time + +import pytest + +from fixed_outcome_executor import ExecutionUnknown, FixedOutcomeExecutor, decode_result + + +def packet(outcome='succeeded', code=0): + return dict(schema_version=1, outcome=outcome, exit_code=code) + + +def executor(tmp_path, code, **options): + return FixedOutcomeExecutor([sys.executable, '-c', code], cwd=tmp_path, env={}, + result_fd_env='RESULT_FD', timeout_s=2, **options) + + +def program(value=None, *, raw=None, code=0, prefix=''): + if raw is None: + raw = json.dumps(packet() if value is None else value).encode() + return ('import os,sys; '+prefix+ + f'os.write(int(os.environ["RESULT_FD"]), {raw!r}); sys.exit({code})') + + +def test_result_invariant_exhausts_normal_exit_domain_and_field_order(): + for code in range(256): + for outcome in ('succeeded', 'failed', 'skipped-by-hold'): + value = packet(outcome, code) + for order in permutations(value): + raw = json.dumps({k: value[k] for k in order}).encode() + if (outcome == 'failed') == (code != 0): + assert decode_result(raw, code) == {'outcome': outcome, 'exit_code': code} + else: + with pytest.raises(ExecutionUnknown): + decode_result(raw, code) + + +@pytest.mark.parametrize('raw,code', [ + (b'', 0), (b'{}', 0), (b'[]', 0), (b'null', 0), (b'\xff', 0), + (b'{"schema_version":1,"schema_version":1,"outcome":"succeeded","exit_code":0}', 0), + (json.dumps(packet()).encode()+b'{}', 0), + (json.dumps(dict(packet(), schema_version=True)).encode(), 0), + (json.dumps(dict(packet(), exit_code=False)).encode(), 0), + (json.dumps(dict(packet(), exit_code=0.0)).encode(), 0), + (json.dumps(dict(packet(), reason=None)).encode(), 0), + (json.dumps(dict(packet(), reason='')).encode(), 0), + (json.dumps(dict(packet(), reason='x'*257)).encode(), 0), + (json.dumps(dict(packet(), extra=True)).encode(), 0), + (b'{"schema_version":1,"outcome":"succeeded","exit_code":NaN}', 0), + (json.dumps(packet()).encode(), 1), (json.dumps(packet()).encode(), -9), + (json.dumps(packet()).encode(), True), (b' '*4097, 0), (b'['*1500+b']'*1500, 0), +]) +def test_unavailable_or_inconsistent_result_is_unknown(raw, code): + with pytest.raises(ExecutionUnknown): + decode_result(raw, code) + + +@pytest.mark.parametrize('outcome,code', [('succeeded', 0), ('failed', 7), ('skipped-by-hold', 0)]) +def test_actual_process_exit_and_hold_are_distinct(tmp_path, outcome, code): + value = dict(packet(outcome, code), reason='fixture') + assert executor(tmp_path, program(value, code=code))(600) == {'outcome': outcome, 'exit_code': code} + + +def test_fixed_argv_and_explicit_environment_are_not_delivery_interpolation(tmp_path, monkeypatch): + monkeypatch.setenv('EXECUTOR_FIXTURE_SECRET', 'synthetic-not-a-credential') + literal = ['${HOME}', '$$', 'a b', "q'uote"] + code = ('import os,sys,json; assert "EXECUTOR_FIXTURE_SECRET" not in os.environ; ' + f'assert sys.argv[1:] == {literal!r}; ' + 'os.write(int(os.environ["RESULT_FD"]),json.dumps(' + f'{packet()!r}).encode())') + argv, env = [sys.executable, '-c', code, *literal], {} + call = FixedOutcomeExecutor(argv, cwd=tmp_path, env=env, + result_fd_env='RESULT_FD', timeout_s=2) + argv[:] = ['not-the-command'] + env['EXECUTOR_FIXTURE_SECRET'] = 'also-synthetic' + assert call(600) == {'outcome': 'succeeded', 'exit_code': 0} + + +@pytest.mark.parametrize('code', [ + 'pass', # exit zero without the explicit result is not success + program(packet('failed', 1), code=0), + program(raw=b'not-json'), + program(raw=b'x'*4097), + program(prefix='os.write(1,b"x"*4096); '), + program(prefix='os.write(2,b"x"*4096); '), +]) +def test_real_process_missing_mismatched_and_bounded_channels(tmp_path, code): + with pytest.raises(ExecutionUnknown): + executor(tmp_path, code, output_limit=1024)(600) + + +def test_timeout_stops_ordinary_descendants_before_their_side_effect(tmp_path): + marker = tmp_path/'late-side-effect' + code = ('import os,time; pid=os.fork(); ' + f'(time.sleep(0.7),open({str(marker)!r},"w").write("late")) if pid==0 else time.sleep(5)') + call = FixedOutcomeExecutor([sys.executable, '-c', code], cwd=tmp_path, env={}, + result_fd_env='RESULT_FD', timeout_s=0.15) + started = time.monotonic() + with pytest.raises(ExecutionUnknown): + call(600) + assert time.monotonic()-started < 3 + time.sleep(0.8) + assert not marker.exists() + + +def test_valid_result_cannot_acknowledge_a_still_running_parent(tmp_path): + code = program(prefix='import time; ', code=0).replace('; sys.exit(0)', '; time.sleep(5)') + call = FixedOutcomeExecutor([sys.executable, '-c', code], cwd=tmp_path, env={}, + result_fd_env='RESULT_FD', timeout_s=0.15) + with pytest.raises(ExecutionUnknown): + call(600) + + +@pytest.mark.parametrize('valid', [True, False]) +def test_durable_receiver_never_repeats_actual_child_after_publication_or_channel_failure(tmp_path, valid): + from scheduled_receiver import ExecutionScope, receive + from slot_receiver import SlotLedger, SlotUnknown + from dataclasses import asdict + + counter = tmp_path/'child-calls' + prefix = (f'from pathlib import Path; p=Path({str(counter)!r}); ' + 'p.write_text(str(int(p.read_text())+1) if p.exists() else "1"); ') + raw = json.dumps(packet('skipped-by-hold')).encode() if valid else b'not-json' + call = executor(tmp_path, program(raw=raw, prefix=prefix)) + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + ledger = SlotLedger(tmp_path/'slots.sqlite') + class Port: + publications = 0 + def admit(self, *_): pass + def started(self, *_): pass + def finished(self, *_): pass + def publish(self, *_): + self.publications += 1 + if self.publications == 1: + raise ConnectionError('synthetic lost publication') + port = Port() + binding = {**asdict(scope), 'scheduled_at': 600} + with pytest.raises(ConnectionError if valid else ExecutionUnknown): + receive(ledger, scope, 600, binding, port, lambda: call(600)) + if valid: + assert receive(ledger, scope, 600, binding, port, lambda: call(600))['outcome'] == 'skipped-by-hold' + assert port.publications == 2 + else: + with pytest.raises(SlotUnknown): + receive(ledger, scope, 600, binding, port, lambda: call(600)) + assert counter.read_text() == '1' + + +@pytest.mark.parametrize('key,value', [ + ('argv', 'not-an-argv'), ('argv', ['relative']), ('argv', [sys.executable, 'a\0b']), + ('env', {'RESULT_FD': '3'}), ('env', {'BAD-KEY': 'x'}), ('env', {'X': 'a\0b'}), + ('result_fd_env', 'PATH'), ('result_fd_env', 'lower_fd'), + ('timeout_s', 0), ('timeout_s', True), ('timeout_s', float('inf')), + ('output_limit', True), ('output_limit', 0), ('output_limit', 8388609), +]) +def test_invalid_config_refuses_before_any_child(tmp_path, key, value): + args = dict(argv=[sys.executable, '-c', 'pass'], cwd=tmp_path, env={}, + result_fd_env='RESULT_FD', timeout_s=1) + args[key] = value + with pytest.raises(ValueError): + FixedOutcomeExecutor(**args) diff --git a/test_scheduled_projection.py b/test_scheduled_projection.py new file mode 100644 index 0000000..1ad6ddf --- /dev/null +++ b/test_scheduled_projection.py @@ -0,0 +1,154 @@ +from copy import deepcopy +from dataclasses import replace +from datetime import datetime +import json +from zoneinfo import ZoneInfo + +import pytest + +from emitter import Emitter, connect, observe +from scheduled_projection import MAX_TIME, ScheduledProjection +from scheduled_receiver import ExecutionScope, run_id + + +@pytest.fixture +def projection(): + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + return ScheduledProjection(scope, 'urn:example:catalogue', 'urn:example:manifest', + 'operator', 600, 600, 'UTC', 1250) + + +def catalogue(p): + return {'schema_version': 1, 'source': 'declarations_only', 'graph': p.graph, + 'membership': {'status': 'required_identities_verified', 'required': [p.scope.job]}, + 'executable_admission': 'UNKNOWN', 'signed_run_status': 'UNKNOWN', + 'firing_eligibility': 'UNKNOWN', 'jobs': [{ + 'iri': p.scope.job, 'name': 'fixture', 'revision': p.scope.revision, + 'owner': p.owner, 'enabled': True, 'host': p.scope.host, 'target': p.target, + 'event': 'urn:example:event', 'schedule': 'urn:example:schedule', + 'interval_s': p.interval_s, 'anchor_s': p.anchor_s, 'timezone': p.timezone}]} + + +def test_boundaries_and_no_historical_catchup(projection): + p = projection + for now in (0, 600, 1249, 1250, 1799.999): + out = p.evaluate(catalogue(p), now) + assert out['records'][0]['verdict'] == 'BLOCKED' + assert out['next_checks'] == {p.scope.job: 1800} + for now, slot in ((1800, 1800), (2399.999, 1800), (2400, 2400), (600000, 600000)): + out = p.evaluate(catalogue(p), now) + assert len(out['records']) == 1 + assert out['records'][0]['event_id'] == run_id(p.scope, slot) + assert out['next_checks'] == {p.scope.job: slot+600} + + +def test_actual_emitter_deduplicates_slot_and_advances_without_receipt(tmp_path, projection): + p = projection + conn = connect(tmp_path/'outbox.sqlite') + e = Emitter('scheduled-fixture', ['unused'], 60, p.owner, + event='scheduled', baseline_emits=True) + try: + for now in (1801, 1802, 2401): + out = p.evaluate(catalogue(p), now) + observe(conn, e, out['records'], now, scope=set(out['scope'])) + rows = conn.execute('SELECT event_id,payload,delivered_at FROM outbox ORDER BY created').fetchall() + assert [r[0] for r in rows] == [run_id(p.scope, 1800), run_id(p.scope, 2400)] + assert all(json.loads(r[1])['recipient'] == p.owner and r[2] is None for r in rows) + before = conn.execute('SELECT * FROM verdicts').fetchall() + broken = catalogue(p) + broken['jobs'] = [] + with pytest.raises(ValueError): + out = p.evaluate(broken, 3001) + observe(conn, e, out['records'], 3001, scope=set(out['scope'])) + assert conn.execute('SELECT * FROM verdicts').fetchall() == before + assert conn.execute('SELECT COUNT(*) FROM outbox').fetchone()[0] == 2 + finally: + conn.close() + + +@pytest.mark.parametrize('field,value', [ + ('revision', 'other'), ('host', 'urn:example:foreign'), ('target', 'urn:example:other'), + ('owner', 'other'), ('anchor_s', True), ('anchor_s', 0), ('interval_s', True), + ('interval_s', 60), ('timezone', 'Europe/London'), ('enabled', 1), ('event', ''), + ('schedule', None), ('name', ''), ('iri', 'urn:example:missing'), +]) +def test_changed_declaration_refuses(projection, field, value): + data = catalogue(projection) + data['jobs'][0][field] = value + with pytest.raises(ValueError): + projection.evaluate(data, 1800) + + +@pytest.mark.parametrize('mutation', [ + lambda d: d.update(schema_version=True), lambda d: d.update(source='local_toml'), + lambda d: d.update(graph='urn:example:other'), lambda d: d.update(extra=True), + lambda d: d.update(executable_admission='ADMITTED'), lambda d: d.update(jobs=[]), + lambda d: d.update(jobs=d['jobs']*2), lambda d: d.update(jobs=[None]), + lambda d: d.update(membership={'status': 'UNKNOWN', 'required': []}), + lambda d: d['membership'].update(required=[]), + lambda d: d['membership'].update(required=['urn:example:missing']), + lambda d: d['membership'].update(required=d['membership']['required']*2), + lambda d: d['jobs'][0].update(extra=True), +]) +def test_failed_membership_and_protocol_never_mean_empty_healthy(projection, mutation): + data = deepcopy(catalogue(projection)) + mutation(data) + with pytest.raises(ValueError): + projection.evaluate(data, 1800) + + +def test_disabled_is_explicit_not_missing(projection): + data = catalogue(projection) + data['jobs'][0]['enabled'] = False + out = projection.evaluate(data, 600000) + assert out['records'] == [{'item': projection.scope.job, 'owner': projection.owner, + 'verdict': 'BLOCKED'}] + assert out['next_checks'] == {projection.scope.job: None} + + +def test_fixed_elapsed_intervals_across_dst_and_many_anchors(projection): + for anchor in (0, 17, 1700000000): + for interval in (1, 7, 600, 86400): + p = replace(projection, anchor_s=anchor, interval_s=interval, + effective_at=anchor+interval+1, timezone='America/New_York') + for k in range(2, 20): + now = anchor+k*interval + out = p.evaluate(catalogue(p), now) + assert out['records'][0]['event_id'] == run_id(p.scope, now) + assert out['next_checks'][p.scope.job] == now+interval + + +def test_dst_transition_preserves_elapsed_seconds_not_wall_clock(projection): + zone = ZoneInfo('America/New_York') + anchor = int(datetime(2023, 11, 4, 12, tzinfo=zone).timestamp()) + p = replace(projection, anchor_s=anchor, interval_s=86400, + effective_at=anchor, timezone=zone.key) + slot = anchor+86400 + assert datetime.fromtimestamp(slot, zone).hour == 11 # fall-back day + out = p.evaluate(catalogue(p), slot) + assert out['records'][0]['event_id'] == run_id(p.scope, slot) + assert out['next_checks'][p.scope.job] == slot+86400 + + +@pytest.mark.parametrize('now', [-1, True, float('nan'), float('inf'), MAX_TIME+1, '1800']) +def test_clock_bounds(projection, now): + with pytest.raises(ValueError): + projection.evaluate(catalogue(projection), now) + + +@pytest.mark.parametrize('field,value', [ + ('anchor_s', True), ('anchor_s', -1), ('effective_at', MAX_TIME+1), + ('interval_s', 0), ('interval_s', True), ('interval_s', MAX_TIME+1), + ('timezone', 'Not/AZone'), ('timezone', '/etc/passwd'), ('owner', ''), ('owner', 'x'*257), +]) +def test_pinned_configuration_bounds(projection, field, value): + with pytest.raises(ValueError): + replace(projection, **{field: value}) + + +def test_last_representable_slot_has_no_overflowing_deadline(projection): + p = replace(projection, anchor_s=MAX_TIME, interval_s=600, effective_at=MAX_TIME) + out = p.evaluate(catalogue(p), MAX_TIME) + assert out['records'][0]['event_id'] == run_id(p.scope, MAX_TIME) + assert out['next_checks'] == {p.scope.job: None}