diff --git a/tests/Fixtures/UpdateRequestKeyConsumer/README.md b/tests/Fixtures/UpdateRequestKeyConsumer/README.md new file mode 100644 index 000000000..bdb7f8f24 --- /dev/null +++ b/tests/Fixtures/UpdateRequestKeyConsumer/README.md @@ -0,0 +1,83 @@ +# Published update-key upgrade consumer + +This fixture covers the stop-first legacy-writer upgrade required by Server +#505. It creates genuine completed updates using an old immutable Server image, +stops every old writer, and checks the same receipts on a corrected immutable +image. It never repairs receipt columns or restarts an old writer after upgrade. +The SDK is a normal PyPI 2.6.1 installation. Its production RECORD hashes are +checked before and after each phase. The mounted PHP/Python files are business +definitions, with no replacement of product or vendor code. + +Use a new Compose project and empty disposable phase directory for each mode, +`service` and `embedded`. Set `KEY_SERVER_IMAGE` to the exact old digest, +`KEY_CLIENT_IMAGE` to a pinned Python image, `KEY_MODE`, `KEY_COHORT`, and an +absolute `KEY_PHASE_ROOT`. Create `data` and `evidence` beneath the phase root +as the runtime UID. No host port is published. Queue and API roles use one CPU +and 512 MiB each; dependency installation uses a private client tmpfs. + +1. Run `docker compose -f compose.yml -p PROJECT up -d --wait api client`. + In embedded mode, also enable `--profile embedded` and start `queue`. +2. In the client, run `/opt/venv/bin/python /fixture/consumer.py create`. + Each of the default/named originals has a separate 30-second observation, + recorded 60-second run and 120-second execution deadline. The SDK worker + and embedded database queue execute actual typed updates and completion. +3. Stop the API and queue explicitly with SIGTERM, require exit 0/no OOM, and + remove those old-writer containers. The completed SDK worker must also have + exited successfully. Retain only the stopped database and client. +4. Audit with ordinary Python: bind the phase root read-only, mount only its + `evidence` subdirectory separately writable, and run `audit.py before` with + a private `/audit` tmpfs. This copies the stopped DELETE-mode + database, opens the copy read-only, proves unchanged source bytes and records + complete original receipt/history/task rows. WAL presence is a refusal. +5. Change only `KEY_SERVER_IMAGE` to the corrected published digest. Recreate + bootstrap and API on that same database. Never start the old image again. +6. In the client run `/opt/venv/bin/python /fixture/consumer.py retained`. + The same body key, a new proxy header, changed arguments and a different + declared handler must return the original typed result and exact durable + identities without changing history. +7. Stop the new API. Run `audit.py after` as above. It must retain every old + receipt/run/history/task field, both original deadlines, zero repairs and + zero registrations. Additional schema columns may exist after migration. + +Record the source commit, image indexes/configs, installed package provenance, +actual start/end times, outcomes and any failed original in #505. Remove the +project's containers, network, dependencies, SQLite data and exports at the +declared cleanup boundary. Preserve borrowed images and other projects. A +passing upgrade does not qualify the separate intake-limit image controls or +the broader signal/update retry contract in shared #216. + +## Published intake-limit and retained-rejection controls + +Use a separate fresh project with the corrected published image and +`KEY_MODE=service`. Keep the same normal Python 2.6.1 installation. This cohort +creates four ordinary waiting runs: pending-update and payload-size controls +in both default and named namespaces. Each run retains its 60/120-second +run/execution deadlines; every consumer stage has a 30-second total bound. + +1. Start API/client with `KEY_PENDING_LIMIT=0 KEY_PAYLOAD_LIMIT=0`, then run + `/opt/venv/bin/python /fixture/intake_limits.py create` in the client. + The real Worker commits each original wait, stops, and two pending updates + are admitted before testing the limits. +2. Stop the API with SIGTERM and require exit 0/no OOM. Run + `intake_limits.py audit-original` in an ordinary Python container, with the + phase root mounted read-only, its evidence directory mounted at `/evidence` + writable, a private `/audit` tmpfs and no network. +3. Recreate only the API with `KEY_PENDING_LIMIT=1 KEY_PAYLOAD_LIMIT=0`. + Run the consumer's `pending_update_count` stage. Both namespaces must return + HTTP 409, `structural_limit_exceeded`, the pending-count kind and limit 1. +4. Stop the API. Recreate it with `KEY_PENDING_LIMIT=0 KEY_PAYLOAD_LIMIT=1` + and run `payload_size`. Both namespaces must return the payload-size refusal. + Stop it and run `audit-refused` with the same read-only mounts. +5. Recreate the API with both limits 0. Run `retained`, which changes the proxy + header, handler and arguments while keeping each rejected caller key. + Every response must retain its original rejection, update/command identities + and limit metadata. Stop the API and run `audit-retained`. + +A new limit refusal deliberately records one `UpdateRejected` event. The audit +requires exactly four such events, preserved original history rows, unchanged +tasks and original run state/deadlines. Only the corresponding run sequence and +progress bookkeeping may advance. Retained retries permit no durable row changes +at all, and the complete stopped database hash must stay exact. Each audit also +checks zero repairs, registrations and failed jobs. Do not extend an original's +deadline or resume an unsuccessful cohort as a new pass. Retire raw databases, +SDK dependencies, runtime and network after review; preserve borrowed images. diff --git a/tests/Fixtures/UpdateRequestKeyConsumer/audit.py b/tests/Fixtures/UpdateRequestKeyConsumer/audit.py new file mode 100644 index 000000000..25d2c8134 --- /dev/null +++ b/tests/Fixtures/UpdateRequestKeyConsumer/audit.py @@ -0,0 +1,48 @@ +"""Run with both API and queue stopped, source bind read-only, /audit tmpfs.""" + +import datetime +import hashlib +import json +import os +import shutil +import sqlite3 +import sys +from pathlib import Path + +root = Path('/phase') +source = root / 'data/database.sqlite' +assert not Path(str(source) + '-wal').exists(), 'Stop and checkpoint SQLite before audit' +before_hash = hashlib.sha256(source.read_bytes()).hexdigest() +shutil.copyfile(source, '/audit/database.sqlite') +db = sqlite3.connect('file:/audit/database.sqlite?mode=ro', uri=True) +db.row_factory = sqlite3.Row +db.execute('PRAGMA query_only=ON') +tables = {} +for table in ('workflow_runs', 'workflow_commands', 'workflow_updates', 'workflow_history_events', 'workflow_tasks'): + tables[table] = [dict(row) for row in db.execute('SELECT * FROM ' + table + ' ORDER BY id')] +assert len(tables['workflow_runs']) == 2 +assert len(tables['workflow_updates']) == 2 +for run in tables['workflow_runs']: + assert run['status'] == 'completed', run + start = datetime.datetime.fromisoformat(run['started_at']) + assert (datetime.datetime.fromisoformat(run['run_deadline_at']) - start).total_seconds() == 60 + assert (datetime.datetime.fromisoformat(run['execution_deadline_at']) - start).total_seconds() == 120 +assert all(row['status'] == 'completed' and row['repair_count'] == 0 for row in tables['workflow_tasks']) +assert all(row['status'] == 'completed' for row in tables['workflow_updates']) +assert db.execute('SELECT COUNT(*) FROM workflow_worker_registrations').fetchone()[0] == 0 +db.close() +snapshot_path = root / 'evidence' / (os.environ['KEY_MODE'] + '-rows.json') +if sys.argv[1] == 'before': + assert any(row.get('request_id') == 'legacy-caller-key' for row in tables['workflow_commands']) + snapshot_path.write_text(json.dumps(tables, ensure_ascii=False, indent=2)) +else: + original = json.loads(snapshot_path.read_text()) + for table, rows in original.items(): + assert len(rows) == len(tables[table]), table + for old, new in zip(rows, tables[table], strict=True): + assert all(new[key] == value for key, value in old.items()), (table, old['id']) +assert hashlib.sha256(source.read_bytes()).hexdigest() == before_hash +print(json.dumps({'stage': sys.argv[1], 'mode': os.environ['KEY_MODE'], 'runs': 2, + 'updates': 2, 'tasks': len(tables['workflow_tasks']), + 'history_events': len(tables['workflow_history_events']), + 'database_sha256': before_hash, 'unchanged_source_bytes': True})) diff --git a/tests/Fixtures/UpdateRequestKeyConsumer/compose.yml b/tests/Fixtures/UpdateRequestKeyConsumer/compose.yml new file mode 100644 index 000000000..d780bf0f9 --- /dev/null +++ b/tests/Fixtures/UpdateRequestKeyConsumer/compose.yml @@ -0,0 +1,86 @@ +services: + bootstrap: + image: ${KEY_SERVER_IMAGE:?Set an immutable published image} + user: "1000:1000" + cpus: 1 + mem_limit: 512m + memswap_limit: 512m + command: [server-bootstrap] + environment: &server_environment + DB_CONNECTION: sqlite + DB_DATABASE: /data/database.sqlite + DB_SQLITE_JOURNAL_MODE: DELETE + CACHE_STORE: file + QUEUE_CONNECTION: database + DW_MODE: ${KEY_MODE:?Select service or embedded} + DW_POLLING_CACHE_PATH: /data/poll-cache + DW_AUTH_DRIVER: token + DW_AUTH_TOKEN: update-key-consumer + DW_ADMIN_TOKEN: update-key-consumer + DW_WORKER_TOKEN: update-key-worker + DW_BOOTSTRAP_RETRIES: 1 + DW_V2_LIMIT_PENDING_UPDATES: ${KEY_PENDING_LIMIT:-500} + DW_V2_LIMIT_PAYLOAD_SIZE_BYTES: ${KEY_PAYLOAD_LIMIT:-2097152} + volumes: &server_volumes + - ${KEY_PHASE_ROOT:?Set a disposable phase directory}/data:/data + - ./definition.php:/fixture/definition.php:ro + - ./consumer.ini:/usr/local/etc/php/conf.d/zz-key-consumer.ini:ro + + api: + image: ${KEY_SERVER_IMAGE:?} + user: "1000:1000" + cpus: 1 + mem_limit: 512m + memswap_limit: 512m + environment: *server_environment + volumes: *server_volumes + depends_on: + bootstrap: + condition: service_completed_successfully + + queue: + image: ${KEY_SERVER_IMAGE:?} + user: "1000:1000" + cpus: 1 + mem_limit: 512m + memswap_limit: 512m + environment: *server_environment + volumes: *server_volumes + command: [php, artisan, queue:work, --queue=default, --sleep=1, --tries=1] + profiles: [embedded] + depends_on: + api: + condition: service_healthy + + client: + image: ${KEY_CLIENT_IMAGE:?Set a pinned ordinary Python image} + user: "1000:1000" + init: true + cpus: 1 + mem_limit: 512m + memswap_limit: 512m + environment: + HOME: /tmp + KEY_MODE: ${KEY_MODE:?} + KEY_COHORT: ${KEY_COHORT:?Set a new original cohort identity} + entrypoint: [sh, -ec] + command: + - | + python -m venv /opt/venv + /opt/venv/bin/pip install --index-url https://pypi.org/simple --no-cache-dir --quiet durable-workflow==2.6.1 + exec sleep infinity + tmpfs: + - /opt/venv:rw,exec,uid=1000,gid=1000,size=268435456 + healthcheck: + test: [CMD, /opt/venv/bin/python, -c, "import durable_workflow"] + interval: 2s + timeout: 5s + retries: 30 + volumes: + - ${KEY_PHASE_ROOT:?}:/phase + - ./consumer.py:/fixture/consumer.py:ro + - ./definition.py:/fixture/definition.py:ro + - ./intake_limits.py:/fixture/intake_limits.py:ro + depends_on: + api: + condition: service_healthy diff --git a/tests/Fixtures/UpdateRequestKeyConsumer/consumer.ini b/tests/Fixtures/UpdateRequestKeyConsumer/consumer.ini new file mode 100644 index 000000000..d93770d1f --- /dev/null +++ b/tests/Fixtures/UpdateRequestKeyConsumer/consumer.ini @@ -0,0 +1 @@ +auto_prepend_file=/fixture/definition.php diff --git a/tests/Fixtures/UpdateRequestKeyConsumer/consumer.py b/tests/Fixtures/UpdateRequestKeyConsumer/consumer.py new file mode 100644 index 000000000..e81e9ff12 --- /dev/null +++ b/tests/Fixtures/UpdateRequestKeyConsumer/consumer.py @@ -0,0 +1,145 @@ +"""Real old-writer receipt creation and separate post-upgrade HTTP readback.""" + +import asyncio +import base64 +import hashlib +import importlib.metadata +import json +import os +import sys +from pathlib import Path + +import httpx +from definition import WaitingWorkflow + +from durable_workflow import Client, Worker, serializer + +ROOT = Path('/phase') +TOKEN = 'update-key-consumer' +EXPECTED = {'count': 17, 'typed': [9223372036854775807, None, 'Київ']} +MODE = os.environ['KEY_MODE'] + + +def verify_install(): + dist = importlib.metadata.distribution('durable-workflow') + assert dist.version == '2.6.1' + checked = 0 + for entry in dist.files or []: + if not str(entry).startswith('durable_workflow/') or entry.hash is None: + continue + path = dist.locate_file(entry) + assert str(path).startswith('/opt/venv/'), path + digest = base64.urlsafe_b64encode(hashlib.sha256(path.read_bytes()).digest()) + assert digest.decode().rstrip('=') == entry.hash.value.rstrip('='), str(entry) + checked += 1 + assert checked > 25, checked + return checked + + +class ObservedClient(Client): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.registered = asyncio.Event() + + async def register_worker(self, **kwargs): + result = await super().register_worker(**kwargs) + self.registered.set() + return result + + +async def original(namespace): + identity = os.environ['KEY_COHORT'] + '-' + MODE + '-' + namespace + queue = 'q-' + identity if MODE == 'service' else 'default' + headers = {'Authorization': 'Bearer ' + TOKEN, 'X-Namespace': namespace, + 'X-Durable-Workflow-Control-Plane-Version': '2'} + evidence_path = ROOT / 'evidence' / (MODE + '-' + namespace + '.json') + evidence_path.parent.mkdir(exist_ok=True) + async with ( + Client('http://api:8080', token=TOKEN, namespace=namespace) as client, + ObservedClient('http://api:8080', token='update-key-worker', namespace=namespace) as worker_client, + httpx.AsyncClient(base_url='http://api:8080', headers=headers, timeout=10) as http, + ): + running = None + worker = None + if MODE == 'service': + worker = Worker(worker_client, task_queue=queue, workflows=[WaitingWorkflow], + worker_id=identity, poll_timeout=1, heartbeat_interval=1, + max_concurrent_workflow_tasks=1, shutdown_timeout=5) + running = asyncio.create_task(worker.run()) + await asyncio.wait_for(worker_client.registered.wait(), 8) + try: + handle = await client.start_workflow( + workflow_type=('update-key-consumer' if MODE == 'service' + else 'UpdateRequestKeyConsumer\\WaitingWorkflow'), + task_queue=queue, workflow_id=identity, input=[], + run_timeout_seconds=60, execution_timeout_seconds=120, + ) + evidence_path.with_suffix('.original.json').write_text(json.dumps({ + 'workflow_id': identity, 'run_id': handle.run_id, + 'namespace': namespace, 'mode': MODE, + })) + response = await http.post('/api/workflows/' + identity + '/update/add', json={ + 'input': {'codec': 'avro', 'blob': serializer.encode([17])}, + 'request_id': 'legacy-caller-key', 'wait_for': 'completed', + 'wait_timeout_seconds': 5, + }, headers={'X-Request-Id': 'legacy-caller-key'}) + assert response.status_code == 200, (response.status_code, response.text) + receipt = response.json() + assert receipt['update_status'] == 'completed', receipt + assert receipt['result'] == EXPECTED, receipt + await handle.signal('finish', []) + assert await handle.result(timeout=8) == EXPECTED + history = await http.get('/api/workflows/' + identity + '/runs/' + handle.run_id + '/history') + history.raise_for_status() + evidence = {'workflow_id': identity, 'run_id': handle.run_id, + 'namespace': namespace, 'mode': MODE, 'receipt': receipt, + 'history': history.json(), 'result': EXPECTED} + evidence_path.write_text(json.dumps(evidence, ensure_ascii=False, indent=2)) + finally: + if worker is not None: + await worker.stop() + if running is not None: + await asyncio.wait_for(running, 8) + + +async def retained(namespace): + path = ROOT / 'evidence' / (MODE + '-' + namespace + '.json') + original = json.loads(path.read_text()) + headers = {'Authorization': 'Bearer ' + TOKEN, 'X-Namespace': namespace, + 'X-Durable-Workflow-Control-Plane-Version': '2'} + async with httpx.AsyncClient(base_url='http://api:8080', headers=headers, timeout=10) as http: + response = await http.post('/api/workflows/' + original['workflow_id'] + '/update/other', json={ + 'input': {'codec': 'avro', 'blob': serializer.encode([99])}, + 'request_id': 'legacy-caller-key', 'wait_for': 'completed', + 'wait_timeout_seconds': 5, + }, headers={'X-Request-Id': 'new-proxy-correlation'}) + assert response.status_code == 200, (response.status_code, response.text) + actual = response.json() + for key in ('update_id', 'command_id', 'run_id', 'update_name', 'update_status', 'result'): + assert actual[key] == original['receipt'][key], (key, actual, original['receipt']) + history = await http.get('/api/workflows/' + original['workflow_id'] + + '/runs/' + original['run_id'] + '/history') + history.raise_for_status() + assert history.json()['events'] == original['history']['events'] + path.with_suffix('.retained.json').write_text(json.dumps(actual, ensure_ascii=False, indent=2)) + + +async def main(): + assert MODE in ('service', 'embedded') + stage = sys.argv[1] + assert stage in ('create', 'retained') + checked = verify_install() + async with httpx.AsyncClient(base_url='http://api:8080', timeout=10) as http: + if stage == 'create': + response = await http.post('/api/namespaces', json={'name': 'key-upgrade'}, headers={ + 'Authorization': 'Bearer ' + TOKEN, 'X-Durable-Workflow-Control-Plane-Version': '2', + }) + assert response.status_code == 201, (response.status_code, response.text) + for namespace in ('default', 'key-upgrade'): + await asyncio.wait_for((original if stage == 'create' else retained)(namespace), 30) + assert verify_install() == checked + print(json.dumps({'stage': stage, 'mode': MODE, 'originals': 2, + 'installed_sdk': '2.6.1', 'verified_files': checked})) + + +asyncio.run(main()) diff --git a/tests/Fixtures/UpdateRequestKeyConsumer/definition.php b/tests/Fixtures/UpdateRequestKeyConsumer/definition.php new file mode 100644 index 000000000..1191961e1 --- /dev/null +++ b/tests/Fixtures/UpdateRequestKeyConsumer/definition.php @@ -0,0 +1,44 @@ +result(); + } + + #[UpdateMethod('add')] + public function add(int $amount): array + { + $this->count += $amount; + + return $this->result(); + } + + #[UpdateMethod('other')] + public function other(int $amount): array + { + return $this->add($amount); + } + + private function result(): array + { + return ['count' => $this->count, 'typed' => [PHP_INT_MAX, null, 'Київ']]; + } +} diff --git a/tests/Fixtures/UpdateRequestKeyConsumer/definition.py b/tests/Fixtures/UpdateRequestKeyConsumer/definition.py new file mode 100644 index 000000000..559c45842 --- /dev/null +++ b/tests/Fixtures/UpdateRequestKeyConsumer/definition.py @@ -0,0 +1,28 @@ +from durable_workflow import workflow + + +@workflow.defn(name="update-key-consumer") +class WaitingWorkflow: + def __init__(self): + self.count = 0 + self.finished = False + + @workflow.update("add") + def add(self, amount: int) -> dict: + self.count += amount + return self.result() + + @workflow.update("other") + def other(self, amount: int) -> dict: + return self.add(amount) + + @workflow.signal("finish") + def finish(self): + self.finished = True + + def run(self, ctx): + yield ctx.wait_condition(lambda: self.finished, key="finish") + return self.result() + + def result(self): + return {"count": self.count, "typed": [9223372036854775807, None, "Київ"]} diff --git a/tests/Fixtures/UpdateRequestKeyConsumer/intake_limits.py b/tests/Fixtures/UpdateRequestKeyConsumer/intake_limits.py new file mode 100644 index 000000000..5eed09552 --- /dev/null +++ b/tests/Fixtures/UpdateRequestKeyConsumer/intake_limits.py @@ -0,0 +1,253 @@ +"""Published API intake limits and immutable retained rejection readback.""" + +import asyncio +import base64 +import datetime +import hashlib +import importlib.metadata +import json +import os +import shutil +import sqlite3 +import sys +from pathlib import Path + +ROOT = Path('/phase') +TABLES = ('workflow_commands', 'workflow_updates', 'workflow_runs', + 'workflow_tasks', 'workflow_history_events') +TOKEN = 'update-key-consumer' + + +def encode_blob(value): + if isinstance(value, bytes): + return {'base64': base64.b64encode(value).decode()} + raise TypeError(type(value).__name__) + + +def audit(stage): + source = ROOT / 'data/database.sqlite' + assert not Path(str(source) + '-wal').exists(), 'stopped DELETE database required' + before = hashlib.sha256(source.read_bytes()).hexdigest() + copied = Path('/audit/database.sqlite') + shutil.copyfile(source, copied) + db = sqlite3.connect('file:' + str(copied) + '?mode=ro', uri=True) + db.row_factory = sqlite3.Row + db.execute('PRAGMA query_only=ON') + snapshot = {} + for table in TABLES: + rows = db.execute('SELECT * FROM ' + table + ' ORDER BY id LIMIT 1001').fetchall() + assert len(rows) <= 1000, table + snapshot[table] = [dict(row) for row in rows] + snapshot = json.loads(json.dumps(snapshot, default=encode_blob)) + assert len(snapshot['workflow_runs']) == 4 + assert all(row['status'] == 'waiting' for row in snapshot['workflow_runs']) + assert sum(row['repair_count'] for row in snapshot['workflow_tasks']) == 0 + assert db.execute('SELECT COUNT(*) FROM workflow_worker_registrations').fetchone()[0] == 0 + assert db.execute('SELECT COUNT(*) FROM failed_jobs').fetchone()[0] == 0 + for run in snapshot['workflow_runs']: + started = datetime.datetime.fromisoformat(run['started_at']) + assert (datetime.datetime.fromisoformat(run['run_deadline_at']) - started).total_seconds() == 60 + assert (datetime.datetime.fromisoformat(run['execution_deadline_at']) - started).total_seconds() == 120 + if stage == 'audit-original': + assert len(snapshot['workflow_updates']) == 2 + assert all(row['status'] == 'accepted' for row in snapshot['workflow_updates']) + elif stage == 'audit-refused': + original = json.loads((ROOT / 'evidence/audit-original.json').read_text())['snapshot'] + assert snapshot['workflow_tasks'] == original['workflow_tasks'], 'tasks changed after refusal' + original_events = {row['id'] for row in original['workflow_history_events']} + preserved = [row for row in snapshot['workflow_history_events'] if row['id'] in original_events] + assert preserved == original['workflow_history_events'] + added = [row for row in snapshot['workflow_history_events'] if row['id'] not in original_events] + assert len(added) == 4 + assert {row['workflow_run_id'] for row in added} == {row['id'] for row in original['workflow_runs']} + for event in added: + assert event['event_type'] == 'UpdateRejected' and event['workflow_task_id'] is None + payload = json.loads(event['payload']) + assert payload['workflow_command_id'] == event['workflow_command_id'] + assert payload['command']['rejection_reason'] == 'structural_limit_exceeded' + assert payload['command']['request_id'] == 'over-limit' + assert payload['validation_errors']['arguments'] + current_runs = {row['id']: row for row in snapshot['workflow_runs']} + bookkeeping = {'last_history_sequence', 'last_command_sequence', 'last_progress_at', 'updated_at'} + for run in original['workflow_runs']: + current = current_runs[run['id']] + assert {k: v for k, v in current.items() if k not in bookkeeping} == { + k: v for k, v in run.items() if k not in bookkeeping + } + assert current['last_history_sequence'] == run['last_history_sequence'] + 1 + assert current['last_command_sequence'] == run['last_command_sequence'] + 1 + for table in ('workflow_commands', 'workflow_updates'): + current = {row['id']: row for row in snapshot[table]} + assert all(current[row['id']] == row for row in original[table]), table + assert len(current) == len(original[table]) + 4, table + assert sum(row['status'] == 'rejected' for row in snapshot['workflow_updates']) == 4 + elif stage == 'audit-retained': + refused = json.loads((ROOT / 'evidence/audit-refused.json').read_text())['snapshot'] + assert snapshot == refused, 'retained receipt read mutated durable rows' + else: + raise ValueError(stage) + db.close() + assert hashlib.sha256(source.read_bytes()).hexdigest() == before + assert hashlib.sha256(copied.read_bytes()).hexdigest() == before + result = {'stage': stage, 'database_sha256': before, 'source_unchanged': True, + 'counts': {table: len(rows) for table, rows in snapshot.items()}, 'snapshot': snapshot} + (Path('/evidence') / (stage + '.json')).write_text(json.dumps(result, ensure_ascii=False, indent=2)) + print(json.dumps({key: value for key, value in result.items() if key != 'snapshot'})) + + +def verify_install(): + dist = importlib.metadata.distribution('durable-workflow') + assert dist.version == '2.6.1' + checked = 0 + for entry in dist.files or []: + if not str(entry).startswith('durable_workflow/') or entry.hash is None: + continue + path = dist.locate_file(entry) + assert str(path).startswith('/opt/venv/'), path + actual = base64.urlsafe_b64encode(hashlib.sha256(path.read_bytes()).digest()).decode().rstrip('=') + assert actual == entry.hash.value.rstrip('='), str(entry) + checked += 1 + assert checked > 25 + return checked + + +async def business(stage): + import httpx + from durable_workflow import Client, Worker, serializer + from definition import WaitingWorkflow + + checked = verify_install() + + def headers(namespace): + return {'Authorization': 'Bearer ' + TOKEN, 'X-Namespace': namespace, + 'X-Durable-Workflow-Control-Plane-Version': '2'} + + async def update(http, case, name, key, amount, proxy): + return await http.post('/api/workflows/' + case['workflow_id'] + '/update/' + name, + json={'input': {'codec': 'avro', 'blob': serializer.encode([amount])}, + 'request_id': key}, + headers={**headers(case['namespace']), 'X-Request-Id': proxy}) + + originals_path = ROOT / 'evidence/intake-originals.json' + async with httpx.AsyncClient(base_url='http://api:8080', timeout=5) as http: + if stage == 'create': + response = await http.post('/api/namespaces', json={'name': 'intake-limit'}, headers=headers('default')) + assert response.status_code == 201, (response.status_code, response.text) + cases = [] + + class ObservedClient(Client): + def __init__(self, **kwargs): + super().__init__('http://api:8080', **kwargs) + self.registered = asyncio.Event() + self.completed = 0 + + async def register_worker(self, **kwargs): + result = await super().register_worker(**kwargs) + self.registered.set() + return result + + async def complete_workflow_task(self, **kwargs): + result = await super().complete_workflow_task(**kwargs) + self.completed += 1 + return result + + for namespace in ('default', 'intake-limit'): + async with ( + Client('http://api:8080', token=TOKEN, namespace=namespace) as client, + ObservedClient(token='update-key-worker', namespace=namespace) as worker_client, + ): + queue = os.environ['KEY_COHORT'] + '-' + namespace + worker = Worker(worker_client, task_queue=queue, workflows=[WaitingWorkflow], + worker_id=queue, poll_timeout=1, heartbeat_interval=1, + max_concurrent_workflow_tasks=1, shutdown_timeout=5) + running = asyncio.create_task(worker.run()) + try: + await asyncio.wait_for(worker_client.registered.wait(), 8) + for kind in ('pending_update_count', 'payload_size'): + identity = queue + '-' + kind + expected = worker_client.completed + 1 + handle = await client.start_workflow( + workflow_type='update-key-consumer', workflow_id=identity, + task_queue=queue, input=[], run_timeout_seconds=60, execution_timeout_seconds=120, + ) + case = {'namespace': namespace, 'kind': kind, 'workflow_id': identity, + 'run_id': handle.run_id} + cases.append(case) + (ROOT / 'evidence/intake-admitted.json').write_text(json.dumps(cases)) + async def completed(): + while worker_client.completed < expected: + assert not running.done(), 'worker stopped before the initial wait' + await asyncio.sleep(0.02) + await asyncio.wait_for(completed(), 8) + finally: + await worker.stop() + await asyncio.wait_for(running, 8) + for case in cases: + if case['kind'] == 'pending_update_count': + response = await update(http, case, 'add', 'first-pending', 17, 'initial-proxy') + assert response.status_code == 202, (response.status_code, response.text) + case['first_receipt'] = response.json() + response = await http.get('/api/workflows/' + case['workflow_id'], headers=headers(case['namespace'])) + response.raise_for_status() + case['original'] = response.json() + assert case['original']['run_id'] == case['run_id'] + assert case['original']['status'] == 'waiting' + response = await http.get('/api/workflows/' + case['workflow_id'] + '/runs/' + case['run_id'] + '/history', + headers=headers(case['namespace'])) + response.raise_for_status() + case['history'] = response.json()['events'] + originals_path.write_text(json.dumps(cases, ensure_ascii=False, indent=2)) + print(json.dumps({'stage': stage, 'originals': len(cases), 'installed_hashes': checked, + 'run_ids': [case['run_id'] for case in cases]})) + else: + cases = json.loads(originals_path.read_text()) + selected = cases if stage == 'retained' else [case for case in cases if case['kind'] == stage] + assert len(selected) == (4 if stage == 'retained' else 2) + results = [] + for case in selected: + retained = stage == 'retained' + response = await update(http, case, 'other' if retained else 'add', 'over-limit', + 99 if retained else 17, 'changed-proxy' if retained else 'original-proxy') + assert response.status_code == 409, (response.status_code, response.text) + result = response.json() + assert result['reason'] == 'structural_limit_exceeded', result + assert result['structural_limit_kind'] == case['kind'], result + assert result['structural_limit_configured'] == 1, result + assert result['structural_limit_value'] >= 1, result + assert result['run_id'] == case['run_id'], result + path = ROOT / 'evidence' / (case['namespace'] + '-' + case['kind'] + '.rejection.json') + if retained: + original = json.loads(path.read_text())['receipt'] + for field in ('reason', 'command_id', 'update_id', 'run_id', 'structural_limit_kind', + 'structural_limit_configured', 'structural_limit_value'): + assert result[field] == original[field], (field, result, original) + else: + path.write_text(json.dumps({'receipt': result}, ensure_ascii=False, indent=2)) + response = await http.get('/api/workflows/' + case['workflow_id'], headers=headers(case['namespace'])) + response.raise_for_status() + current = response.json() + for field in ('status', 'run_id', 'run_deadline_at', 'execution_deadline_at'): + assert current[field] == case['original'][field], (field, current, case['original']) + response = await http.get('/api/workflows/' + case['workflow_id'] + '/runs/' + case['run_id'] + '/history', + headers=headers(case['namespace'])) + response.raise_for_status() + events = response.json()['events'] + assert events[:len(case['history'])] == case['history'] + assert len(events) == len(case['history']) + 1 + assert events[-1]['event_type'] == 'UpdateRejected' + if retained: + assert events == json.loads(path.read_text())['history'] + else: + path.write_text(json.dumps({'receipt': result, 'history': events}, ensure_ascii=False, indent=2)) + results.append({'run_id': case['run_id'], 'kind': case['kind'], 'namespace': case['namespace'], + 'command_id': result['command_id'], 'update_id': result['update_id']}) + (ROOT / 'evidence' / (stage + '.json')).write_text(json.dumps(results, ensure_ascii=False, indent=2)) + print(json.dumps({'stage': stage, 'results': results, 'installed_hashes': checked})) + + +if __name__ == '__main__': + selected_stage = sys.argv[1] + if selected_stage.startswith('audit-'): + audit(selected_stage) + else: + asyncio.run(asyncio.wait_for(business(selected_stage), timeout=30))