engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

mesh: sync default-holds records younger than 1 hour (kindergarten window)

Browse source 
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.
This commit is contained in:
russell@unturf.com 2026-04-29 16:07:24 -04:00
parent e3cbe3bb37
commit b48646bc0c
No known key found for this signature in database
2 changed files with 157 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

97
aborist/cli.py
View file

@ -533,6 +533,12 @@ def _render_inspect_human(result: dict) -> str:
if diag == "trailing_artifact":
lines.append(f" matched_prefix_chars: {d.get('matched_prefix_chars')}")
lines.append(f" trailing_artifact: {_short(d.get('trailing_artifact', ''), 100)}")
elif diag == "interior_elision":
lines.append(
f" matched: {d.get('matched_prefix_chars')} prefix + "
f"{d.get('matched_suffix_chars')} suffix chars (parenthetical aside dropped)"
)
lines.append(f" dropped_aside: {_short(d.get('dropped_aside', ''), 120)}")
elif diag in ("paraphrase", "partial_paraphrase"):
lines.append(f" token_coverage: {d.get('token_coverage')}")
counts = d.get("token_counts", {})
@ -2127,22 +2133,37 @@ def _cmd_mesh_sync(args: argparse.Namespace) -> int:
Two pushes happen by default (unless ``--no-roots`` /
``--no-falsifications`` opts one out):
1. **ANNOUNCE_ROOT** for the most-recent ``--limit`` documents.
Receivers that already have the root no-op (idempotent on the
receiver via dedup against ``documents.document_root``).
1. **ANNOUNCE_ROOT** for the most-recent ``--limit`` documents
older than the kindergarten window. Receivers dedup by
``documents.document_root``.
2. **ANNOUNCE_FALSIFICATION** for the most-recent ``--limit``
falsifications. Per the protocol, this is the audit-preserving
channel for "this answer is wrong" — burns are deliberately
NOT propagated (kindergarten-only local cleanup).
falsifications older than the kindergarten window. Burns
deliberately NOT propagated — local kindergarten cleanup.
**Kindergarten window.** Records younger than
``--kindergarten-seconds`` (default 3600 = 1 hour) are NOT
broadcast. Gives the operator time to inspect a fresh ingest or
falsification & burn it before the network sees it. Override per
invocation; ``--kindergarten-seconds 0`` broadcasts everything
(cron-friendly opt-out for operators who prefer immediate
propagation). The window is sender-side discipline; receivers
don't enforce it because they have no view into when the sender
created the record.
Receivers verify the Ed25519 signature, run per-peer chain-of-
claims fork detection, then write one ``mesh_received`` audit
event per accepted envelope. Duplicate broadcasts produce
duplicate audit-log entries on the receiver but no state corruption.
duplicate audit-log entries on the receiver but no state
corruption.
"""
import time as _time
from aborist.mesh import is_enabled, load_identity
from aborist.mesh.wire import MeshWireClient
now_ts = int(_time.time())
cutoff_ts = now_ts - max(0, args.kindergarten_seconds)
conn = connect(args.db)
try:
if load_identity(conn) is None:
@ -2151,28 +2172,47 @@ def _cmd_mesh_sync(args: argparse.Namespace) -> int:
if not is_enabled(conn):
print("error: mesh.enabled is off", file=sys.stderr)
return 2
root_rows = (
[]
if args.no_roots
else conn.execute(
# Total counts inform skipped-by-kindergarten reporting.
total_roots = 0
total_falsifications = 0
root_rows: list = []
falsification_rows: list = []
if not args.no_roots:
total_roots = conn.execute(
"SELECT COUNT(*) FROM documents"
).fetchone()[0]
root_rows = conn.execute(
"SELECT document_root, document_uri, chunking_version, "
" canonicalization_version, schema_version "
"FROM documents ORDER BY rowid DESC LIMIT ?",
(args.limit,),
"FROM documents WHERE ingest_ts <= ? "
"ORDER BY rowid DESC LIMIT ?",
(cutoff_ts, args.limit),
).fetchall()
)
falsification_rows = (
[]
if args.no_falsifications
else conn.execute(
if not args.no_falsifications:
total_falsifications = conn.execute(
"SELECT COUNT(*) FROM falsifications"
).fetchone()[0]
falsification_rows = conn.execute(
"SELECT cache_key, reason FROM falsifications "
"WHERE at <= ? "
"ORDER BY at DESC LIMIT ?",
(args.limit,),
(cutoff_ts, args.limit),
).fetchall()
)
finally:
conn.close()
# Skipped-by-kindergarten = (rows younger than cutoff that would have
# been in the most-recent --limit) — approximated by total minus what
# we pulled, capped at limit.
fresh_roots_held = max(
0,
min(total_roots, args.limit) - len(root_rows),
) if not args.no_roots else 0
fresh_falsifications_held = max(
0,
min(total_falsifications, args.limit) - len(falsification_rows),
) if not args.no_falsifications else 0
sent_roots: list[dict] = []
sent_falsifications: list[dict] = []
errors: list[dict] = []
@ -2209,8 +2249,11 @@ def _cmd_mesh_sync(args: argparse.Namespace) -> int:
"peer": args.peer,
"peer_member_id": peer_info.get("member_id"),
"peer_epoch": peer_info.get("current_epoch"),
"kindergarten_seconds": args.kindergarten_seconds,
"announced_roots": len(sent_roots),
"announced_falsifications": len(sent_falsifications),
"kindergarten_held_roots": fresh_roots_held,
"kindergarten_held_falsifications": fresh_falsifications_held,
"errors": len(errors),
"sent_roots": sent_roots[: args.verbose],
"sent_falsifications": sent_falsifications[: args.verbose],
@ -2675,7 +2718,8 @@ def build_parser() -> argparse.ArgumentParser:
help=(
"sidecar diagnostic for a providence_cache record — pulls "
"source chunks and classifies each unverified span "
"(paraphrase / trailing_artifact / no_overlap). Read-only, "
"(paraphrase / trailing_artifact / interior_elision / "
"no_overlap). Read-only, "
"no audit events, no v9.8 field changes."
),
)
@ -3002,6 +3046,17 @@ def build_parser() -> argparse.ArgumentParser:
action="store_true",
help="skip ANNOUNCE_FALSIFICATION broadcast (only push roots)",
)
mesh_sync.add_argument(
"--kindergarten-seconds",
dest="kindergarten_seconds",
type=int,
default=3600,
help=(
"hold records younger than this many seconds back from the "
"broadcast (default: 3600 = 1 hour). Gives operators time to "
"burn or falsify before peers see it. 0 = broadcast everything."
),
)
mesh_sync.set_defaults(func=_cmd_mesh_sync)
mesh_pull = mesh_sub.add_parser(

81
tests/test_mesh_cli.py
View file

@ -136,6 +136,7 @@ def test_sync_announces_local_roots(tmp_path, capsys):
"--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
@ -198,6 +199,7 @@ def test_sync_announces_falsifications(tmp_path, capsys):
"--peer", srv.url,
"--limit", "10",
"--no-roots",
"--kindergarten-seconds", "0",
])
rc = args.func(args)
assert rc == 0
@ -259,6 +261,7 @@ def test_sync_no_falsifications_flag_skips_them(tmp_path, capsys):
"--peer", srv.url,
"--limit", "10",
"--no-falsifications",
"--kindergarten-seconds", "0",
])
rc = args.func(args)
assert rc == 0
@ -311,3 +314,81 @@ def test_serve_responds_on_info_when_started_in_thread(tmp_path):
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