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

mesh: 'pull' CLI verb — fetch document body from a peer

Browse source 
Closes the request half of the gossip loop. MeshWireClient already
verifies signature + Merkle root on delivered bodies; this verb adds
the CLI surface, gating, local re-ingest through the standard chunker,
and a 'mesh_pulled' audit event. Re-ingest is rejected if the local
document_root differs from the requested root.
This commit is contained in:
russell@unturf.com 2026-04-28 17:33:27 -04:00
parent e868b95530
commit bc67ad80e8
No known key found for this signature in database
2 changed files with 365 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

115
aborist/cli.py
View file

@ -1681,6 +1681,113 @@ def _cmd_mesh_sync(args: argparse.Namespace) -> int:
return 0 if not errors else 1
def _cmd_mesh_pull(args: argparse.Namespace) -> int:
"""Pull a single document body from a peer by document_root.
Closes the request half of the gossip loop. The wire client already
verifies the peer's signature and re-derives the Merkle root from the
delivered leaves before returning. This verb then re-ingests the
delivered text through the standard ingest path so chunking_version /
canonicalization_version stay consistent — and rejects with rc=2 if
the local re-ingest produces a different document_root than requested.
"""
from aborist.document import Document
from aborist.ingest import ingest_source
from aborist.mesh import is_enabled, load_identity
from aborist.mesh.wire import MeshWireClient
conn = connect(args.db)
try:
if load_identity(conn) is None:
print("error: mesh not initialized", file=sys.stderr)
return 2
if not is_enabled(conn):
print("error: mesh.enabled is off", file=sys.stderr)
return 2
already = conn.execute(
"SELECT document_root, document_uri FROM documents WHERE document_root=?",
(args.root,),
).fetchone()
finally:
conn.close()
if already is not None:
print(json.dumps({
"status": "already_present",
"document_root": already["document_root"],
"document_uri": already["document_uri"],
"shard": str(args.db),
}, indent=2))
return 0
try:
with MeshWireClient(args.db, args.peer) as client:
body = client.request_body(root=args.root)
except Exception as e:
print(f"error: pull failed: {e}", file=sys.stderr)
return 2
delivered_uri = body.get("document_uri") or ""
delivered_text = body.get("text") or ""
class _PulledSource:
source_type = "mesh_pull"
def iter_documents(self):
yield Document(
uri=delivered_uri,
content=delivered_text,
source_type="mesh_pull",
title=None,
)
conn = connect(args.db)
try:
ingest_source(conn, _PulledSource())
row = conn.execute(
"SELECT document_root FROM documents WHERE document_root=?",
(args.root,),
).fetchone()
if row is None:
# Re-ingest produced a different root than the peer claimed.
# The pulled text doesn't reproduce the requested root under
# this peer's chunker/canonicalization. Fail closed.
actual = conn.execute(
"SELECT document_root FROM documents WHERE document_uri=? "
"ORDER BY ingest_ts DESC LIMIT 1",
(delivered_uri,),
).fetchone()
actual_root = actual["document_root"] if actual else None
print(
"error: local re-ingest produced "
f"{actual_root!r}, expected {args.root!r}",
file=sys.stderr,
)
return 2
with transaction(conn):
event_hash = append_audit(
conn,
event_type="mesh_pulled",
body={
"document_root": args.root,
"document_uri": delivered_uri,
"peer": args.peer,
},
subject_root=args.root,
)
finally:
conn.close()
print(json.dumps({
"status": "pulled",
"document_root": args.root,
"document_uri": delivered_uri,
"shard": str(args.db),
"audit_event_hash": event_hash,
}, indent=2))
return 0
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="aborist",
@ -2210,6 +2317,14 @@ def build_parser() -> argparse.ArgumentParser:
mesh_sync.add_argument("--verbose", type=int, default=10, help="include this many ack details in output (default: 10)")
mesh_sync.set_defaults(func=_cmd_mesh_sync)
mesh_pull = mesh_sub.add_parser(
"pull",
help="pull one document body from a peer by document_root",
)
mesh_pull.add_argument("--root", required=True, help="64-char hex document_root to pull")
mesh_pull.add_argument("--peer", required=True, help="peer URL, e.g. http://other.example.com:8400")
mesh_pull.set_defaults(func=_cmd_mesh_pull)
return p

250
tests/test_mesh_cli_pull.py Normal file
View file

@ -0,0 +1,250 @@
"""CLI integration: `aborist mesh pull`.
Closes the request half of the gossip loop. Wire-layer Merkle / signature
verification is covered in test_mesh_wire_e2e.py — these tests cover the
argparse plumbing, gating, and the local-ingest round-trip that the pull
verb performs after the wire client returns a verified body.
"""
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_pull_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_pull_refuses_when_mesh_not_initialized(tmp_path, capsys):
db = tmp_path / "x.db"
parser = build_parser()
args = parser.parse_args([
"--db", str(db),
"mesh", "pull",
"--root", "ab" * 32,
"--peer", "http://0:0",
])
rc = args.func(args)
assert rc == 2
assert "not initialized" in capsys.readouterr().err
def test_pull_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")
# No set_enabled — flag stays off.
finally:
conn.close()
parser = build_parser()
args = parser.parse_args([
"--db", str(db),
"mesh", "pull",
"--root", "ab" * 32,
"--peer", "http://0:0",
])
rc = args.func(args)
assert rc == 2
assert "enabled" in capsys.readouterr().err
# ---------------------------------------------------------------------------
# Happy path: bob pulls a doc alice has
# ---------------------------------------------------------------------------
def test_pull_fetches_and_ingests_doc_from_peer(tmp_path, capsys):
"""Alice has a document; bob pulls it. Bob's local DB ends up with the
same document_root, an audit event is written, and the response JSON
reports `pulled`."""
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
# Alice ingests; bob does not have it yet.
document_root = _ingest(
alice_db,
uri="https://example.com/alpha",
content="Alpha content. The quick brown fox jumps over the lazy dog.",
title="Alpha",
)
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", "pull",
"--root", document_root,
"--peer", srv.url,
])
rc = args.func(args)
assert rc == 0
finally:
srv.stop()
payload = json.loads(capsys.readouterr().out)
assert payload["status"] == "pulled"
assert payload["document_root"] == document_root
assert payload["document_uri"] == "https://example.com/alpha"
assert payload["shard"] == str(bob_db)
assert isinstance(payload["audit_event_hash"], str)
assert len(payload["audit_event_hash"]) == 64
# Bob's DB now contains the document at the same root.
b_conn = connect(bob_db)
try:
row = b_conn.execute(
"SELECT document_root, document_uri FROM documents WHERE document_root=?",
(document_root,),
).fetchone()
audit = b_conn.execute(
"SELECT event_type, subject_root, body FROM audit_events "
"WHERE event_type='mesh_pulled' ORDER BY seq DESC LIMIT 1"
).fetchone()
finally:
b_conn.close()
assert row is not None
assert row["document_uri"] == "https://example.com/alpha"
assert audit is not None
assert audit["subject_root"] == document_root
body = json.loads(audit["body"])
assert body["document_root"] == document_root
assert body["document_uri"] == "https://example.com/alpha"
assert body["peer"] == srv.url
# ---------------------------------------------------------------------------
# Idempotence: pulling a doc bob already has returns `already_present`
# ---------------------------------------------------------------------------
def test_pull_already_present_when_local_has_doc(tmp_path, capsys):
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
# Both peers ingest the same content -> identical document_root.
content = "Beta content. The quick brown fox jumps over the lazy dog."
a_root = _ingest(alice_db, uri="https://example.com/beta", content=content, title="Beta")
b_root = _ingest(bob_db, uri="https://example.com/beta", content=content, title="Beta")
assert a_root == b_root
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", "pull",
"--root", a_root,
"--peer", srv.url,
])
rc = args.func(args)
assert rc == 0
finally:
srv.stop()
payload = json.loads(capsys.readouterr().out)
assert payload["status"] == "already_present"
assert payload["document_root"] == a_root
assert payload["document_uri"] == "https://example.com/beta"
# No mesh_pulled event was written since we short-circuited.
b_conn = connect(bob_db)
try:
rows = b_conn.execute(
"SELECT COUNT(*) FROM audit_events WHERE event_type='mesh_pulled'"
).fetchone()[0]
finally:
b_conn.close()
assert rows == 0
# ---------------------------------------------------------------------------
# Failure: peer returns 404 for unknown root
# ---------------------------------------------------------------------------
def test_pull_returns_rc2_on_unknown_root(tmp_path, capsys):
"""Peer responds 404 (no such document_root) -> rc=2, error to stderr."""
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
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", "pull",
"--root", "ff" * 32,
"--peer", srv.url,
])
rc = args.func(args)
assert rc == 2
finally:
srv.stop()
err = capsys.readouterr().err
assert "pull failed" in err