test: verify installed readbacks after physical SQLite upgrade

scripts/acceptance/command-keys/physical-sqlite-profile.json

@@ -0,0 +1,78 @@
+{
+  "baseline_index": "sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e",
+  "baseline_seed_execution": {
+    "backup_sha256": "7d1826f7681342bb3a5974509648e2e8d75e7a21d262f65f3ec1f58727ccbc0c",
+    "driver_git_blob": "96f3950e5113ac8484d72e64677ea2fbfab2facc",
+    "driver_path": "scripts/ci/command_key_upgrade_seed.py",
+    "driver_sha256": "01c0fa1251ac3c809723e58d71817881fe2c5c9498579bc4c2e6097324a3618b",
+    "receipts_sha256": "8c728ce827df3a11621dfb0e82cdd404bbab48784182a5e92c8d632d49047fc5",
+    "source_commit": "09a4ccda3da11853606af031a8970bbde4b2d6ec"
+  },
+  "baseline_source": "28f084513235f6647e740c4b3d9e3aace7a323b7",
+  "baseline_version": "3.8.0",
+  "journal_checksum_sha384": "53b752529a1f7addaefe62ab61160dd42b3990f805cfdc4ecd1b48927a6c2f7ce42526fc04522ac887a37208f343fc73",
+  "journal_description": "full frozen PHP schema, native worker incarnations, session holders, stream reclaim deadlines, service call deadlines, service admission counters and committed external payload deletions and task ready origins and poll expiry indexes and bound run command contracts and task queue activity dispatch pauses and keyed command intents",
+  "new_schema_objects": [
+    "dw_command_intents",
+    "dw_command_intents_namespace",
+    "sqlite_autoindex_dw_command_intents_1"
+  ],
+  "producer_contract_commit": "a09f156399416216d2f097c257a2164f83c7eda7",
+  "producer_manifest_sha256": "ceb4695643e48c7bfe3f81eea380c6ab9aa0b973da830cb552d070344047319f",
+  "schema": "durable-workflow.physical-sqlite-upgrade-profile.v1",
+  "schema_after": 13,
+  "schema_before": 12,
+  "server_index": "sha256:8ffcdb2a36af3129a1987849aafe8715fe0d3bf9a9648467f0b5cb8eeef43ba9",
+  "server_source": "5614274d2629d03f92932a07dfee0fb10349ddd9",
+  "server_version": "3.9.2",
+  "source_inputs": {
+    "rust/migrations/sqlite/0003_full_schema.sql": {
+      "git_blob": "3e58585408652a8e0622db183915f04e62349551",
+      "sha256": "661e8dc4ca329b5798bd661ebf1678fb43d813cdf61dd2d0c4e29978d908ccee"
+    },
+    "rust/migrations/sqlite/0004_worker_incarnations.sql": {
+      "git_blob": "375d6507b12fb469714180d4d7f57bdda0efc520",
+      "sha256": "51c59ea1bbd59129e5696660114139cadde1f0afdd984bba34f51c485b4798fe"
+    },
+    "rust/migrations/sqlite/0005_worker_session_holders.sql": {
+      "git_blob": "d7e099452e707fe6c93cb0b01187d2d92b8e3826",
+      "sha256": "9c6bc35eafb7f62535388b2f890509f63b2c877676e3e8fe63eb938e589e3fee"
+    },
+    "rust/migrations/sqlite/0006_stream_reclaims.sql": {
+      "git_blob": "481e73c9f0df21950e797dcd2caf784bde6aed20",
+      "sha256": "b7e99137ee8690612535639682c94e71a632d1ec45bb2b071b098425986a8a51"
+    },
+    "rust/migrations/sqlite/0007_service_calls.sql": {
+      "git_blob": "fa2de782c9ab06e195430bf47760c6b847f00451",
+      "sha256": "752a1a40556d940c0a99d22fd5e4b4419e9a97078cde29802d80337fbcf17ed9"
+    },
+    "rust/migrations/sqlite/0008_external_payload_deletions.sql": {
+      "git_blob": "3faec1972524eaa783fd8039c26d92be591107cf",
+      "sha256": "d66f652cb2bdd96ffb57c402d94b9ac3831e6425040c90cc6c2653c50dad961b"
+    },
+    "rust/migrations/sqlite/0009_task_ready_origin.sql": {
+      "git_blob": "559740942e5e7c15afbd75241adcea616bc564b0",
+      "sha256": "e5d47c5062ca0953151e315086323e173f5b2bd222c0806cef035aaa9417fcd8"
+    },
+    "rust/migrations/sqlite/0010_poll_expiry_indexes.sql": {
+      "git_blob": "23d7c3e6797e8c0a211bab6706fba2359d4e4e39",
+      "sha256": "4d4fb5e03ad1780c191dda35af0721de022009af2efedb9408bdefe1ace3a5b4"
+    },
+    "rust/migrations/sqlite/0011_run_command_contracts.sql": {
+      "git_blob": "eaacac26c1ee44289da98804cc669f1cddd8b416",
+      "sha256": "385f4134f409493178a11044348643ee15022ce050919ea88836a3e1d00c19f9"
+    },
+    "rust/migrations/sqlite/0012_task_queue_pauses.sql": {
+      "git_blob": "fa43906db894d91d7142c0a76d6213b311796ba2",
+      "sha256": "484ab3a84c09fbc27b6bb39965b4075a1ce6fac4751a209670fe148812beed1f"
+    },
+    "rust/migrations/sqlite/0013_command_intents.sql": {
+      "git_blob": "295c7f89e281fd9a8275a4100d525d566687145b",
+      "sha256": "fc3ad1730682c09704027dcbabc17cd6ded9c642a3ff20d7988d5f4b4a5726fd"
+    },
+    "rust/src/runtime/schema.rs": {
+      "git_blob": "78f4fc9f804f9c8b5130846db9cc9164c85179aa",
+      "sha256": "1575bb959f0abf3092da1c369bcd5ac8f13f621706f1f640da1d8e7cf2ddadd2"
+    }
+  }
+}

