Complete bounded recovery checks and reconcile blocked workplans
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e738-9e15-7600-87e6-90dec3a7beef
This commit is contained in:
parent
aea9796908
commit
9baf6f1829
21 changed files with 2301 additions and 53 deletions
238
research/experiments/storage-receiver/experiment.py
Normal file
238
research/experiments/storage-receiver/experiment.py
Normal file
|
|
@ -0,0 +1,238 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Bounded local mechanism experiment; never touches the checkout's Git refs."""
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import selectors
|
||||
import shutil
|
||||
import signal
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
HERE = Path(__file__).resolve()
|
||||
|
||||
|
||||
def connect(path):
|
||||
db = sqlite3.connect(path)
|
||||
assert db.execute('PRAGMA journal_mode=WAL').fetchone()[0] == 'wal'
|
||||
db.execute('PRAGMA synchronous=FULL')
|
||||
assert db.execute('PRAGMA synchronous').fetchone()[0] == 2
|
||||
return db
|
||||
|
||||
|
||||
def run(*args, cwd=None, check=True, input=None):
|
||||
# Ignore user Git configuration/hooks and inherited repository targeting.
|
||||
env = {k: v for k, v in os.environ.items() if not k.startswith('GIT_')}
|
||||
env.update(GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null',
|
||||
GIT_AUTHOR_NAME='Recovery Fixture', GIT_AUTHOR_EMAIL='fixture@example.invalid',
|
||||
GIT_COMMITTER_NAME='Recovery Fixture', GIT_COMMITTER_EMAIL='fixture@example.invalid',
|
||||
GIT_AUTHOR_DATE='2026-09-28T00:00:00Z', GIT_COMMITTER_DATE='2026-09-28T00:00:00Z')
|
||||
return subprocess.run(args, cwd=cwd, env=env, text=True, input=input,
|
||||
capture_output=True, check=check, timeout=20)
|
||||
|
||||
|
||||
def checkpoint():
|
||||
print('checkpoint', flush=True)
|
||||
# Parent kills only this dedicated fixture subprocess, after its marker.
|
||||
signal.pause()
|
||||
|
||||
|
||||
def worker(mode, root):
|
||||
db = connect(root / 'local.db')
|
||||
if mode == 'send':
|
||||
oid = db.execute("SELECT oid FROM pending WHERE id='send'").fetchone()[0]
|
||||
run('git', 'push', str(root / 'receiver.git'), f'{oid}:refs/heads/main', cwd=root / 'client')
|
||||
checkpoint() # Accepted remotely, local acknowledgement deliberately absent.
|
||||
return
|
||||
db.execute('BEGIN IMMEDIATE')
|
||||
db.execute("UPDATE pending SET state='superseded' WHERE id='old'")
|
||||
if mode == 'split':
|
||||
db.commit() # Deliberately broken negative control.
|
||||
checkpoint()
|
||||
return
|
||||
db.execute("UPDATE document SET body='new'")
|
||||
if mode == 'middle':
|
||||
checkpoint()
|
||||
return
|
||||
db.execute("INSERT INTO pending VALUES ('new','new-identity','pending')")
|
||||
if mode == 'before':
|
||||
checkpoint()
|
||||
return
|
||||
db.commit()
|
||||
checkpoint()
|
||||
|
||||
|
||||
def kill_at(mode, root):
|
||||
p = subprocess.Popen([sys.executable, str(HERE), '--worker', mode, '--root', str(root)],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||||
try:
|
||||
with selectors.DefaultSelector() as selector:
|
||||
selector.register(p.stdout, selectors.EVENT_READ)
|
||||
if not selector.select(timeout=20):
|
||||
raise RuntimeError(f'{mode}: checkpoint timeout')
|
||||
line = p.stdout.readline().strip()
|
||||
if line != 'checkpoint':
|
||||
raise RuntimeError(f'{mode}: no checkpoint: {line} {p.stderr.read()}')
|
||||
p.kill()
|
||||
p.communicate(timeout=5)
|
||||
assert p.returncode == -signal.SIGKILL
|
||||
return p.returncode
|
||||
finally:
|
||||
if p.poll() is None:
|
||||
p.kill()
|
||||
p.communicate(timeout=5)
|
||||
|
||||
|
||||
def initialize(root):
|
||||
root.mkdir()
|
||||
with connect(root / 'local.db') as db:
|
||||
db.executescript('CREATE TABLE document(body TEXT); '
|
||||
'CREATE TABLE pending(id TEXT PRIMARY KEY, oid TEXT, state TEXT); '
|
||||
"INSERT INTO document VALUES ('old'); "
|
||||
"INSERT INTO pending VALUES ('old','old-identity','pending');")
|
||||
|
||||
|
||||
def state(root):
|
||||
with connect(root / 'local.db') as db:
|
||||
assert db.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
|
||||
return {'body': db.execute('SELECT body FROM document').fetchone()[0],
|
||||
'pending': db.execute('SELECT * FROM pending ORDER BY id').fetchall()}
|
||||
|
||||
|
||||
def sha(path):
|
||||
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def experiment(report):
|
||||
with tempfile.TemporaryDirectory(prefix='fep-storage-receiver-') as temp:
|
||||
root = Path(temp)
|
||||
old = {'body': 'old', 'pending': [('old', 'old-identity', 'pending')]}
|
||||
new = {'body': 'new', 'pending': [('new', 'new-identity', 'pending'),
|
||||
('old', 'old-identity', 'superseded')]}
|
||||
for mode, expected in [('before', old), ('after', new), ('middle', old),
|
||||
('split', {'body': 'old', 'pending': [('old', 'old-identity', 'superseded')]})]:
|
||||
case = root / mode
|
||||
initialize(case)
|
||||
code = kill_at(mode, case)
|
||||
observed = state(case)
|
||||
report['checks'].append({'id': f'local-{mode}', 'pass': observed == expected,
|
||||
'negative_control': mode == 'split', 'worker_exit': code,
|
||||
'observed': observed})
|
||||
assert observed == expected
|
||||
case = root / 'receiver'
|
||||
initialize(case)
|
||||
client = case / 'client'
|
||||
receiver = case / 'receiver.git'
|
||||
run('git', 'init', '--quiet', '--initial-branch=main', str(client))
|
||||
run('git', 'init', '--quiet', '--bare', '--initial-branch=main', str(receiver))
|
||||
run('git', 'config', 'receive.denyNonFastForwards', 'true', cwd=receiver)
|
||||
tree = run('git', 'mktree', cwd=client, input='').stdout.strip()
|
||||
|
||||
def commit(parent, text):
|
||||
args = ['git', 'commit-tree', tree]
|
||||
if parent:
|
||||
args += ['-p', parent]
|
||||
return run(*args, cwd=client, input=text + '\n').stdout.strip()
|
||||
|
||||
def push(oid, force=False):
|
||||
return run('git', 'push', '--porcelain', str(receiver),
|
||||
('+' if force else '') + f'{oid}:refs/heads/main', cwd=client, check=False)
|
||||
|
||||
def tip():
|
||||
return run('git', 'rev-parse', 'refs/heads/main', cwd=receiver).stdout.strip()
|
||||
|
||||
base = commit(None, 'base')
|
||||
accepted = commit(base, 'operation A: synthetic intent')
|
||||
assert push(base).returncode == 0
|
||||
with connect(case / 'local.db') as db:
|
||||
db.execute("INSERT INTO pending VALUES ('send',?,'pending')", (accepted,))
|
||||
code = kill_at('send', case)
|
||||
with connect(case / 'local.db') as db:
|
||||
pending = db.execute("SELECT state FROM pending WHERE id='send'").fetchone()[0]
|
||||
report['checks'].append({'id': 'lost-local-ack', 'pass': tip() == accepted and pending == 'pending',
|
||||
'worker_exit': code, 'receiver_tip': tip(), 'outbox_state': pending})
|
||||
assert tip() == accepted and pending == 'pending'
|
||||
replay = push(accepted)
|
||||
count = int(run('git', 'rev-list', '--count', 'main', cwd=receiver).stdout)
|
||||
report['checks'].append({'id': 'same-identity-replay', 'pass': replay.returncode == 0 and count == 2,
|
||||
'exit': replay.returncode, 'history_count': count,
|
||||
'up_to_date': 'up to date' in replay.stdout})
|
||||
assert replay.returncode == 0 and count == 2
|
||||
with connect(case / 'local.db') as db:
|
||||
db.execute("UPDATE pending SET state='acknowledged' WHERE id='send'")
|
||||
concurrent = commit(accepted, 'operation B')
|
||||
assert push(concurrent).returncode == 0
|
||||
stale = commit(base, 'stale operation C')
|
||||
with connect(case / 'local.db') as db:
|
||||
db.execute("INSERT INTO pending VALUES ('stale',?,'pending')", (stale,))
|
||||
rejected = push(stale, force=True) # Force transmission; real server denies it.
|
||||
with connect(case / 'local.db') as db:
|
||||
pending = db.execute("SELECT state FROM pending WHERE id='stale'").fetchone()[0]
|
||||
server_refused = 'non-fast-forward' in rejected.stderr and 'remote rejected' in rejected.stdout
|
||||
report['checks'].append({'id': 'stale-ancestry', 'pass': rejected.returncode != 0 and server_refused and tip() == concurrent and pending == 'pending',
|
||||
'exit': rejected.returncode, 'server_refused': server_refused,
|
||||
'receiver_tip_unchanged': tip() == concurrent, 'outbox_state': pending})
|
||||
assert report['checks'][-1]['pass']
|
||||
run('git', 'gc', '--prune=now', cwd=receiver)
|
||||
retained = run('git', 'cat-file', '-t', accepted, cwd=receiver).stdout.strip()
|
||||
ancestor = run('git', 'merge-base', '--is-ancestor', accepted, 'main', cwd=receiver, check=False)
|
||||
report['checks'].append({'id': 'reachable-identity-after-gc', 'pass': retained == 'commit' and ancestor.returncode == 0})
|
||||
assert report['checks'][-1]['pass']
|
||||
# The same business intent under a NEW identity is another accepted commit.
|
||||
fresh = commit(concurrent, 'operation A: synthetic intent')
|
||||
assert push(fresh).returncode == 0
|
||||
count = int(run('git', 'rev-list', '--count', 'main', cwd=receiver).stdout)
|
||||
report['checks'].append({'id': 'fresh-identity-duplicates-intent', 'negative_control': True,
|
||||
'pass': fresh != accepted and count == 4, 'history_count': count})
|
||||
assert report['checks'][-1]['pass']
|
||||
report['identities'] = {'base': base, 'accepted': accepted, 'concurrent': concurrent,
|
||||
'stale': stale, 'fresh': fresh}
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument('--worker', choices=['before', 'after', 'middle', 'split', 'send'])
|
||||
parser.add_argument('--root', type=Path)
|
||||
parser.add_argument('--output', type=Path)
|
||||
args = parser.parse_args()
|
||||
if args.worker:
|
||||
worker(args.worker, args.root)
|
||||
return
|
||||
if not args.output:
|
||||
parser.error('--output is required (use a new file to preserve prior receipts)')
|
||||
# Reserve the receipt before execution and refuse to overwrite retained evidence.
|
||||
with args.output.open('x') as output:
|
||||
start = time.monotonic()
|
||||
import _sqlite3
|
||||
report = {'started_utc': datetime.now(timezone.utc).isoformat(), 'checks': [],
|
||||
'provenance': {'python': sys.version.split()[0], 'sqlite': sqlite3.sqlite_version,
|
||||
'sqlite_source_id': sqlite3.connect(':memory:').execute('SELECT sqlite_source_id()').fetchone()[0],
|
||||
'git': run('git', '--version').stdout.strip(),
|
||||
'git_binary_sha256': sha(shutil.which('git')),
|
||||
'sqlite_python_extension_sha256': sha(_sqlite3.__file__),
|
||||
'experiment_sha256': sha(HERE)},
|
||||
'settings': {'sqlite_journal': 'wal', 'sqlite_synchronous': 'FULL',
|
||||
'receiver': 'local bare git receive-pack', 'denyNonFastForwards': True},
|
||||
'scope': 'Process kills and discarded local acknowledgement; no power-loss or network-loss simulation.'}
|
||||
try:
|
||||
experiment(report)
|
||||
report['status'] = 'passed'
|
||||
except Exception as exc:
|
||||
report['status'] = 'failed'
|
||||
report['error'] = f'{type(exc).__name__}: {exc}'
|
||||
raise
|
||||
finally:
|
||||
report['execution_seconds'] = round(time.monotonic() - start, 6)
|
||||
json.dump(report, output, indent=2)
|
||||
output.write('\n')
|
||||
print(f"{len(report['checks'])} checks: {report['status']}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue