russell@unturf.com
b48646bc0c
Fox 2026-04-29: "part of the gossip protocol should be some default
delay that allows a user to catch and burn kindergarteners before
they are synced." Currently `mesh sync` enumerates the most-recent
N items with no age filter — a doc ingested 30 seconds ago goes
out on the next sync, & if a peer ingests it before the operator
notices a problem, burn no longer suffices (peer has its own copy).
Adds a sender-side kindergarten window:
--kindergarten-seconds N default 3600 (1 hour)
Records younger than `now - N` are held back from broadcast. Both
ANNOUNCE_ROOT (filtered on documents.ingest_ts) & ANNOUNCE_FALSIFICATION
(filtered on falsifications.at). N=0 = broadcast everything (cron-
friendly opt-out for operators preferring immediate propagation).
Result JSON now reports the held counts so the operator can see
what the window protected:
kindergarten_seconds: 3600
announced_roots: 12
kindergarten_held_roots: 3
announced_falsifications: 0
kindergarten_held_falsifications: 1
Sender-side discipline only — the receiver has no view into when
the sender created the record, so it can't enforce. Adding a
created_at on the envelope would let receivers reject too-fresh
gossip, but that's a future protocol bump (envelopes today don't
carry sender wall-clock; ts is the send time).
Tests:
- existing tests now pass `--kindergarten-seconds 0` so freshly-
ingested fixtures broadcast immediately for the test
- new test_sync_kindergarten_holds_fresh_records: a 30-second-old
doc is held; an artificially-aged doc broadcasts. announced=1,
held>=1.
- new test_sync_kindergarten_zero_broadcasts_everything: explicit
opt-out works.
342 passed, 1 skipped.
394 lines
13 KiB
Python
394 lines
13 KiB
Python
"""CLI integration: `aborist mesh serve` and `aborist mesh sync`.
|
|
|
|
Wire-layer protocol behavior is covered in test_mesh_wire.py and
|
|
test_mesh_wire_e2e.py — these tests just exercise the argparse plumbing,
|
|
the gating checks (mesh initialized + enabled), and the round-trip
|
|
through MeshWireClient that the sync verb performs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from aborist.cli import build_parser
|
|
from aborist.document import Document
|
|
from aborist.ingest import ingest_source
|
|
from aborist.mesh import init_identity, set_enabled
|
|
from aborist.mesh.members import add_member
|
|
from aborist.mesh.wire import MeshWireServer
|
|
from aborist.store import connect
|
|
|
|
|
|
class _OneDocSource:
|
|
source_type = "mesh_cli_test"
|
|
|
|
def __init__(self, uri: str, content: str, title: str):
|
|
self._doc = Document(
|
|
uri=uri, content=content, source_type=self.source_type, title=title
|
|
)
|
|
|
|
def iter_documents(self):
|
|
yield self._doc
|
|
|
|
|
|
def _ingest(db_path, *, uri: str, content: str, title: str) -> str:
|
|
conn = connect(db_path)
|
|
try:
|
|
ingest_source(conn, _OneDocSource(uri, content, title))
|
|
row = conn.execute(
|
|
"SELECT document_root FROM documents WHERE document_uri=?", (uri,)
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
return row["document_root"]
|
|
|
|
|
|
def _bootstrap_two_peers(tmp_path: Path):
|
|
alice_db = tmp_path / "alice.db"
|
|
bob_db = tmp_path / "bob.db"
|
|
a_conn = connect(alice_db)
|
|
try:
|
|
alice = init_identity(a_conn, group_name="t", member_id="alice")
|
|
set_enabled(a_conn, True)
|
|
finally:
|
|
a_conn.close()
|
|
b_conn = connect(bob_db)
|
|
try:
|
|
bob = init_identity(b_conn, group_name="t", member_id="bob")
|
|
set_enabled(b_conn, True)
|
|
finally:
|
|
b_conn.close()
|
|
a_conn = connect(alice_db)
|
|
try:
|
|
add_member(a_conn, member_id="bob", sign_pub=bob.sign_pub, dh_pub=bob.dh_pub)
|
|
finally:
|
|
a_conn.close()
|
|
b_conn = connect(bob_db)
|
|
try:
|
|
add_member(b_conn, member_id="alice", sign_pub=alice.sign_pub, dh_pub=alice.dh_pub)
|
|
finally:
|
|
b_conn.close()
|
|
return alice_db, bob_db, alice, bob
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gating: mesh must be initialized AND enabled
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_serve_refuses_when_mesh_not_initialized(tmp_path, capsys):
|
|
db = tmp_path / "x.db"
|
|
parser = build_parser()
|
|
args = parser.parse_args(["--db", str(db), "mesh", "serve", "--port", "0"])
|
|
rc = args.func(args)
|
|
assert rc == 2
|
|
assert "not initialized" in capsys.readouterr().err
|
|
|
|
|
|
def test_serve_refuses_when_mesh_disabled(tmp_path, capsys):
|
|
db = tmp_path / "x.db"
|
|
conn = connect(db)
|
|
try:
|
|
init_identity(conn, group_name="t", member_id="alice")
|
|
# Don't call set_enabled — flag stays off by default.
|
|
finally:
|
|
conn.close()
|
|
parser = build_parser()
|
|
args = parser.parse_args(["--db", str(db), "mesh", "serve", "--port", "0"])
|
|
rc = args.func(args)
|
|
assert rc == 2
|
|
assert "enabled" in capsys.readouterr().err
|
|
|
|
|
|
def test_sync_refuses_when_mesh_disabled(tmp_path, capsys):
|
|
db = tmp_path / "x.db"
|
|
conn = connect(db)
|
|
try:
|
|
init_identity(conn, group_name="t", member_id="alice")
|
|
finally:
|
|
conn.close()
|
|
parser = build_parser()
|
|
args = parser.parse_args(["--db", str(db), "mesh", "sync", "--peer", "http://0:0"])
|
|
rc = args.func(args)
|
|
assert rc == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# sync: announces local document_roots to a peer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_sync_announces_local_roots(tmp_path, capsys):
|
|
"""Bob's CLI sync sends ANNOUNCE_ROOT for each local doc to Alice's
|
|
server; Alice's audit chain records each receive."""
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
|
|
# Bob ingests two documents locally — these are what he'll announce.
|
|
_ingest(bob_db, uri="https://example.com/a", content="Doc A content here.", title="A")
|
|
_ingest(bob_db, uri="https://example.com/b", content="Doc B content here.", title="B")
|
|
|
|
srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
parser = build_parser()
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", srv.url,
|
|
"--limit", "100",
|
|
"--kindergarten-seconds", "0", # immediate broadcast for the test
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 0
|
|
finally:
|
|
srv.stop()
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["status"] == "synced"
|
|
assert payload["peer_member_id"] == "alice"
|
|
assert payload["announced_roots"] == 2
|
|
assert payload["announced_falsifications"] == 0 # bob has none
|
|
assert payload["errors"] == 0
|
|
|
|
# Alice's chain gained one mesh_received event per announce.
|
|
a_conn = connect(alice_db)
|
|
try:
|
|
rcvd = a_conn.execute(
|
|
"SELECT COUNT(*) FROM audit_events WHERE event_type='mesh_received'"
|
|
).fetchone()[0]
|
|
finally:
|
|
a_conn.close()
|
|
assert rcvd == 2
|
|
|
|
|
|
def test_sync_announces_falsifications(tmp_path, capsys):
|
|
"""Bob falsifies a local cache record; mesh sync to alice broadcasts
|
|
the falsification. Alice's audit chain gains one mesh_received event
|
|
whose body carries an ANNOUNCE_FALSIFICATION envelope."""
|
|
import time as _time
|
|
from aborist.store import append_audit, transaction
|
|
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
|
|
# Seed a falsification on bob's side.
|
|
b_conn = connect(bob_db)
|
|
try:
|
|
with transaction(b_conn):
|
|
event_hash = append_audit(
|
|
b_conn,
|
|
event_type="falsify",
|
|
subject_root="ab" * 32,
|
|
body={"cache_key": "ab" * 32, "to_state": "failed", "reason": "wrong"},
|
|
)
|
|
b_conn.execute(
|
|
"INSERT INTO falsifications "
|
|
"(cache_key, state, reason, by_actor, at, audit_event_hash) "
|
|
"VALUES (?, 'failed', 'wrong', 'bob', ?, ?)",
|
|
("ab" * 32, int(_time.time()), event_hash),
|
|
)
|
|
finally:
|
|
b_conn.close()
|
|
|
|
srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
parser = build_parser()
|
|
# --no-roots so we isolate the falsification path.
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", srv.url,
|
|
"--limit", "10",
|
|
"--no-roots",
|
|
"--kindergarten-seconds", "0",
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 0
|
|
finally:
|
|
srv.stop()
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["announced_roots"] == 0
|
|
assert payload["announced_falsifications"] == 1
|
|
assert payload["errors"] == 0
|
|
|
|
# Alice's chain gained one mesh_received event with the falsification body.
|
|
import json as _json
|
|
a_conn = connect(alice_db)
|
|
try:
|
|
last = a_conn.execute(
|
|
"SELECT event_type, body FROM audit_events "
|
|
"WHERE event_type = 'mesh_received' "
|
|
"ORDER BY seq DESC LIMIT 1"
|
|
).fetchone()
|
|
finally:
|
|
a_conn.close()
|
|
body = _json.loads(last["body"])
|
|
assert body["wire_type"] == "ANNOUNCE_FALSIFICATION"
|
|
assert body["envelope_body"]["cache_key"] == "ab" * 32
|
|
assert body["envelope_body"]["reason"] == "wrong"
|
|
|
|
|
|
def test_sync_no_falsifications_flag_skips_them(tmp_path, capsys):
|
|
"""--no-falsifications skips the falsification broadcast. Bob's
|
|
falsification is NOT announced. Alice gets only the root broadcasts."""
|
|
import time as _time
|
|
from aborist.store import append_audit, transaction
|
|
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
b_conn = connect(bob_db)
|
|
try:
|
|
with transaction(b_conn):
|
|
event_hash = append_audit(
|
|
b_conn, event_type="falsify",
|
|
subject_root="cd" * 32,
|
|
body={"cache_key": "cd" * 32, "to_state": "failed", "reason": "x"},
|
|
)
|
|
b_conn.execute(
|
|
"INSERT INTO falsifications "
|
|
"(cache_key, state, reason, by_actor, at, audit_event_hash) "
|
|
"VALUES (?, 'failed', 'x', 'bob', ?, ?)",
|
|
("cd" * 32, int(_time.time()), event_hash),
|
|
)
|
|
finally:
|
|
b_conn.close()
|
|
|
|
srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
parser = build_parser()
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", srv.url,
|
|
"--limit", "10",
|
|
"--no-falsifications",
|
|
"--kindergarten-seconds", "0",
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 0
|
|
finally:
|
|
srv.stop()
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["announced_falsifications"] == 0
|
|
|
|
|
|
def test_sync_reports_unreachable_peer(tmp_path, capsys):
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
parser = build_parser()
|
|
# Bind 127.0.0.1:1 — almost certainly nothing listening here.
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", "http://127.0.0.1:1",
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 2
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["status"] == "peer_unreachable"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# serve: smoke-test the server starts and responds via /mesh/info
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_serve_responds_on_info_when_started_in_thread(tmp_path):
|
|
"""We don't actually invoke `mesh serve` (which blocks); we exercise
|
|
MeshWireServer's start_in_thread path the verb relies on."""
|
|
import httpx
|
|
|
|
db = tmp_path / "x.db"
|
|
conn = connect(db)
|
|
try:
|
|
init_identity(conn, group_name="t", member_id="alice")
|
|
set_enabled(conn, True)
|
|
finally:
|
|
conn.close()
|
|
|
|
srv = MeshWireServer(db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
r = httpx.get(srv.url + "/mesh/info", timeout=5.0)
|
|
r.raise_for_status()
|
|
info = r.json()
|
|
finally:
|
|
srv.stop()
|
|
assert info["member_id"] == "alice"
|
|
assert info["group_name"] == "t"
|
|
|
|
|
|
def test_sync_kindergarten_holds_fresh_records(tmp_path, capsys):
|
|
"""Fox 2026-04-29: gossip should default-delay so an operator can
|
|
burn or falsify within a window before peers see fresh records.
|
|
A document ingested 30 seconds ago must NOT broadcast under the
|
|
default 1-hour kindergarten window."""
|
|
import time as _time
|
|
from aborist.store import transaction
|
|
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
_ingest(bob_db, uri="https://example.com/old", content="Old prose.", title="Old")
|
|
_ingest(bob_db, uri="https://example.com/fresh", content="Fresh prose.", title="Fresh")
|
|
|
|
# Backdate the "old" doc so it's >>> kindergarten window. Leave the
|
|
# "fresh" one at its just-set ingest_ts (within the last second).
|
|
long_ago = int(_time.time()) - 7200 # 2 hours ago
|
|
b_conn = connect(bob_db)
|
|
try:
|
|
with transaction(b_conn):
|
|
b_conn.execute(
|
|
"UPDATE documents SET ingest_ts = ? "
|
|
"WHERE document_uri = 'https://example.com/old'",
|
|
(long_ago,),
|
|
)
|
|
finally:
|
|
b_conn.close()
|
|
|
|
srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
parser = build_parser()
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", srv.url,
|
|
"--limit", "10",
|
|
# Default kindergarten = 3600s. Fresh doc held back; old doc
|
|
# broadcast.
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 0
|
|
finally:
|
|
srv.stop()
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["kindergarten_seconds"] == 3600
|
|
assert payload["announced_roots"] == 1 # only "old" broadcast
|
|
assert payload["kindergarten_held_roots"] >= 1 # "fresh" held
|
|
# Confirm exactly which root was sent.
|
|
sent_uris = [r["ack"]["wire_type"] for r in payload["sent_roots"]]
|
|
assert sent_uris == ["ANNOUNCE_ROOT"]
|
|
|
|
|
|
def test_sync_kindergarten_zero_broadcasts_everything(tmp_path, capsys):
|
|
"""--kindergarten-seconds 0 is the explicit opt-out: cron-friendly
|
|
immediate broadcast. Verifies the existing "no kindergarten" tests
|
|
in this file aren't accidentally protected by the default window."""
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
_ingest(bob_db, uri="https://example.com/x", content="Just-ingested.", title="X")
|
|
|
|
srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
parser = build_parser()
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", srv.url,
|
|
"--limit", "10",
|
|
"--kindergarten-seconds", "0",
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 0
|
|
finally:
|
|
srv.stop()
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["announced_roots"] == 1
|
|
assert payload["kindergarten_held_roots"] == 0
|