universe/k8s/lux-mainnet/rpc-tip-gate-test.py

"""Proof for the rpc-tip-gate decision logic.

Extracts tip_gate.py out of the values file CD deploys, so the script that
actually ships is the script under test -- not a copy that can drift from it.

    python3 k8s/lux-mainnet/rpc-tip-gate-test.py

It reads deploy/lux-mainnet/rpc-tip-gate.yaml, NOT the manifest next to this
file. That is where the script lives now, because that is what cd.hanzo.ai
renders: the Deployment and the ConfigMap it cannot start without are one
chart release instead of two planes. It used to read the manifest, the code
was not there, and it asserted `tip_gate.py not found` before running a single
case -- so this file proved nothing at all while looking like it did.
"""
import hashlib
import os
import pathlib
import sys
import types

import yaml

HERE = pathlib.Path(__file__).resolve().parent
VALUES = HERE.parents[1] / "deploy" / "lux-mainnet" / "rpc-tip-gate.yaml"
SHA_ANNOTATION = "lux.network/tip-gate-script-sha256"


def load_gate():
    """Import the tip_gate.py the chart ships, and prove the roll-trigger matches it."""
    values = yaml.safe_load(VALUES.read_text())
    code = None
    for cm in values.get("configMaps") or []:
        if "tip_gate.py" in (cm.get("data") or {}):
            code = cm["data"]["tip_gate.py"]
    assert code, "tip_gate.py not found in %s" % VALUES

    # A ConfigMap change does NOT restart the pods that mount it, and this
    # script is read once at boot, so podAnnotations carries its sha256 to move
    # the pod template with the content. If the two disagree, editing the script
    # would update the file and leave the running pod on the old one -- silently.
    want = hashlib.sha256(code.encode()).hexdigest()[:16]
    got = (values.get("podAnnotations") or {}).get(SHA_ANNOTATION)
    assert got == want, (
        "podAnnotations[%s] is %r but the script hashes to %r -- bump it, or the "
        "workload will not roll when the script changes" % (SHA_ANNOTATION, got, want)
    )

    os.environ.update(
        NAMESPACE="lux-mainnet",
        POD_SELECTOR="app.kubernetes.io/name=luxd",
        RPC_PORT="9630",
        RPC_PATH="/v1/chain/c/rpc",
        LABEL_KEY="lux.network/rpc",
        LABEL_VALUE="fresh",
        MAX_LAG_BLOCKS="1",
        INTERVAL_SECONDS="10",
        PROBE_TIMEOUT_SECONDS="5",
    )
    mod = types.ModuleType("tip_gate")
    mod.__file__ = "tip_gate.py"
    exec(compile(code, "tip_gate.py", "exec"), mod.__dict__)
    return mod


GATE = load_gate()
FRESH = {"lux.network/rpc": "fresh"}


def run(fleet):
    """Drive one reconcile() over a synthetic fleet.

    fleet: name -> (tip or None, currently_labelled, phase)
    Returns (serving_set, patches) where patches is name -> label value/None.
    """
    pods = [
        {
            "metadata": {"name": n, "labels": dict(FRESH) if lab else {}},
            "status": {"podIP": "10.0.0.%d" % (i + 1), "phase": phase},
        }
        for i, (n, (_, lab, phase)) in enumerate(fleet.items())
    ]
    patches = {}

    def fake_kube(method, path, body=None):
        if method == "GET":
            return {"items": pods}
        name = path.rsplit("/", 1)[1]
        patches[name] = body["metadata"]["labels"]["lux.network/rpc"]
        return {}

    ip_to_name = {p["status"]["podIP"]: p["metadata"]["name"] for p in pods}
    GATE.kube = fake_kube
    GATE.tip = lambda ip: fleet[ip_to_name[ip]][0]
    GATE.reconcile()

    serving = set()
    for name, (_, lab, _) in fleet.items():
        after = patches[name] == "fresh" if name in patches else lab
        if after:
            serving.add(name)
    return serving, patches


FAILS = []


def check(label, got, want):
    ok = got == want
    print("%-4s %-58s got=%s" % ("PASS" if ok else "FAIL", label, sorted(got) if isinstance(got, set) else got))
    if not ok:
        FAILS.append("%s: got %r want %r" % (label, got, want))


R = "Running"

# 1. The live mainnet fleet, exactly as measured 2026-07-27: luxd-1 at
#    1098193, luxd-0/3/4 two behind at 1098191, luxd-2 404ing the C-Chain
#    while 1/1 Running. Only the leader may serve.
serving, patches = run({
    "luxd-0": (1098191, False, R),
    "luxd-1": (1098193, False, R),
    "luxd-2": (None, False, R),
    "luxd-3": (1098191, False, R),
    "luxd-4": (1098191, True, R),   # the stale pod that was taking 2/3 of traffic
})
check("live fleet: only the leader serves", serving, {"luxd-1"})
# (present, value) rather than .get(), so "patched to remove" cannot be
# confused with "no patch issued" -- both would read back as None.
check("live fleet: stale luxd-4 is evicted", ("luxd-4" in patches, patches.get("luxd-4")), (True, None))
check("live fleet: leader luxd-1 is admitted", ("luxd-1" in patches, patches.get("luxd-1")), (True, "fresh"))
check("live fleet: 404ing luxd-2 never enters", "luxd-2" in patches, False)

# 2. Leader dies -> the rest become the new max and rejoin on their own.
#    This is what stops the pool from collapsing to zero.
serving, _ = run({
    "luxd-0": (1098191, False, R),
    "luxd-3": (1098191, False, R),
    "luxd-4": (1098191, False, R),
})
check("leader gone: cohort self-heals to full capacity", serving, {"luxd-0", "luxd-3", "luxd-4"})

# 3. A converged fleet serves everything -- the gate must not cost capacity
#    once the chain is healthy.
serving, patches = run({
    "luxd-%d" % i: (1098193, True, R) for i in range(5)
})
check("converged fleet: all five serve", serving, {"luxd-%d" % i for i in range(5)})
check("converged fleet: idempotent, zero patches", patches, {})

# 4. MAX_LAG_BLOCKS=1 tolerates the ordinary one-block spread of a healthy
#    fleet but not the two-block divergence that caused the incident.
serving, _ = run({"a": (100, False, R), "b": (99, False, R), "c": (98, False, R)})
check("lag 1 admitted, lag 2 rejected", serving, {"a", "b"})

# 5. Nobody answers -> change nothing. A blind gate must not empty the pool.
serving, patches = run({
    "luxd-0": (None, True, R),
    "luxd-1": (None, True, R),
})
check("total blackout: no patches issued", patches, {})
check("total blackout: pool left intact", serving, {"luxd-0", "luxd-1"})

# 6. A pod that is not Running is never probed and never serves.
serving, _ = run({
    "luxd-0": (1098193, False, R),
    "luxd-1": (1098193, True, "Pending"),
})
check("non-Running pod excluded", serving, {"luxd-0"})

print()
if FAILS:
    print("%d FAILED" % len(FAILS))
    for f in FAILS:
        print("  " + f)
    sys.exit(1)
print("all checks passed")