scripts/acceptance/command-keys/physical_upgrade.py

@@ -0,0 +1,207 @@
+"""Read a producer-owned physical seed without synthesizing legacy identities."""
+from concurrent.futures import ThreadPoolExecutor
+from copy import deepcopy
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import stat
+
+import execution_guards as guards
+import sqlite_legacy
+
+
+PROFILE = Path(__file__).with_name('physical-sqlite-profile.json')
+PROFILE_SHA256 = '8f9c4695b814a6c3f923885e51d1900506906138b7e0d27dc87b8df592a36e9e'
+
+
+def profile():
+    raw = PROFILE.read_bytes()
+    if hashlib.sha256(raw).hexdigest() != PROFILE_SHA256:
+        raise ValueError('physical profile differs from the reviewed tagged source')
+    return json.loads(raw)
+
+
+def require_handoff(commit):
+    selected = profile()
+    if commit != selected['producer_contract_commit']:
+        raise ValueError('producer handoff commit differs from the reviewed physical profile')
+    return selected
+
+
+def attribution(commit, declared_execution):
+    selected = require_handoff(commit)
+    if declared_execution != selected['baseline_seed_execution']:
+        raise ValueError('historical seed source/driver identity differs from the frozen execution')
+    return {'accepted_contract_source':commit,
+            'baseline_seed_execution':deepcopy(selected['baseline_seed_execution'])}
+
+
+def read_owned(path, digest, limit):
+    path = Path(path)
+    if (not path.is_absolute() or path.resolve() != path
+            or not re.fullmatch('[0-9a-f]{64}', digest)):
+        raise ValueError('physical input needs a canonical path and exact SHA-256')
+    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
+    try:
+        if not 0 < os.fstat(descriptor).st_size <= limit:
+            raise ValueError('physical input exceeds its byte bound')
+        record = guards._record(descriptor)
+        raw = os.pread(descriptor, limit + 1, 0)
+        if (record['sha256'] != digest or len(raw) != record['size']
+                or hashlib.sha256(raw).hexdigest() != digest
+                or guards._record(descriptor) != record
+                or guards._identity(path.lstat()) != tuple(record['identity'])):
+            raise ValueError('physical input identity/content differs')
+        return raw, record
+    finally:
+        os.close(descriptor)
+
+
+def prepare(backup, backup_sha, receipts, receipts_sha):
+    selected = profile()
+    seed = selected['baseline_seed_execution']
+    if backup_sha != seed['backup_sha256'] or receipts_sha != seed['receipts_sha256']:
+        raise ValueError('physical baseline provenance differs from the frozen seed execution')
+    raw, original = read_owned(backup, backup_sha, sqlite_legacy.MAX_DATABASE_BYTES)
+    for suffix in ('-wal', '-journal'):
+        sidecar = Path(str(backup) + suffix)
+        if sidecar.is_symlink() or (sidecar.exists() and sidecar.stat().st_size):
+            raise ValueError('physical backup must be stopped and checkpointed')
+    packet_raw, _ = read_owned(receipts, receipts_sha, 4 * 1024 * 1024)
+    packet = json.loads(packet_raw)
+    assert packet['schema'] == 'durable-workflow.command-key-physical-seed.v1'
+    assert packet['manifest_sha256'] == selected['producer_manifest_sha256']
+    assert packet['expected_external_baseline'] == {
+        'tag':selected['baseline_version'],'source_commit':selected['baseline_source'],
+        'index_digest':selected['baseline_index']}
+    assert len(packet['cases']) == 6
+    assert len({row['workflow_id'] for row in packet['cases']}) == 6
+    assert len({row['receipt']['command_id'] for row in packet['cases']}) == 6
+    assert {(row['kind'],row['state']) for row in packet['cases']} == {
+        (kind,state) for kind in ('signal','update') for state in ('open','completed','continued')}
+    for row in packet['cases']:
+        receipt = row['receipt']
+        assert receipt['workflow_id'] == row['workflow_id']
+        assert all(isinstance(receipt.get(key),str) and receipt[key] for key in ('command_id','run_id'))
+        if row['kind'] == 'update':
+            assert all(isinstance(receipt.get(key),str) and receipt[key] for key in ('update_id','accepted_at','update_name'))
+            assert isinstance(row['original_body']['request_id'],str) and row['original_body']['request_id']
+    return raw, original, packet, selected
+
+
+def copy_backup(raw, destination):
+    parent = destination.parent.stat()
+    if (not destination.is_absolute() or destination.parent.resolve() != destination.parent
+            or not stat.S_ISDIR(parent.st_mode) or parent.st_uid != os.getuid()
+            or stat.S_IMODE(parent.st_mode) != 0o700):
+        raise ValueError('physical copy needs an owned private parent')
+    with open(os.open(destination,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600),'wb') as output:
+        output.write(raw)
+
+
+def validate_upgrade(before, after, observed, started, finished, selected):
+    assert observed == {'backend':'sqlite','from_schema_version':selected['schema_before'],
+                        'schema_version':selected['schema_after'],'status':'upgraded'}, observed
+    assert set(after['tables']) == set(before['tables']) | {'dw_command_intents'}
+    assert not after['tables']['dw_command_intents']['rows']
+    expected = deepcopy(before)
+    version = expected['tables']['dw_server_schema']['columns'].index('version')
+    assert len(expected['tables']['dw_server_schema']['rows']) == 1
+    assert expected['tables']['dw_server_schema']['rows'][0][version] == ['integer',selected['schema_before']]
+    expected['tables']['dw_server_schema']['rows'][0][version] = ['integer',selected['schema_after']]
+    journal = after['tables']['_sqlx_migrations']
+    assert len(journal['rows']) == 1
+    entry = dict(zip(journal['columns'],journal['rows'][0]))
+    assert set(entry) == {'version','description','installed_on','success','checksum','execution_time'}
+    assert entry['version'] == ['integer',selected['schema_after']] and entry['success'] == ['integer',1]
+    assert entry['execution_time'] == ['integer',0]
+    assert entry['description'] == ['text',selected['journal_description']]
+    assert entry['checksum'] == ['blob',selected['journal_checksum_sha384']]
+    assert entry['installed_on'][0] == 'text'
+    installed = datetime.strptime(entry['installed_on'][1],'%Y-%m-%d %H:%M:%S').replace(tzinfo=timezone.utc).timestamp()
+    assert started - 1 <= installed <= finished + 1
+    expected['tables']['_sqlx_migrations'] = journal
+    projected = deepcopy(after)
+    del projected['tables']['dw_command_intents']
+    added = [row for row in projected['schema'] if row not in before['schema']]
+    assert {row[1] for row in added} == set(selected['new_schema_objects'])
+    assert all(row[2] == 'dw_command_intents' for row in added)
+    projected['schema'] = [row for row in projected['schema'] if row in before['schema']]
+    assert projected == expected, 'physical upgrade altered undeclared business rows/catalog'
+
+
+def validate_original_commands(value, packet):
+    commands = value['tables']['workflow_commands']
+    columns = commands['columns']
+    for case in packet['cases']:
+        original = case['receipt']
+        matches = [dict(zip(columns,row)) for row in commands['rows']
+                   if row[columns.index('id')] == ['text',original['command_id']]]
+        assert len(matches) == 1
+        row = matches[0]
+        assert row['command_type'] == ['text',case['kind']]
+        assert row['workflow_instance_id'] == ['text',case['workflow_id']]
+        assert row['workflow_run_id'] == ['text',original['run_id']]
+        expected_key = ['null'] if case['kind'] == 'signal' else ['text',case['original_body']['request_id']]
+        assert row['request_id'] == expected_key
+
+
+def readbacks(packet, invoke, snapshot, work):
+    before = snapshot()
+    validate_original_commands(before,packet)
+    assert not before['tables']['dw_command_intents']['rows']
+    outcomes = []
+    for case in packet['cases']:
+        original = case['receipt']
+        history_args = {'op':'history','workflow_id':case['workflow_id'],'run_id':original['run_id']}
+        history = invoke(history_args)
+        assert history['kind'] == 'ok'
+        events = history['value']['history_events']
+        event_type = 'SignalReceived' if case['kind'] == 'signal' else 'UpdateAccepted'
+        expected_count = sum(event['event_type'] == event_type for event in case['history']['events'])
+        assert expected_count > 0 and sum(event['type'] == event_type for event in events) == expected_count
+        if case['kind'] == 'signal':
+            outcomes.append({'kind':'signal','state':case['state'],'scope':'NULL-key compatibility only',
+                             'original_command_id':original['command_id'],'original_run_id':original['run_id'],
+                             'history_sha256':sqlite_legacy.fingerprint(events)})
+            continue
+        request = {'op':'update_raw','workflow_id':case['workflow_id'],'name':original['update_name'],
+                   'request_id':case['original_body']['request_id'],'input':[999]}
+        fields = ('command_id','run_id','update_id','accepted_at')
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            replies = list(pool.map(lambda _:invoke(request),range(4)))
+        replies.append(invoke(dict(request,run_id=original['run_id'])))
+        for reply in replies:
+            assert reply['kind'] == 'ok' and reply['value']['status'] in (200,202), reply
+            body = reply['value']['body']
+            assert body['retained'] is True and body['intent_verified'] is False
+            assert body['request_id'] == request['request_id']
+            assert all(body[field] == original[field] for field in fields)
+            if case['state'] != 'open':assert body['update_status'] == 'completed'
+        high = invoke(dict(request,op='update'))
+        if case['state'] == 'open':
+            assert high['kind'] == 'pending' and high['receipt']['retained'] is True
+            assert high['receipt']['intent_verified'] is False
+            assert all(high['receipt'][field] == original[field] for field in fields)
+        else:
+            assert high['kind'] == 'ok' and high['value'] is None
+        for changed in (dict(request,name='other'),dict(request,run_id='00000000000000000000000000')):
+            refused = invoke(changed)
+            assert refused['kind'] == 'ok' and refused['value']['status'] == 409
+            assert refused['value']['body']['reason'] == 'request_id_conflict'
+            assert refused['value']['body']['retained_command_id'] == original['command_id']
+        assert invoke(history_args)['value']['history_events'] == events
+        outcomes.append({'kind':'update','state':case['state'],'outcome':'pass',
+                         'original':{field:original[field] for field in fields},
+                         'returned':{field:replies[0]['value']['body'][field] for field in fields},
+                         'retained':True,'intent_verified':False,'concurrency':4})
+    after = snapshot()
+    assert after == before, 'physical SDK request window altered full logical rows/schema'
+    for label,value in (('before',before),('after',after)):
+        with open(os.open(work/('physical-'+label+'-snapshot.json'),os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600),'w') as output:
+            json.dump(value,output,sort_keys=True)
+    return {'cell':'physical-sqlite-upgrade-sdk-readback-only','outcome':'pass',
+            'cases':outcomes,'request_window_sha256':sqlite_legacy.fingerprint(before)}