Add published update receipt upgrade acceptance

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.

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}))

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

tests/Fixtures/UpdateRequestKeyConsumer/consumer.ini

@@ -0,0 +1 @@
+auto_prepend_file=/fixture/definition.php

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())