frontend-patterns/research/experiments/storage-receiver/experiment.py

#!/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()