check_multiscale_filler_index.py

#!/usr/bin/env python3
"""Independent finite semantic and release-budget audit for multiscale fillers.

Python sets/copies are oracle machinery, not native storage. Unit jobs below
are bounded parser/key macros, not claims about individual machine operations.
"""
from itertools import product
import json
import random


def words(n):
    return [''.join(p) for p in product('01', repeat=n)]


def enc(w):
    return ''.join('0' + x for x in w) + '11'


def parse(owner):
    m = len(owner)
    assert m >= 17 and m % 2 and owner[0] == '1'
    n = (m - 1) // 2
    alpha, beta = owner[1:n + 1], owner[n + 1:]
    v = ''
    for i in range(0, len(beta) - 1, 2):
        pair = beta[i:i + 2]
        if pair == '11':
            assert v
            return n, alpha, v, beta[i + 2:]
        assert pair[0] == '0'
        v += pair[1]
    raise AssertionError('missing delimiter')


def key(owner):
    n, alpha, v, w = parse(owner)
    return '0' * n + '1' + enc(w) + alpha + enc(v)


def scale(m):
    return m.bit_length() - 1


def empty():
    return frozenset(), {}


def update(state, owner, alias):
    literal, buckets = state
    m = len(owner)
    chosen = scale(m)
    out = dict(buckets)
    # Completed windows at every crossed boundary are retained provisionally.
    for j in range(chosen + 1):
        installed, pending = out.get(j, (frozenset(), None))
        if pending:
            installed = installed | {key(pending)}
        out[j] = frozenset(installed), None
    installed, _ = out[chosen]
    out[chosen] = installed, owner if owner[0] == '1' else None
    proposed = literal | {owner}, out
    return state if alias else proposed


def query(state, view, z, support):
    if not support:
        return False
    n, alpha, ell, filler, mapping = view
    literal, buckets = state
    if z[:n - ell] != '0' * (n - ell):
        return '1' + alpha + z in literal
    t = z[-ell:]
    v = ''.join(mapping[t[:j]] for j in range(1, ell + 1))
    installed, pending = buckets.get(scale(2 * n + 1), (frozenset(), None))
    direct_pending = False
    if pending:
        pn, pa, pv, pw = parse(pending)
        direct_pending = pn == n and pa == alpha and pv == v and pw == filler
    return ('0' * n + '1' + enc(filler) + alpha + enc(v) in installed) or direct_pending


def direct_owner(view, z):
    n, alpha, ell, filler, mapping = view
    if z[:n - ell] == '0' * (n - ell):
        t = z[-ell:]
        v = ''.join(mapping[t[:j]] for j in range(1, ell + 1))
        beta = enc(v) + filler
    else:
        beta = z
    return '1' + alpha + beta


def invariant(state, known):
    literal, buckets = state
    for owner in known:
        if owner[0] == '1':
            installed, pending = buckets.get(scale(len(owner)), (frozenset(), None))
            assert (owner in literal) == (key(owner) in installed or owner == pending)


