diff --git a/Makefile b/Makefile index 816b842..493b488 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ # SourceOS Continuum — lifecycle entry points. # Control-plane targets delegate to Makefile.porter (the rehomed Porter control plane). -.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp spine run loop verify +.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp spine run loop verify lease validate: ## repo hygiene + CapD validity python3 tools/validate.py @@ -39,6 +39,9 @@ loop: ## autonomous control loop demo: sense -> governed spine -> act, once per verify: ## volunteer-mesh verification demo: redundant quorum over untrusted worker results cd tools && python3 work_unit.py +lease: ## pull/lease scheduler demo: workers pull WUs, crash-stop re-lending, ordered re-merge + cd tools && python3 lease_scheduler.py + onboard: ## bring up a workstation: local sovereign forge + local cluster + sourceosctl @echo "[continuum] onboard — scaffold: wires Gitea bring-up + kind/k3s + sourceos-devtools/sourceosctl" diff --git a/capd/volunteer-mesh-verification.mesh.capd.json b/capd/volunteer-mesh-verification.mesh.capd.json index f24080e..5624359 100644 --- a/capd/volunteer-mesh-verification.mesh.capd.json +++ b/capd/volunteer-mesh-verification.mesh.capd.json @@ -6,6 +6,7 @@ "description": "How a Folding@home-scale volunteer grid (hundreds of thousands of anonymous, churny, possibly-malicious workers) becomes trustworthy. A Grant proves WHO ran a Work Unit; it cannot prove the RESULT is correct. This plane verifies results fail-closed: a Work Unit is run redundantly on N independent workers and only a result a quorum agrees on (identical output digest) is accepted; stochastic tasks use spot-check canaries with known answers; reliable backends (a cluster, per CluBORun) can stand in as reference verifiers. Reputation is a per-worker verified-success record that weights allocation and routes around bad actors. Implements the Dual-Orchestration PROOF_MODE (redundant | spot_check | tee | zk | optimistic).", "links": { "engine": "tools/work_unit.py", + "scheduler": "tools/lease_scheduler.py", "executor": "tools/executor.py", "placement": "tools/compute_plane.py", "grant_authority": "tools/mcp_a2a_grant.py", diff --git a/docs/VOLUNTEER_MESH.md b/docs/VOLUNTEER_MESH.md index bccd1b9..e587ae9 100644 --- a/docs/VOLUNTEER_MESH.md +++ b/docs/VOLUNTEER_MESH.md @@ -90,7 +90,7 @@ results back. | Governed dispatch of a WU | `executor` adapters (local/k8s/slurm/wasm/descriptor/connector) | ✅ | | Identity/attestation (move 4) | `mcp_a2a_grant` (SPIFFE binding + TPM/cosign attestation) | ✅ | | Quotas/admission | `admission.py` (additive; make credits additive-only per move 4) | ✅ | -| Pull/lease streaming dispatch (move 2) | StreamLender + Limiter; heartbeat liveness (`mesh_telemetry` is the base) | ○ next | +| **Pull/lease streaming dispatch (move 2)** | `lease_scheduler.py` — StreamLender + Limiter: workers pull/lease when idle (adaptive, no speed estimation), conservative single-copy, crash-stop re-lending by index, ordered re-merge, heartbeat liveness | ✅ (this change) | | Split data plane + DataBridge (move 3) | signed-URL nearest-endpoint upload + async replication + verify-download barrier | ○ next | | Instrumented signals + refusal (move 5) | wrap reputation/confidence as `{value,uncertainty,provenance,validity,refusal}` | ○ next | | Domain-scoped reputation (move 7) | per-`(worker,domain)` card on `work_unit.Reputation` | ○ next | diff --git a/tools/lease_scheduler.py b/tools/lease_scheduler.py new file mode 100644 index 0000000..5826dd9 --- /dev/null +++ b/tools/lease_scheduler.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +"""Pull/lease streaming dispatch — Pando's StreamLender + Limiter, governed. + +Push WU-assignment is brittle on a churny volunteer mesh: you'd estimate each worker's speed and +re-push on every crash. The elegant alternative is PULL: a worker LEASES work when it is idle, so +faster/idler devices simply pull more (adaptive, no speed estimation) and slow ones pull less +(automatic backpressure). Properties: + + * conservative — one copy of a Work Unit to at most one worker at a time, maximizing DISTINCT + units in flight (redundant-quorum is an optional overlay layered on top, not the default). + * crash-stop work-stealing — a worker that misses its lease deadline (crashed/left) has its + borrowed unit RE-LENT by index to someone else; nothing is lost. + * ordered — results re-merge by index for determinism. + * bounded — a Limiter caps in-flight leases per worker. + +Heartbeat extends a lease's deadline (progress signal); silence past the deadline reclaims it. +""" +from __future__ import annotations + +import time + + +class LeaseScheduler: + def __init__(self, work_units, *, max_in_flight: int = 4, lease_ttl_s: float = 30.0, + clock=time.time): + self._pending: list[tuple[int, object]] = list(enumerate(work_units)) # ordered by index + self._leased: dict[str, dict] = {} + self._done: dict[int, object] = {} + self._inflight: dict[str, set] = {} + self._max = int(max_in_flight) + self._ttl = float(lease_ttl_s) + self._clock = clock + self._seq = 0 + + def _reclaim(self, now: float) -> list[str]: + """Return WUs whose lease deadline passed (worker crashed/left) to the pool — re-lent by + index. A unit already completed is not re-lent.""" + expired = [lid for lid, l in self._leased.items() if now > l["deadline"]] + for lid in expired: + l = self._leased.pop(lid) + self._inflight.get(l["worker"], set()).discard(lid) + if l["index"] not in self._done: + self._pending.append((l["index"], l["wu"])) + if expired: + self._pending.sort(key=lambda t: t[0]) # keep ordered for deterministic re-merge + return expired + + def lease(self, worker: str, n: int = 1) -> list[dict]: + """A worker pulls up to n Work Units (idle workers call this more → adaptive). Respects the + per-worker Limiter.""" + now = self._clock() + self._reclaim(now) + held = self._inflight.setdefault(worker, set()) + out = [] + while self._pending and len(held) < self._max and len(out) < n: + index, wu = self._pending.pop(0) + self._seq += 1 + lid = f"lease-{self._seq}" + self._leased[lid] = {"index": index, "wu": wu, "worker": worker, "deadline": now + self._ttl} + held.add(lid) + out.append({"lease_id": lid, "index": index, "wu": wu}) + return out + + def complete(self, lease_id: str, output) -> bool: + """Return a result. False if the lease was already reclaimed (a slow/crashed worker whose + unit was re-lent) — its late result is dropped, the re-lent copy is authoritative.""" + l = self._leased.pop(lease_id, None) + if l is None: + return False + self._inflight.get(l["worker"], set()).discard(lease_id) + self._done.setdefault(l["index"], output) + return True + + def heartbeat(self, worker: str, lease_id: str | None = None) -> None: + """Extend the deadline of a worker's lease(s) — a progress signal that keeps it from being + reclaimed.""" + now = self._clock() + for lid, l in self._leased.items(): + if l["worker"] == worker and (lease_id is None or lid == lease_id): + l["deadline"] = now + self._ttl + + def progress(self) -> dict: + return {"pending": len(self._pending), "leased": len(self._leased), "done": len(self._done)} + + def drained(self) -> bool: + self._reclaim(self._clock()) + return not self._pending and not self._leased + + def results_in_order(self) -> list: + return [self._done[i] for i in sorted(self._done)] + + +if __name__ == "__main__": + import json + + class Clock: + def __init__(self, t=0.0): + self.t = t + + def __call__(self): + return self.t + + clk = Clock() + sched = LeaseScheduler([f"wu{i}" for i in range(5)], max_in_flight=2, lease_ttl_s=10, clock=clk) + # fast worker A pulls + completes; worker B leases one then "crashes" (never completes). + a1 = sched.lease("A", 1)[0] + b1 = sched.lease("B", 1)[0] + sched.complete(a1["lease_id"], a1["wu"] + "-done-by-A") + clk.t = 20 # B missed its deadline -> its unit is re-lent + a2 = sched.lease("A", 1) # picks up B's abandoned unit + more + for lz in a2: + sched.complete(lz["lease_id"], lz["wu"] + "-done-by-A") + while not sched.drained(): + for lz in sched.lease("A", 4): + sched.complete(lz["lease_id"], lz["wu"] + "-done-by-A") + print(json.dumps({"drained": sched.drained(), "results": sched.results_in_order()}, indent=2)) diff --git a/tools/test_lease_scheduler.py b/tools/test_lease_scheduler.py new file mode 100644 index 0000000..b075dfa --- /dev/null +++ b/tools/test_lease_scheduler.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +"""Tests for the pull/lease scheduler. Load-bearing: conservative single-copy dispatch (one WU to +one worker), crash-stop re-lending (a dead worker's unit is not lost), ordered re-merge, and the +per-worker Limiter. Time is injected so crash detection is deterministic.""" +import lease_scheduler as ls + + +class Clock: + def __init__(self, t=0.0): + self.t = t + + def __call__(self): + return self.t + + +def test_limiter_caps_in_flight_per_worker(): + s = ls.LeaseScheduler(range(10), max_in_flight=3, clock=Clock()) + assert len(s.lease("w", n=10)) == 3 + + +def test_conservative_single_copy_no_double_lease(): + s = ls.LeaseScheduler(["a", "b"], max_in_flight=4, clock=Clock()) + w1 = s.lease("w1", 2) + assert len(w1) == 2 and s.lease("w2", 2) == [] # both held by w1; nothing for w2 + + +def test_results_re_merge_in_index_order(): + s = ls.LeaseScheduler(["a", "b", "c"], max_in_flight=4, clock=Clock()) + got = s.lease("w", 3) + s.complete(got[2]["lease_id"], "C") # complete out of order + s.complete(got[0]["lease_id"], "A") + s.complete(got[1]["lease_id"], "B") + assert s.results_in_order() == ["A", "B", "C"] + + +def test_crashed_worker_unit_is_relent_and_late_result_dropped(): + clk = Clock() + s = ls.LeaseScheduler(["only"], max_in_flight=4, lease_ttl_s=10, clock=clk) + a = s.lease("A", 1)[0] + assert s.lease("B", 1) == [] # conservative: B gets nothing while A holds it + clk.t = 20 # A missed its deadline -> crashed + b = s.lease("B", 1) + assert len(b) == 1 and b[0]["wu"] == "only" # re-lent to B + assert s.complete(a["lease_id"], "stale") is False # A's late result is dropped + assert s.complete(b[0]["lease_id"], "fresh") is True + assert s.results_in_order() == ["fresh"] # the unit ran exactly once, authoritative copy + + +def test_heartbeat_prevents_reclaim(): + clk = Clock() + s = ls.LeaseScheduler(["x"], lease_ttl_s=10, clock=clk) + s.lease("A", 1) + clk.t = 8 + s.heartbeat("A") # extends deadline to 18 + clk.t = 15 + assert s.lease("B", 1) == [] # still alive; not reclaimed + + +def test_adaptive_a_fast_worker_drains_more(): + s = ls.LeaseScheduler(range(6), max_in_flight=2, clock=Clock()) + a_total = 0 + while not s.drained(): + got = s.lease("A", 2) # A keeps pulling; B never does + for lz in got: + s.complete(lz["lease_id"], lz["index"]) + a_total += 1 + assert a_total == 6 + + +def test_end_to_end_two_workers_one_crash_each_wu_once(): + clk = Clock() + s = ls.LeaseScheduler([f"wu{i}" for i in range(4)], max_in_flight=1, lease_ttl_s=10, clock=clk) + a = s.lease("A", 1)[0] + s.lease("B", 1) # B takes one then crashes (never completes) + s.complete(a["lease_id"], a["wu"]) + clk.t = 20 # B reclaimed + while not s.drained(): + for lz in s.lease("A", 4): + s.complete(lz["lease_id"], lz["wu"]) + assert sorted(s.results_in_order()) == ["wu0", "wu1", "wu2", "wu3"] # every WU exactly once + + +if __name__ == "__main__": + import sys + fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for fn in fns: + fn() + print(f"ok: {len(fns)} lease-scheduler tests passed") + sys.exit(0) diff --git a/tools/validate.py b/tools/validate.py index 16f1803..ee88cb6 100644 --- a/tools/validate.py +++ b/tools/validate.py @@ -37,6 +37,7 @@ "tools/control_loop.py", "tools/devspace.py", "tools/work_unit.py", + "tools/lease_scheduler.py", ] CAPD_KEYS = ("capability_id", "kind", "status", "links", "composes_with", "policy") # Every CapD in capd/ must carry the core keys and parse — not just the flagship control-plane one. diff --git a/tools/work_unit.py b/tools/work_unit.py index 7d0c04e..d961241 100644 --- a/tools/work_unit.py +++ b/tools/work_unit.py @@ -16,6 +16,9 @@ Reputation is a per-worker moving record of verified successes; it weights future allocation and is how the mesh routes around bad actors without trusting any single node. + +Dispatch of Work Units to workers is pull/lease (see `lease_scheduler.py`): verification here is the +optional overlay layered on top, keyed on stakes x reputation. """ from __future__ import annotations