mesh: 'serve' and 'sync' CLI verbs over the wire layer
Wires the gossip-wire core (commit f141bab) into the user-facing CLI:
aborist mesh serve --host 127.0.0.1 --port 8400
Spawns MeshWireServer, blocks until SIGINT. Refuses if mesh isn't
initialized or .enabled is off — same gate every other mesh verb
enforces. Stdout is unbuffered JSON status lines.
aborist mesh sync --peer http://other.example:8400 [--limit N]
Enumerates local documents (most-recent N, default 100), fires one
ANNOUNCE_ROOT per document via MeshWireClient, reports counts of
acknowledged vs. errored. Pulling missing roots back from the peer
is a v2 addition — this verb pushes only.
Tests (6): gating refusals on uninitialized + disabled mesh; sync
round-trip with two peers proving Alice's audit chain gains one
'mesh_received' per Bob announce; unreachable-peer reporting; serve
smoke-test via start_in_thread + GET /mesh/info.
This commit is contained in:
parent
4a994b7ead
commit
5427a61768
2 changed files with 309 additions and 0 deletions
106
aborist/cli.py
106
aborist/cli.py
|
|
@ -1592,6 +1592,95 @@ def _cmd_mesh_rotate(args: argparse.Namespace) -> int:
|
|||
return 0
|
||||
|
||||
|
||||
def _cmd_mesh_serve(args: argparse.Namespace) -> int:
|
||||
"""Run the HTTP gossip server until SIGINT."""
|
||||
from aborist.mesh import is_enabled, load_identity
|
||||
from aborist.mesh.wire import MeshWireServer
|
||||
|
||||
conn = connect(args.db)
|
||||
try:
|
||||
if load_identity(conn) is None:
|
||||
print("error: mesh not initialized; run 'aborist mesh init' first", file=sys.stderr)
|
||||
return 2
|
||||
if not is_enabled(conn):
|
||||
print("error: mesh.enabled is off; run 'aborist mesh enable' first", file=sys.stderr)
|
||||
return 2
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
srv = MeshWireServer(args.db, host=args.host, port=args.port)
|
||||
print(json.dumps({"status": "serving", "url": srv.url, "db": str(args.db)}))
|
||||
sys.stdout.flush()
|
||||
try:
|
||||
srv.serve()
|
||||
except KeyboardInterrupt:
|
||||
print(json.dumps({"status": "stopped", "reason": "SIGINT"}))
|
||||
finally:
|
||||
srv.stop()
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_mesh_sync(args: argparse.Namespace) -> int:
|
||||
"""Send ANNOUNCE_ROOT for local documents to a peer.
|
||||
|
||||
v1: enumerates local document_roots (limited by --limit) and announces
|
||||
each. Receives one ack per announce. Pulling missing roots back from the
|
||||
peer is a future addition; this verb currently only pushes.
|
||||
"""
|
||||
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
|
||||
rows = conn.execute(
|
||||
"SELECT document_root, document_uri, chunking_version, "
|
||||
" canonicalization_version, schema_version "
|
||||
"FROM documents ORDER BY rowid DESC LIMIT ?",
|
||||
(args.limit,),
|
||||
).fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
sent: list[dict] = []
|
||||
errors: list[dict] = []
|
||||
with MeshWireClient(args.db, args.peer) as client:
|
||||
try:
|
||||
peer_info = client.info()
|
||||
except Exception as e:
|
||||
print(json.dumps({"status": "peer_unreachable", "peer": args.peer, "error": str(e)}, indent=2))
|
||||
return 2
|
||||
for r in rows:
|
||||
try:
|
||||
resp = client.announce_root(
|
||||
document_root=r["document_root"],
|
||||
source_uri=r["document_uri"],
|
||||
chunking_version=r["chunking_version"],
|
||||
canonicalization_version=r["canonicalization_version"],
|
||||
schema_version=r["schema_version"],
|
||||
)
|
||||
sent.append({"document_root": r["document_root"], "ack": resp})
|
||||
except Exception as e:
|
||||
errors.append({"document_root": r["document_root"], "error": str(e)})
|
||||
|
||||
print(json.dumps({
|
||||
"status": "synced",
|
||||
"peer": args.peer,
|
||||
"peer_member_id": peer_info.get("member_id"),
|
||||
"peer_epoch": peer_info.get("current_epoch"),
|
||||
"announced": len(sent),
|
||||
"errors": len(errors),
|
||||
"sent": sent[: args.verbose],
|
||||
"error_samples": errors[:5],
|
||||
}, indent=2))
|
||||
return 0 if not errors else 1
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
p = argparse.ArgumentParser(
|
||||
prog="aborist",
|
||||
|
|
@ -2104,6 +2193,23 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
mesh_rotate.add_argument("--reason", default="scheduled")
|
||||
mesh_rotate.set_defaults(func=_cmd_mesh_rotate)
|
||||
|
||||
mesh_serve = mesh_sub.add_parser(
|
||||
"serve",
|
||||
help="run the HTTP gossip server (blocks until SIGINT)",
|
||||
)
|
||||
mesh_serve.add_argument("--host", default="127.0.0.1", help="bind host (default: 127.0.0.1)")
|
||||
mesh_serve.add_argument("--port", type=int, default=8400, help="bind port (default: 8400)")
|
||||
mesh_serve.set_defaults(func=_cmd_mesh_serve)
|
||||
|
||||
mesh_sync = mesh_sub.add_parser(
|
||||
"sync",
|
||||
help="announce local document_roots to a peer's gossip server",
|
||||
)
|
||||
mesh_sync.add_argument("--peer", required=True, help="peer URL, e.g. http://other.example.com:8400")
|
||||
mesh_sync.add_argument("--limit", type=int, default=100, help="announce at most N most-recent docs (default: 100)")
|
||||
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)
|
||||
|
||||
return p
|
||||
|
||||
|
||||
|
|
|
|||
203
tests/test_mesh_cli.py
Normal file
203
tests/test_mesh_cli.py
Normal file
|
|
@ -0,0 +1,203 @@
|
|||
"""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",
|
||||
])
|
||||
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"] == 2
|
||||
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_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"
|
||||
Loading…
Add table
Add a link
Reference in a new issue