def audit_semantics():
    rng = random.Random(243)
    views, known = [], set()
    for n in (8, 9, 12, 15, 16, 20, 31, 32, 64):
        known.add('0' * (2 * n + 1))
        for ell in (1, max(1, (n - 4) // 4)):
            for _ in range(4):
                alpha = ''.join(rng.choice('01') for _ in range(n))
                filler = ''.join(rng.choice('01') for _ in range(n - 2 * ell - 2))
                # Use sparse sampled prefix-consistent maps without exponential growth.
                samples = ['0' * ell, '1' * ell, ('01' * ell)[:ell]]
                mapping = {}
                for t in samples:
                    for j in range(1, ell + 1):
                        mapping.setdefault(t[:j], rng.choice('01'))
                view = n, alpha, ell, filler, mapping
                zs = ['0' * (n - ell) + t for t in samples]
                beta = enc('1') + '1' * (n - 4)
                zs.append(beta)  # ordinary branch, with valid parsed foreign owner
                views.append((view, zs))
                known.update(direct_owner(view, z) for z in zs)
    states = {i: empty() for i in range(4)}
    history = []
    reads = snapshots = 0
    owners = sorted(known)
    for step in range(360):
        source = rng.randrange(4)
        b = rng.choice(owners)
        alias = rng.randrange(2)
        old = states[source]
        states[source] = update(old, b, alias)
        if alias:
            assert states[source] == old
        invariant(states[source], known)
        history.append((states[source], frozenset(states[source][0])))
        for state, expected in rng.sample(history, min(3, len(history))):
            assert state[0] == expected
            invariant(state, known)
            snapshots += 1
            for view, zs in rng.sample(views, 6):
                for z in zs:
                    for support in (False, True):
                        assert query(state, view, z, support) == (support and direct_owner(view, z) in expected)
                        reads += 1
    return len(known), snapshots, reads


def audit_service():
    windows = reads = 0
    for j in range(4, 15):
        old_max = 2 ** (j + 2) - 1
        # Next bucket j+1 has length <2^(j+2). A complete parsing pass m'
        # plus key insertion <=2m', and eight fixed setup jobs, are paid.
        total = 3 * old_max + 8
        remaining = total
        for _ in range(2 ** j):
            remaining = max(0, remaining - 32)
        assert remaining == 0
        windows += 1
    for n in range(8, 513):
        m, j = 2 * n + 1, scale(2 * n + 1)
        pending_max = 2 ** (j + 1) - 1
        assert pending_max < 2 * m
        assert pending_max + 8 <= 8 * n
        # Query follows exactly one directory edge per boundary; endpoint has
        # bucket j without searching a numeric address.
        cursor = 0
        for i in range(1, m + 1):
            if i > 1 and i & (i - 1) == 0:
                cursor += 1
        assert cursor == j
        for ell in range(1, (n - 4) // 4 + 1):
            wlen = n - 2 * ell - 2
            prefix = n + 1 + 2 * wlen + 2
            assert prefix == 3 * n - 4 * ell - 1
            assert prefix <= 4 * (n - 1)  # tolerate first-symbol setup delay
            backlog = n + 2
            for _ in range(n - ell):
                backlog = max(0, backlog - 4)
            assert backlog == 0
            assert wlen <= 4 * (n - ell)
            for k in range(ell):
                backlog += 4 if k == ell - 1 else 2
                backlog = max(0, backlog - 4)
                assert backlog == 0
            reads += 1
    return windows, reads


def audit_keys():
    seen = {}
    count = 0
    # Exhaust all valid beta for small local n; includes multiple v lengths,
    # multiple n within each bucket, and empty/arbitrary fillers.
    for n in range(8, 13):
        alpha = '0' * n
        for beta in words(n):
            owner = '1' + alpha + beta
            try:
                parsed = parse(owner)
            except AssertionError:
                continue
            k = key(owner)
            assert k not in seen or seen[k] == owner
            seen[k] = owner
            assert len(k) == 4 * n - 2 * len(parsed[2]) + 1 <= 2 * len(owner)
            count += 1
    # Explicit same-class unequal-width pending must never match.
    state = update(empty(), '1' + '0' * 9 + enc('0') + '0' * 5, 0)
    view = 8, '0' * 8, 1, '0' * 4, {'0': '0'}
    assert not query(state, view, '0' * 8, True)
    return count


def main():
    known, snapshots, semantic_reads = audit_semantics()
    windows, schedules = audit_service()
    print(json.dumps({'seed': 243, 'known_owners': known, 'updates': 360,
                      'immutable_snapshot_checks': snapshots, 'semantic_reads': semantic_reads,
                      'installer_window_cases': windows, 'query_deadline_cases': schedules,
                      'injectivity_cases': audit_keys(), 'status': 'passed'}, indent=2))


if __name__ == '__main__':
    main()