diff --git a/AGENTS.md b/AGENTS.md index c5812d8..b1147f8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,37 +1,42 @@ # AGENTS.md ## Repo shape -- `rfc.txt` (FRX — Federated Retrieval Exchange, Draft 0.3) is the normative spec; `src/` is the Phase 1 `frxd` implementation (single crate, two binaries). +- `rfc.txt` (FRX — Federated Retrieval Exchange, Draft 0.4) is the normative spec; `src/` is the Phase 1 `frxd` implementation (single crate, two binaries). - `frxd` is the member node (init/add/index/serve/relay/query/status); `frx` is the thin client (search/query/status). Relay and node roles are separate subcommands. -- Commands: `cargo build`, `cargo test` (62 tests: unit in `src/`; e2e `tests/phase1.rs`; conformance `tests/conformance.rs`; concurrency/restart `tests/concurrency.rs`; real subprocess CLI `tests/cli.rs`; 1000-doc `tests/scale.rs`; purge-log absence `tests/purges.rs`; shared fixtures `tests/common/mod.rs`). No CI/lint config. +- Commands: `cargo build`, `cargo test` (68 tests: unit in `src/`; e2e `tests/phase1.rs`; conformance `tests/conformance.rs`; aggregates + member directory `tests/aggregates.rs`; concurrency/restart `tests/concurrency.rs`; real subprocess CLI `tests/cli.rs`; 1000-doc `tests/scale.rs`; purge-log absence `tests/purges.rs`; shared fixtures `tests/common/mod.rs`). No CI/lint config. - E2E pattern: relay + nodes in-process on ephemeral ports with tempdir corpora; use `tests/common/mod.rs` helpers (`spawn_relay*`, `query_envelope`, `poll_messages`, `register`) for new coverage. Raw relay polls return envelopes (payload under `body`), not response bodies. ## Editing the spec - Read all of `rfc.txt` before editing; it is the sole source of truth and is deliberately terse. - Keep the plain-text single-file format. Don't restructure into Markdown files unless asked. -- Invariants I1–I9 (§2) are normative; proposals contradicting them (scores in responses, topic taxonomy, announce stream, dispute messages, replayable broadcast) are out of scope by design. +- Invariants I1–I9 (§2) are normative; proposals contradicting them (scores in responses, topic taxonomy, announce stream, dispute messages, replayable broadcast, in-protocol pricing/settlement) are out of scope by design. - Appendix B (Purge Log) is normative: a rejected mechanism may only be re-proposed if the written rationale is addressed. - §10 Open Issues are known gaps, not oversights (e.g., signature canonicalization blocks Phase-1 interop). Check it before "fixing" something. -- Use the spec's vocabulary — member/querier/responder, citations/receipts/aggregates, source/enrichment members — not client/server or search-engine terms. +- Use the spec's vocabulary — member/querier/responder, aggregates, source/enrichment members — not client/server or search-engine terms. +- A new `MUST` is only legitimate if it is observable at the boundary, deterministically verifiable by a peer, beneficial to the counterparty, and not derivable from local policy. Ranking/ordering/presentation fails this test and stays local (§5); scores never travel (I6). ## Implementation notes -- Envelope signing is provisional (`src/crypto.rs`): `FRX/0.3` + fields + sorted-key canonical JSON body. §10's signature canonicalization open issue is unsolved — never present this scheme as interoperable. -- Phase 1 only: no aggregates (Phase 2), no receipts/lineage/delegation (Phase 3). Responses travel relay-mediated unicast; transport is HTTP long-poll, not SSE. +- Envelope signing is provisional (`src/crypto.rs`): `FRX/0.4` + fields + sorted-key canonical JSON body. §10's signature canonicalization open issue is unsolved — never present this scheme as interoperable. +- Built: envelope/query/response, Tantivy index, aggregates, member directory. Not built: dashboard UI, directory watching, TLS, lineage/delegation. Responses travel relay-mediated unicast; transport is HTTP long-poll, not SSE. +- Economics is out of protocol scope (I4: aggregates advise, contracts govern): no receipt, citation, pricing, or settlement fields or message types exist or may be added. - Relay verifies signatures, carries only `query` broadcasts, holds no history (queue drained on poll), and returns 429 + Retry-After under backpressure — never silent drops. -- Responder searches only collections marked shared (I9), stays silent when nothing matches, and emits ordered results with honest `truncated`/`more_available` and no scores (I6). +- Responder searches only collections marked shared (I9), stays silent when nothing matches, and emits results with honest `truncated`/`more_available` and no scores (I6). BM25 order is a local implementation detail, not protocol surface. - Index layout: Tantivy at `/index`, collections manifest at `/collections.toml`; `exposure` (metadata|full) gates whether `content` is returned. - Egress checks live in the responder path (`src/node.rs` `respond`), not the relay — keep private collections unreachable there. -- Appendix B is executable policy: `tests/purges.rs` has one absence test per rejected mechanism (11 rows, not the RFC's informal "nine"). Add a negative test there before ever re-proposing one, and only if the rationale is addressed. +- Appendix B is executable policy: `tests/purges.rs` has one absence test per rejected mechanism (12 rows). Add a negative test there before ever re-proposing one, and only if the rationale is addressed. - `add`/`reindex` reset a collection (delete by manifest `name`) before re-adding, so deleted files don't linger; collection identity is its name, and same-named collections replace each other. - Relay backpressure is global: any member's full queue 429s every publisher until drained (visible per §3, but one lagging member can stall the firehose — revisit before scale). +- Member directory lives at `/members.toml` (name, pubkey, class source|enrichment), mtime-reloaded so MA updates need no restart; empty directory = open bootstrap mode (admission is MA policy, §10). Receivers drop content-bearing responses from members listed as enrichment (metadata-only, §6). +- Aggregate semantics are our implementation choices from a terse spec: requests are `aggregate` envelopes carrying only `period`; replies carry `sent` (broadcasts that month) / `passed` (responses consumed from that member); granularity floor is enforced as YYYY or YYYY-MM only (finer rejected), yearly rolls up months. Revisit with §10 sufficiency review. ## Known gaps (Phase 2/3, intentional — don't fake them) -- No aggregate serving/dashboard (RFC §9 conformance is partial without it), no directory watching (new files need `reindex`), no receipts/lineage/delegation, no TLS, no consumer admission/relay discovery. +- No dashboard UI, no directory watching (new files need `reindex`), no TLS, no user-supplied URL ingestion, no node-side (bilateral) rate limiting. +- Receipts/settlement are out of protocol scope, not unimplemented (I4; Appendix B row 2). Lineage and delegation remain §10 open issues — unspecified without a supply stream, so not buildable as written; don't invent them silently. - Node query dedup is by `qid` only; there is no envelope replay/nonce window (RFC doesn't require one). ## Technical plans (deliberately not in the RFC) - Record plans here — not as spec edits — when they are implementation/demo choices rather than protocol surface. -- Demo plan: build a useful end-to-end demo on GDELT and Common Crawl (CC-NEWS; sometimes called "OpenCrawl" in discussion) as enrichment members / backfill seeding. RFC §6 and Appendix A already name both as example derived corpora, so no new mechanisms are required; enrichment members are metadata-only exposure and sit outside the citation market. +- Demo plan: build a useful end-to-end demo on GDELT and Common Crawl (CC-NEWS; sometimes called "OpenCrawl" in discussion) as enrichment members / backfill seeding. RFC §6 and Appendix A already name both as example derived corpora, so no new mechanisms are required; enrichment members are metadata-only exposure. - Phase 1 (two-node query/response) is built and tested; the enrichment demo layers on top of it. - Language: Rust (settled, matches §7). Decided by the engine requirement, not preference: Tantivy gives in-process Lucene-class BM25 + incremental indexing; C/C++ embedded alternatives are worse (Xapian GPL-2+, CLucene unmaintained, SQLite FTS5 thin), plus single static musl binaries for the install story and memory safety on the untrusted network/crypto path. Don't re-litigate. - frxd modes (one binary, config toggles, no code required of publishers): querier (broadcast/local-first search), responder (match incoming queries against shared collections, sign), local index (watch dirs, extract text, explicit shared marking per I9). Use RFC terms querier/responder, not "subscriber/publisher". diff --git a/rfc.txt b/rfc.txt index f66baf1..018ab97 100644 --- a/rfc.txt +++ b/rfc.txt @@ -1,19 +1,19 @@ FRX — Federated Retrieval Exchange -Status: Draft 0.3. Experimental. Reference implementation: frxd (Rust). +Status: Draft 0.4. Experimental. Reference implementation: frxd (Rust). 1. Summary -FRX is a membership federation for retrieval. Content owners answer broadcast queries; citation traffic is the payment. The protocol's normative surface is minimal: signed messages, budgets, truncation honesty, aggregate courtesy. Ranking, reputation, caching, filtering, verification, and quality judgment are node-local. There is no supply announcement stream: a responder answers only from content it already holds. A member's mandatory work per incoming query is a local lookup — nothing heavier is required. +FRX is a membership federation for retrieval. Content owners answer broadcast queries. The protocol's normative surface is minimal: signed messages, budgets, truncation honesty, aggregate courtesy. Ranking, reputation, caching, filtering, verification, and quality judgment are node-local. There is no supply announcement stream: a responder answers only from content it already holds. A member's mandatory work per incoming query is a local lookup — nothing heavier is required. 2. Invariants I1 Ingress consent — content enters a member's index only via the publisher adding its own content or a user-supplied URL. I2 Local sovereignty — nothing centralizes a decision a member could make locally. I3 Broadcast privacy — broadcast payloads MUST NOT contain third-party private content. Queries derived from scoring others' posts MUST be canonicalized (claims/entities). First-party user-initiated search text MAY be sent as typed. - I4 Channel separation — the market pays (citations), statements advise (aggregates), contracts punish (membership). Never cross-wired. - I5 Mode neutrality — no privileged roles. Eagerness is a local choice: an eager member fetches and retains proactively; a lazy member answers only from content already at hand. Neither holds a privileged role. - I6 Ordering travels, scores don't — ranked orderings, never numeric relevance. + I4 Channel separation — statements advise (aggregates), contracts govern (membership); never cross-wired. The protocol carries no pricing, metering, or settlement. + I5 Role symmetry — no privileged roles. Any member may originate queries or responses; no peer may require remote work per incoming query (I7). Neither role is privileged. + I6 No scores — responses never carry numeric relevance; selection and presentation order are local policy (I2, §5). I7 Local-lookup cost — a member's mandatory work per incoming query is a local lookup. There is no supply stream to maintain, and no heavier mandatory work (fetch, embed, LLM). Cost scales with received query volume at lookup cost. I8 No shared vocabulary — no topic taxonomy; all filtering is receiver-local; the querier never classifies on behalf of receivers. I9 Egress consent — a member serves queries only from collections explicitly marked shared. Default is private. @@ -29,8 +29,6 @@ ORIGIN Demand query Broadcast Any member Bid response Unicast to querier Any member holding content Courtesy aggregate Bilateral, on request Querier -Settlement receipt (app layer) Public artifact Citing platform - Silence is conformant and informative: an unanswered query means no member's available content produced a response — not that the content does not exist. A member holding matching content may still choose silence (I2, I5). 4. Messages @@ -54,24 +52,24 @@ jsonc "results": [ { "url": "...", "title": "...", "summary": "...", "published": "...", "exposure": "...", "content": null } ], "truncated": false, "more_available": 0, "cursor": null } -MUST NOT exceed max_results. MUST set truncated honestly if more ranked results exist within budget (DNS TC-bit pattern). MUST be ordered; MUST NOT carry numeric relevance scores (I6). One result is a conformant, good response. exposure is "metadata" | "full" (paywall compatibility); content is present only when exposure is "full". Republication of another member's response before the querier publishes a citing post is a contract matter; the receipt is the sanctioned republication form, at citation granularity, synchronized to settlement. +MUST NOT exceed max_results. MUST set truncated honestly if more matching results exist within budget (DNS TC-bit pattern). MUST NOT carry numeric relevance scores (I6). One result is a conformant, good response. exposure is "metadata" | "full" (paywall compatibility); content is present only when exposure is "full". Republication of another member's response is a contract matter. aggregate (on request, per member, per period) jsonc -{ "period": "2026-03", "sent": 12400, "passed": 310, "cited": 44 } + { "period": "2026-03", "sent": 12400, "passed": 310 } Counters are the querier's own; no dispute or appeal messages exist, and none may be added (I4). Granularity floor is normative. -There is no publish/announce message. Document metadata is carried in responses (above). Document lineage (revision/supersedes) and delegation to an indexer are unspecified in pull-only mode; see §10. +There is no publish/announce message. Document metadata is carried in responses (above). Document lineage (revision/supersedes) and delegation to an indexer are unspecified without a supply stream; see §10. 5. Local Policy Domains -Protocol-silent by design (I2): ranking, relevance gating, reputation counters and throttles, verification/spot-checks, caching and invalidation, claim minting, mode (eager/lazy), external fallback, sharing policy above the I9 floor. Advisory reputation bureaus MAY exist; no member is bound. +Protocol-silent by design (I2): ranking, ordering, presentation, relevance gating, reputation counters and throttles, verification/spot-checks, caching and invalidation, claim minting, mode (eager/lazy), external fallback, sharing policy above the I9 floor. Advisory reputation bureaus MAY exist; no member is bound. 6. Membership -The MA governs identity, contract, expulsion — who, never quality. Admission cost is the Sybil defense. Expulsion grounds: fabrication, admission fraud, sustained abuse — never low quality (priced by the citation market: zero citations, zero traffic). Escalation: local throttle → advisory aggregates → MA warning → delisting → expulsion. Aggregates are inadmissible as sanction evidence (I4). Membership classes: source members (own content) and enrichment members (derived corpora, e.g. GDELT/CC-NEWS bots — metadata-only exposure, earn no traffic, sit outside the citation market, transformation logic open and auditable). +The MA governs identity, contract, expulsion — who, never quality. Admission cost is the Sybil defense. Expulsion grounds: fabrication, admission fraud, sustained abuse — never low quality. Escalation: local throttle → advisory aggregates → MA warning → delisting → expulsion. Aggregates are inadmissible as sanction evidence (I4). Membership classes: source members (own content) and enrichment members (derived corpora, e.g. GDELT/CC-NEWS bots — metadata-only exposure, transformation logic open and auditable). 7. Reference Implementation — frxd @@ -81,21 +79,21 @@ First run: generate keypair, write config.toml, open localhost web UI (plus frx As responder: subscribe to the live query stream via configured relays; match incoming queries against shared collections only (receiver-local lexical/embedding match); respond within budget with honest truncation; sign. -As consumer: search is local-first; a network toggle broadcasts the query and merges responses, provenance-marked ("your files" / "member X"). Click-through preserves attribution — receipts at consumer scale. Fetched content is retained (fetch-on-miss-and-retain): the kept-set converges to the demand-weighted corpus. +As consumer: search is local-first; a network toggle broadcasts the query and merges responses, provenance-marked ("your files" / "member X"). Fetched content is retained (fetch-on-miss-and-retain): the kept-set converges to the demand-weighted corpus. -Dashboard: sent / passed / cited per period (self-derived from referrer traffic and observable receipts; aggregates requested from queriers on demand). "Your files were cited 14 times this week" is the reward loop. +Dashboard: sent / passed per period (self-derived from local counters; aggregates requested from queriers on demand). Stack: tokio, tantivy, axum (localhost UI), reqwest, ed25519-dalek, notify. One binary; frxd relay runs a relay for sovereignty-minded users. -Build order: Phase 1 — envelope, query stream, query, response (demoable between two nodes; consumer packaging is a skin over this). Phase 2 — aggregates, dashboard. Phase 3 — receipts, lineage, delegation (app layer). +Build order: Phase 1 — envelope, query stream, query, response (demoable between two nodes; consumer packaging is a skin over this). Phase 2 — aggregates, dashboard. Phase 3 — lineage, delegation (app layer). 8. Security & Privacy Considerations -Query visibility is total among members — the claim stream is the attention product; abstraction level and membership are the boundary (I3). Derived queries SHOULD minimize personal data (I3); each member is responsible for the content of its own messages. Response streams are strategic disclosure (corpus mapping, intake intelligence) — unicast, need-to-know. Amplification is bounded by bilateral transport limits and contract, not routing. Publisher self-promotion is the expected adversarial mode; defense is local (gate, pass-rate throttle, citation-weighted source reputation). Enrichment members' filters are an editorial power — auditable openness is the mitigation. +Query visibility is total among members; abstraction level and membership are the boundary (I3). Derived queries SHOULD minimize personal data (I3); each member is responsible for the content of its own messages. Response streams are strategic disclosure (corpus mapping, intake intelligence) — unicast, need-to-know. Amplification is bounded by bilateral transport limits and contract, not routing. Publisher self-promotion is the expected adversarial mode; defense is local (gate, pass-rate throttle, local source reputation). Enrichment members' filters are an editorial power — auditable openness is the mitigation. 9. Conformance -A conforming implementation: signs all messages with a listed member key; publishes queries to the firehose only; respects I3 and I9; respects max_results; truncates honestly; sends orderings without scores; enforces visible transport backpressure, never silent transport drops; serves aggregates on request at or above the granularity floor; ingests content only via publisher-added content or user-supplied URL; implements no dispute messages. +A conforming implementation: signs all messages with a listed member key; publishes queries to the firehose only; respects I3 and I9; respects max_results; truncates honestly; sends results without scores; enforces visible transport backpressure, never silent transport drops; serves aggregates on request at or above the granularity floor; ingests content only via publisher-added content or user-supplied URL; implements no dispute messages. 10. Open Issues @@ -103,13 +101,12 @@ Consumer admission tier — automated/invite admission for distributed binaries Relay discovery and default-relay governance (shipped defaults are soft centralization; mitigate with multiple defaults + one-command self-host). Delegation grant mechanism. Signature canonicalization scheme (blocks Phase-1 interop; required before two implementations can exchange a valid envelope). -Receipt data model (app layer; specify for cross-member consistency). -Document lineage (revision/supersedes) and delegation in pull-only mode — previously carried by publish; now unspecified. +Document lineage (revision/supersedes) and delegation without a supply stream — previously carried by publish; now unspecified. Claim/entity minting conventions — recommended, non-normative (I8). Appendix A. Recommended Local Practices (Non-Normative) -Two-stage ingestion (cheap gate before any LLM attention); per-peer sent/passed/cited counters, throttling on ratios; verdict memoization and claim normalization; demand-driven prefetch (fetch and retain on attention); hot-set replication; backfill seeding (Wikipedia/Wikidata, GDELT, CC-NEWS); measure query-to-claim collapse and domain concentration before sizing anything. +Two-stage ingestion (cheap gate before any LLM attention); per-peer sent/passed counters, throttling on ratios; verdict memoization and claim normalization; demand-driven prefetch (fetch and retain on attention); hot-set replication; backfill seeding (Wikipedia/Wikidata, GDELT, CC-NEWS); measure query-to-claim collapse and domain concentration before sizing anything. Appendix B. Purge Log (Normative) @@ -118,15 +115,15 @@ Re-proposals MUST address the rationale. MECHANISM REJECTED BECAUSE Per-query bounty, winner-selection, slashing Pays for query execution when responders own content; race-to-first rewards speed over honesty; anonymous-trust machinery with no anonymous peers +In-protocol citation accounting / receipts as settlement Economics is not protocol surface; a retrieval protocol cannot observe citations on the web, and self-issued artifacts have no trust anchor Protocol query dedup Undecidable at network layer; duplicates cost receivers ~nothing; dedup is local caching k-fetch ingestion attestations Swarm trust machinery; signatures + local spot-checks suffice Result-count etiquette (SERP min/max) Rendering convention from the one-actor world; quantity is querier-local, selection responder-local; budget + truncation flag suffice Global reputation score Relevance is locally defined; consumption is observable where it happens Aggregate appeals Converts courtesy into litigation; drags local policy into network process Topic channels Shared vocabulary is a governance object; sender-side routing by the least-informed party yields undetectable false negatives; at membership N, broadcast + receiver filtering is cheaper than the coordination -Broadcast responses ("evidence commons") Multiplies the heavy stream by N; requires network-layer query equivalence (purged); receipts provide post-settlement publicity +Broadcast responses ("evidence commons") Multiplies the heavy stream by N; requires network-layer query equivalence (purged); unicast need-to-know suffices Normative query canonical form Receiver-local matching handles arbitrary phrasing; intelligibility is self-enforcing; the protocol constrains only what must not enter a broadcast (I3) Supply announce firehose Pull-only supply: responders answer from content already held; a visible announce stream adds cost and privacy exposure with no discovery benefit, since queries already reach all members Durable / replayable broadcast stream A live, ephemeral broadcast suffices for routing; persistence and history-replay buy nothing and invite retention and erasure problems - -Two closing notes. First, the open issue that matters most is consumer admission — everything else in §7 is straightforward engineering, but "naive user installs a binary and joins a membership org" has real tension with the admission-cost Sybil defense, and I'd resolve it deliberately rather than by accident (invite codes at launch; a lightweight consumer tier later; or defaulting consumer nodes to delegation through an indexer member until they've earned standing). Second, when the reference implementation exists, the purge log stops being documentation and becomes a test suite: nine unit tests, each asserting the absence of a mechanism. A spec this small can afford to test what it refuses to do. +Two closing notes. First, the open issue that matters most is consumer admission — everything else in §7 is straightforward engineering, but "naive user installs a binary and joins a membership org" has real tension with the admission-cost Sybil defense, and I'd resolve it deliberately rather than by accident (invite codes at launch; a lightweight consumer tier later; or defaulting consumer nodes to delegation through an indexer member until they've earned standing). Second, when the reference implementation exists, the purge log stops being documentation and becomes a test suite: one unit test per row, each asserting the absence of a mechanism. A spec this small can afford to test what it refuses to do. diff --git a/src/commands.rs b/src/commands.rs index 39d6ad2..69f4b99 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -1,8 +1,9 @@ use std::path::Path; +use std::time::Duration; -use anyhow::{Context, Result}; +use anyhow::{Context, Result, anyhow}; -use crate::config::Config; +use crate::config::{CLASS_ENRICHMENT, CLASS_SOURCE, Config, Member, load_members, save_members}; use crate::index::{Collection, LocalIndex, load_collections, save_collections}; use crate::message::{EXPOSURE_FULL, EXPOSURE_METADATA}; use crate::node; @@ -91,6 +92,98 @@ pub async fn query( Ok(()) } +pub fn member_add(config_path: &Path, name: &str, pubkey: &str, class: &str) -> Result<()> { + let config = Config::load(config_path)?; + let bytes = hex::decode(pubkey).context("pubkey must be hex")?; + if bytes.len() != 32 { + return Err(anyhow!("pubkey must be 32 bytes (64 hex characters)")); + } + let class = if class == CLASS_ENRICHMENT { + CLASS_ENRICHMENT + } else { + CLASS_SOURCE + }; + let mut members = load_members(&config.members_path())?; + members.retain(|member| member.name != name && member.pubkey != pubkey); + members.push(Member { + name: name.to_string(), + pubkey: pubkey.to_ascii_lowercase(), + class: class.to_string(), + }); + save_members(&config.members_path(), &members)?; + println!( + "listed {name} ({class}) in {}", + config.members_path().display() + ); + Ok(()) +} + +pub fn member_remove(config_path: &Path, name: &str) -> Result<()> { + let config = Config::load(config_path)?; + let mut members = load_members(&config.members_path())?; + let before = members.len(); + members.retain(|member| member.name != name); + if members.len() == before { + println!("no member named {name}"); + return Ok(()); + } + save_members(&config.members_path(), &members)?; + println!("removed {name}"); + Ok(()) +} + +pub fn member_list(config_path: &Path) -> Result<()> { + let config = Config::load(config_path)?; + let members = load_members(&config.members_path())?; + if members.is_empty() { + println!( + "member directory {} is empty (open bootstrap mode: any valid signature is accepted)", + config.members_path().display() + ); + return Ok(()); + } + for member in members { + println!("{} [{}] {}", member.name, member.class, member.pubkey); + } + Ok(()) +} + +pub async fn aggregates( + config_path: &Path, + from: Option<&str>, + period: Option<&str>, + timeout_ms: Option, +) -> Result<()> { + let config = Config::load(config_path)?; + let base = format!("http://{}", config.node.listen); + let period = period + .map(str::to_string) + .unwrap_or_else(node::current_period); + match from { + Some(to) => { + let value = node::control_aggregate_request(&base, to, &period, timeout_ms).await?; + println!("{}", serde_json::to_string_pretty(&value)?); + } + None => { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(2)) + .build()?; + let response = client + .get(format!("{base}/v1/local/aggregates?period={period}")) + .send() + .await + .context("calling local node (is `frxd serve` running?)")?; + let status = response.status(); + let value: serde_json::Value = response.json().await?; + if !status.is_success() { + return Err(anyhow!("local node error {status}: {value}")); + } + println!("{}", serde_json::to_string_pretty(&value)?); + } + } + Ok(()) +} + pub async fn status(config_path: &Path) -> Result<()> { let config = Config::load(config_path)?; let base = format!("http://{}", config.node.listen); diff --git a/src/config.rs b/src/config.rs index 0756dd4..5b29eb2 100644 --- a/src/config.rs +++ b/src/config.rs @@ -20,12 +20,47 @@ pub struct NodeSection { pub name: String, pub listen: String, pub relays: Vec, - #[serde(default)] - pub trusted_keys: Vec, #[serde(default = "default_true")] pub responder: bool, } +pub const CLASS_SOURCE: &str = "source"; +pub const CLASS_ENRICHMENT: &str = "enrichment"; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Member { + pub name: String, + pub pubkey: String, + #[serde(default = "default_class")] + pub class: String, +} + +#[derive(Debug, Default, Serialize, Deserialize)] +struct MembersFile { + #[serde(default, rename = "member")] + members: Vec, +} + +pub fn load_members(path: &Path) -> Result> { + if !path.exists() { + return Ok(Vec::new()); + } + let raw = fs::read_to_string(path).context("reading member directory")?; + let parsed: MembersFile = toml::from_str(&raw).context("parsing member directory")?; + Ok(parsed.members) +} + +pub fn save_members(path: &Path, members: &[Member]) -> Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let file = MembersFile { + members: members.to_vec(), + }; + fs::write(path, toml::to_string_pretty(&file)?)?; + Ok(()) +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct QuerySection { #[serde(default = "default_max_results")] @@ -61,6 +96,10 @@ fn default_true() -> bool { true } +fn default_class() -> String { + CLASS_SOURCE.to_string() +} + fn default_max_results() -> usize { 5 } @@ -80,7 +119,6 @@ impl Config { name: name.to_string(), listen: listen.to_string(), relays: vec![relay.to_string()], - trusted_keys: Vec::new(), responder: true, }, query: QuerySection::default(), @@ -122,6 +160,10 @@ impl Config { self.data_dir().join("collections.toml") } + pub fn members_path(&self) -> PathBuf { + self.data_dir().join("members.toml") + } + pub fn load_key(&self) -> Result { let raw = fs::read_to_string(self.key_path()) .with_context(|| format!("reading key {}", self.key_path().display()))?; diff --git a/src/lib.rs b/src/lib.rs index 2916636..da06286 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,4 +9,4 @@ pub mod relay; pub mod render; pub const VERSION: &str = env!("CARGO_PKG_VERSION"); -pub const PROTOCOL: &str = "FRX/0.3"; +pub const PROTOCOL: &str = "FRX/0.4"; diff --git a/src/main.rs b/src/main.rs index b20b4d5..13e7673 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,7 +11,7 @@ use frxd::{commands, node, relay}; #[command( name = "frxd", version, - about = "FRX member node — querier, responder, and local index (Draft 0.3)" + about = "FRX member node — querier, responder, and local index (Draft 0.4)" )] struct Cli { #[arg(long, global = true, default_value = "frxd.toml")] @@ -66,6 +66,32 @@ enum Command { capacity: usize, }, Status, + Member { + #[command(subcommand)] + command: MemberCommand, + }, + Aggregates { + #[arg(long)] + from: Option, + #[arg(long)] + period: Option, + #[arg(long)] + timeout_ms: Option, + }, +} + +#[derive(Subcommand)] +enum MemberCommand { + Add { + name: String, + pubkey: String, + #[arg(long, default_value = "source")] + class: String, + }, + Remove { + name: String, + }, + List, } #[derive(Clone, Copy, ValueEnum)] @@ -136,6 +162,23 @@ async fn main() -> Result<()> { relay::run(listener, capacity).await?; } Command::Status => commands::status(&cli.config).await?, + Command::Member { command } => match command { + MemberCommand::Add { + name, + pubkey, + class, + } => commands::member_add(&cli.config, &name, &pubkey, &class)?, + MemberCommand::Remove { name } => commands::member_remove(&cli.config, &name)?, + MemberCommand::List => commands::member_list(&cli.config)?, + }, + Command::Aggregates { + from, + period, + timeout_ms, + } => { + commands::aggregates(&cli.config, from.as_deref(), period.as_deref(), timeout_ms) + .await? + } } Ok(()) } diff --git a/src/message.rs b/src/message.rs index 30cb9e7..4584875 100644 --- a/src/message.rs +++ b/src/message.rs @@ -115,7 +115,6 @@ pub struct AggregateBody { pub period: String, pub sent: u64, pub passed: u64, - pub cited: u64, } pub fn require_type(envelope: &Envelope, expected: &str) -> Result<()> { diff --git a/src/node.rs b/src/node.rs index 1470f19..7ab37f3 100644 --- a/src/node.rs +++ b/src/node.rs @@ -1,11 +1,12 @@ use std::collections::{HashMap, HashSet}; +use std::fs; use std::net::SocketAddr; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; -use std::time::{Duration, Instant}; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::{Duration, Instant, SystemTime}; use anyhow::{Context, Result, anyhow}; -use axum::extract::State; +use axum::extract::{Query, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; @@ -15,25 +16,51 @@ use serde_json::{Value, json}; use tokio::net::TcpListener; use tokio::task::JoinHandle; -use crate::config::Config; +use crate::config::{CLASS_ENRICHMENT, Config, Member, load_members}; use crate::crypto::Keypair; use crate::index::{LocalIndex, SearchHit, response_items}; use crate::message::{ - Envelope, QueryBody, ResponseBody, ResponseItem, TYPE_AGGREGATE, TYPE_QUERY, TYPE_RESPONSE, - build_response, + AggregateBody, Envelope, QueryBody, ResponseBody, ResponseItem, TYPE_AGGREGATE, TYPE_QUERY, + TYPE_RESPONSE, build_response, }; +#[derive(Default)] +struct Aggregates { + sent: HashMap, + passed: HashMap<(String, String), u64>, +} + pub struct Node { pub config: Config, pub key: Keypair, index: LocalIndex, + members: RwLock>, + members_mtime: Mutex>, + aggregates: Mutex, seen: Mutex>, pending: Mutex>>, + pending_aggregates: Mutex>, client: reqwest::Client, sent: AtomicU64, received: AtomicU64, } +pub fn current_period() -> String { + chrono::Utc::now().format("%Y-%m").to_string() +} + +pub fn is_valid_period(period: &str) -> bool { + if period.len() == 4 { + return period.chars().all(|c| c.is_ascii_digit()); + } + if period.len() != 7 || period.as_bytes()[4] != b'-' { + return false; + } + let digits = period[0..4].chars().all(|c| c.is_ascii_digit()); + let month: Result = period[5..7].parse(); + digits && month.map(|m| (1..=12).contains(&m)).unwrap_or(false) +} + pub struct NodeHandle { pub addr: SocketAddr, pub pubkey: String, @@ -83,6 +110,11 @@ impl Node { pub fn open(config: Config) -> Result> { let key = config.load_key()?; let index = LocalIndex::open(&config.index_dir())?; + let members_path = config.members_path(); + let members = load_members(&members_path)?; + let members_mtime = fs::metadata(&members_path) + .and_then(|metadata| metadata.modified()) + .ok(); let client = reqwest::Client::builder() .timeout(Duration::from_secs(15)) .build() @@ -91,14 +123,73 @@ impl Node { config, key, index, + members: RwLock::new(members), + members_mtime: Mutex::new(members_mtime), + aggregates: Mutex::new(Aggregates::default()), seen: Mutex::new(HashSet::new()), pending: Mutex::new(HashMap::new()), + pending_aggregates: Mutex::new(HashMap::new()), client, sent: AtomicU64::new(0), received: AtomicU64::new(0), })) } + fn refresh_members(&self) { + let path = self.config.members_path(); + let mtime = fs::metadata(&path) + .and_then(|metadata| metadata.modified()) + .ok(); + { + let last = self.members_mtime.lock().expect("members mtime lock"); + if *last == mtime { + return; + } + } + let Ok(members) = load_members(&path) else { + return; + }; + *self.members.write().expect("members lock") = members; + *self.members_mtime.lock().expect("members mtime lock") = mtime; + } + + fn member_class(&self, members: &[Member], pubkey: &str) -> Option { + members + .iter() + .find(|member| member.pubkey == pubkey) + .map(|member| member.class.clone()) + } + + pub fn aggregate_for(&self, period: &str, member: Option<&str>) -> AggregateBody { + let aggregates = self.aggregates.lock().expect("aggregates lock"); + let in_period = |candidate: &str| { + if period.len() == 4 { + candidate.starts_with(period) + } else { + candidate == period + } + }; + let sent = aggregates + .sent + .iter() + .filter(|(key, _)| in_period(key)) + .map(|(_, count)| *count) + .sum(); + let passed = aggregates + .passed + .iter() + .filter(|((key, key_period), _)| { + in_period(key_period) && member.map(|m| m == key).unwrap_or(true) + }) + .map(|(_, count)| *count) + .sum(); + AggregateBody { + period: period.to_string(), + sent, + passed, + } + } + pub fn doc_count(&self) -> u64 { self.index.doc_count() } @@ -158,6 +249,10 @@ impl Node { .expect("pending lock") .insert(qid.clone(), Vec::new()); self.sent.fetch_add(1, Ordering::SeqCst); + { + let mut aggregates = self.aggregates.lock().expect("aggregates lock"); + *aggregates.sent.entry(current_period()).or_default() += 1; + } let envelope = Envelope::new(&self.key, TYPE_QUERY, serde_json::to_value(&query)?); self.publish(&envelope).await; let timeout = Duration::from_millis(timeout_ms.unwrap_or(self.config.query.timeout_ms)); @@ -174,6 +269,13 @@ impl Node { self.received .fetch_add(collected.len() as u64, Ordering::SeqCst); for (member, body) in collected { + { + let mut aggregates = self.aggregates.lock().expect("aggregates lock"); + *aggregates + .passed + .entry((member.clone(), current_period())) + .or_default() += 1; + } responses.push(RemoteResponse { member, results: body.results, @@ -231,9 +333,14 @@ impl Node { if envelope.verify().is_err() { return; } - if !self.config.node.trusted_keys.is_empty() - && !self.config.node.trusted_keys.contains(&envelope.from) - { + self.refresh_members(); + let (class, listed) = { + let members = self.members.read().expect("members lock"); + let class = self.member_class(&members, &envelope.from); + let listed = members.is_empty() || class.is_some(); + (class, listed) + }; + if !listed { return; } match envelope.msg_type.as_str() { @@ -269,12 +376,45 @@ impl Node { let Ok(body) = envelope.parse_body::() else { return; }; + if class.as_deref() == Some(CLASS_ENRICHMENT) + && body.results.iter().any(|result| result.content.is_some()) + { + return; + } let mut pending = self.pending.lock().expect("pending lock"); if let Some(list) = pending.get_mut(&body.qid) { list.push((envelope.from.clone(), body)); } } - TYPE_AGGREGATE => {} + TYPE_AGGREGATE => { + let Ok(value) = envelope.parse_body::() else { + return; + }; + let Some(period) = value.get("period").and_then(Value::as_str) else { + return; + }; + if !is_valid_period(period) { + return; + } + if value.get("sent").is_some() { + let Ok(body) = envelope.parse_body::() else { + return; + }; + let mut pending = self + .pending_aggregates + .lock() + .expect("aggregate reply lock"); + pending.insert((envelope.from.clone(), body.period.clone()), body); + return; + } + let node = self.clone(); + let period = period.to_string(); + let requester = envelope.from.clone(); + let relay = relay.to_string(); + tokio::spawn(async move { + node.serve_aggregate(&period, &requester, &relay).await; + }); + } _ => {} } } @@ -287,6 +427,59 @@ impl Node { } let body = build_response(&query.qid, response_items(&hits), total, max); let envelope = Envelope::new(&self.key, TYPE_RESPONSE, serde_json::to_value(&body)?); + self.send_unicast(querier, envelope, relay).await + } + + pub async fn request_aggregate( + &self, + to: &str, + period: &str, + timeout_ms: Option, + ) -> Result { + if !is_valid_period(period) { + return Err(anyhow!("period must be YYYY or YYYY-MM (monthly floor)")); + } + let relay = self + .config + .node + .relays + .first() + .ok_or_else(|| anyhow!("no relays configured"))? + .clone(); + let envelope = Envelope::new(&self.key, TYPE_AGGREGATE, json!({ "period": period })); + self.send_unicast(to, envelope, &relay).await?; + let key = (to.to_string(), period.to_string()); + let deadline = Instant::now() + + Duration::from_millis(timeout_ms.unwrap_or(self.config.query.timeout_ms)); + loop { + { + let pending = self + .pending_aggregates + .lock() + .expect("aggregate reply lock"); + if let Some(body) = pending.get(&key) { + return Ok(body.clone()); + } + } + if Instant::now() >= deadline { + return Err(anyhow!("aggregate request timed out")); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + } + + async fn serve_aggregate(&self, period: &str, requester: &str, relay: &str) { + let body = self.aggregate_for(period, Some(requester)); + let Ok(value) = serde_json::to_value(&body) else { + return; + }; + let envelope = Envelope::new(&self.key, TYPE_AGGREGATE, value); + if let Err(error) = self.send_unicast(requester, envelope, relay).await { + eprintln!("aggregate reply failed: {error}"); + } + } + + async fn send_unicast(&self, to: &str, envelope: Envelope, relay: &str) -> Result<()> { let mut relays = vec![relay.to_string()]; for configured in &self.config.node.relays { if !relays.contains(configured) { @@ -295,16 +488,12 @@ impl Node { } let mut last_error: Option = None; for candidate in relays { - let url = format!( - "{}/v1/unicast?to={}", - candidate.trim_end_matches('/'), - querier - ); + let url = format!("{}/v1/unicast?to={}", candidate.trim_end_matches('/'), to); match self.client.post(&url).json(&envelope).send().await { Ok(response) if response.status().is_success() => return Ok(()), Ok(response) => { last_error = Some(anyhow!( - "relay {candidate} rejected response: {}", + "relay {candidate} rejected message: {}", response.status() )) } @@ -351,6 +540,8 @@ pub fn router(node: Arc) -> Router { Router::new() .route("/v1/local/query", post(local_query)) .route("/v1/local/status", get(local_status)) + .route("/v1/local/aggregates", get(local_aggregates)) + .route("/v1/local/aggregate/request", post(local_aggregate_request)) .with_state(node) } @@ -390,19 +581,94 @@ async fn local_query( } async fn local_status(State(node): State>) -> Response { + let aggregates = node.aggregate_for(¤t_period(), None); Json(json!({ "name": node.config.node.name, "pubkey": node.key.public_hex(), "listen": node.config.node.listen, "relays": node.config.node.relays, "responder": node.config.node.responder, + "members": node.members.read().expect("members lock").len(), "doc_count": node.doc_count(), "sent": node.sent(), "received": node.received(), + "aggregates": aggregates, })) .into_response() } +#[derive(Deserialize)] +struct AggregateParams { + period: Option, +} + +async fn local_aggregates( + State(node): State>, + Query(params): Query, +) -> Response { + let period = params.period.unwrap_or_else(current_period); + if !is_valid_period(&period) { + return ( + StatusCode::BAD_REQUEST, + Json(json!({ "error": "period must be YYYY or YYYY-MM (monthly floor)" })), + ) + .into_response(); + } + Json(node.aggregate_for(&period, None)).into_response() +} + +#[derive(Deserialize)] +struct AggregateRequest { + to: String, + period: String, + timeout_ms: Option, +} + +async fn local_aggregate_request( + State(node): State>, + Json(request): Json, +) -> Response { + match node + .request_aggregate(&request.to, &request.period, request.timeout_ms) + .await + { + Ok(body) => Json(body).into_response(), + Err(error) => ( + StatusCode::BAD_GATEWAY, + Json(json!({ "error": error.to_string() })), + ) + .into_response(), + } +} + +pub async fn control_aggregate_request( + base: &str, + to: &str, + period: &str, + timeout_ms: Option, +) -> Result { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(30)) + .build()?; + let url = format!("{}/v1/local/aggregate/request", base.trim_end_matches('/')); + let response = client + .post(&url) + .json(&json!({ + "to": to, + "period": period, + "timeout_ms": timeout_ms, + })) + .send() + .await + .with_context(|| format!("calling local node at {url} (is `frxd serve` running?)"))?; + let status = response.status(); + let value: Value = response.json().await.context("parsing node response")?; + if !status.is_success() { + return Err(anyhow!("local node error {status}: {value}")); + } + Ok(value) +} + pub async fn control_query( base: &str, text: &str, diff --git a/tests/aggregates.rs b/tests/aggregates.rs new file mode 100644 index 0000000..f9d8875 --- /dev/null +++ b/tests/aggregates.rs @@ -0,0 +1,275 @@ +mod common; + +use std::fs; +use std::path::Path; + +use common::{ask, client, collection, config_for, spawn_relay}; +use frxd::config::{CLASS_ENRICHMENT, CLASS_SOURCE, Member, save_members}; +use frxd::index::LocalIndex; +use frxd::message::EXPOSURE_FULL; +use frxd::node::{self, Node, NodeHandle, current_period}; +use serde_json::Value; + +fn corpus_dir(root: &Path, name: &str, files: &[(&str, &str)]) -> std::path::PathBuf { + let dir = root.join(format!("{name}-docs")); + fs::create_dir_all(&dir).unwrap(); + for (file, text) in files { + fs::write(dir.join(file), text).unwrap(); + } + dir +} + +async fn start_node( + root: &Path, + name: &str, + relay_url: &str, + exposure: &str, + files: &[(&str, &str)], +) -> NodeHandle { + let docs = corpus_dir(root, name, files); + let config = config_for(&root.join(name), name, relay_url); + { + let index = LocalIndex::open(&config.index_dir()).unwrap(); + index + .add_collection(&collection("docs", &docs, true, exposure)) + .unwrap(); + } + Node::start(config).await.unwrap() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn aggregates_count_sent_and_passed_per_member() { + let root = tempfile::tempdir().unwrap(); + let relay_url = spawn_relay().await; + let _bob = start_node( + root.path(), + "bob", + &relay_url, + EXPOSURE_FULL, + &[("doc.txt", "aggregate rust document")], + ) + .await; + let alice = start_node( + root.path(), + "alice", + &relay_url, + EXPOSURE_FULL, + &[("mine.txt", "alice rust note")], + ) + .await; + + let (_status, _raw, value) = ask(&client(), &alice.addr.to_string(), "rust", 5).await; + assert_eq!( + value + .get("responses") + .and_then(Value::as_array) + .unwrap() + .len(), + 1 + ); + + let period = current_period(); + let aggregate = node::control_aggregate_request( + &format!("http://{}", _bob.addr), + &alice.pubkey, + &period, + Some(700), + ) + .await + .unwrap(); + assert_eq!(aggregate.get("sent").and_then(Value::as_u64), Some(1)); + assert_eq!(aggregate.get("passed").and_then(Value::as_u64), Some(1)); + assert!(aggregate.get("cited").is_none()); + assert_eq!( + aggregate.get("period").and_then(Value::as_str), + Some(period.as_str()) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn aggregate_rollup_for_year_sums_months() { + let root = tempfile::tempdir().unwrap(); + let relay_url = spawn_relay().await; + let bob = start_node( + root.path(), + "bob", + &relay_url, + EXPOSURE_FULL, + &[("doc.txt", "rollup rust document")], + ) + .await; + let alice = start_node( + root.path(), + "alice", + &relay_url, + EXPOSURE_FULL, + &[("mine.txt", "alice rust note")], + ) + .await; + + let (_status, _raw, _value) = ask(&client(), &alice.addr.to_string(), "rust", 5).await; + let year = current_period()[0..4].to_string(); + let aggregate = node::control_aggregate_request( + &format!("http://{}", bob.addr), + &alice.pubkey, + &year, + Some(700), + ) + .await + .unwrap(); + assert_eq!(aggregate.get("sent").and_then(Value::as_u64), Some(1)); + assert_eq!( + aggregate.get("period").and_then(Value::as_str), + Some(year.as_str()) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn aggregate_floor_rejects_finer_than_month() { + let root = tempfile::tempdir().unwrap(); + let alice = start_node( + root.path(), + "alice", + &spawn_relay().await, + EXPOSURE_FULL, + &[("mine.txt", "alice rust note")], + ) + .await; + let http = client(); + + let response = http + .get(format!( + "http://{}/v1/local/aggregates?period=2026-09-15", + alice.addr + )) + .send() + .await + .unwrap(); + assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); + + let local = http + .get(format!("http://{}/v1/local/aggregates", alice.addr)) + .send() + .await + .unwrap(); + assert!(local.status().is_success()); + + let config = alice.node.config.clone(); + let direct = alice + .node + .request_aggregate(&config.node.relays[0], "2026-09-15", Some(50)) + .await; + assert!(direct.is_err()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn enrichment_members_cannot_send_content() { + let root = tempfile::tempdir().unwrap(); + let relay_url = spawn_relay().await; + let bob_full = start_node( + root.path(), + "bob", + &relay_url, + EXPOSURE_FULL, + &[("doc.txt", "enrichment test rust content")], + ) + .await; + let carol_meta = start_node( + root.path(), + "carol", + &relay_url, + "metadata", + &[("doc.txt", "enrichment test rust metadata")], + ) + .await; + + let alice = start_node( + root.path(), + "alice", + &relay_url, + EXPOSURE_FULL, + &[("mine.txt", "alice local")], + ) + .await; + save_members( + &alice.node.config.members_path(), + &[ + Member { + name: "bob".to_string(), + pubkey: bob_full.pubkey.clone(), + class: CLASS_ENRICHMENT.to_string(), + }, + Member { + name: "carol".to_string(), + pubkey: carol_meta.pubkey.clone(), + class: CLASS_ENRICHMENT.to_string(), + }, + ], + ) + .unwrap(); + + let (_status, _raw, value) = ask(&client(), &alice.addr.to_string(), "rust", 5).await; + let responses = value.get("responses").and_then(Value::as_array).unwrap(); + assert_eq!( + responses.len(), + 1, + "full-exposure enrichment reply must be dropped" + ); + assert_eq!( + responses[0].get("member").and_then(Value::as_str), + Some(carol_meta.pubkey.as_str()) + ); + assert!( + responses[0] + .get("results") + .and_then(Value::as_array) + .unwrap()[0] + .get("content") + .unwrap() + .is_null() + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn source_members_may_send_content() { + let root = tempfile::tempdir().unwrap(); + let relay_url = spawn_relay().await; + let bob = start_node( + root.path(), + "bob", + &relay_url, + EXPOSURE_FULL, + &[("doc.txt", "source test rust content")], + ) + .await; + let alice = start_node( + root.path(), + "alice", + &relay_url, + EXPOSURE_FULL, + &[("mine.txt", "alice local")], + ) + .await; + save_members( + &alice.node.config.members_path(), + &[Member { + name: "bob".to_string(), + pubkey: bob.pubkey.clone(), + class: CLASS_SOURCE.to_string(), + }], + ) + .unwrap(); + + let (_status, _raw, value) = ask(&client(), &alice.addr.to_string(), "rust", 5).await; + let responses = value.get("responses").and_then(Value::as_array).unwrap(); + assert_eq!(responses.len(), 1); + assert!( + responses[0] + .get("results") + .and_then(Value::as_array) + .unwrap()[0] + .get("content") + .unwrap() + .is_string() + ); +} diff --git a/tests/cli.rs b/tests/cli.rs index 35c10f9..7805a6b 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -125,6 +125,25 @@ fn cli_init_add_search_status() { assert_eq!(mode, 0o600, "key file must not be world readable"); } + let stdout = run_ok(&[ + "--config".to_string(), + config_arg.clone(), + "member".to_string(), + "add".to_string(), + "carol".to_string(), + "ab".repeat(32), + "--class".to_string(), + "enrichment".to_string(), + ]); + assert!(stdout.contains("carol")); + let stdout = run_ok(&[ + "--config".to_string(), + config_arg.clone(), + "member".to_string(), + "list".to_string(), + ]); + assert!(stdout.contains("carol [enrichment]")); + let stdout = run_ok(&add_args(&config, &docs, "docs", true)); assert!(stdout.contains("indexed 1 file(s)")); diff --git a/tests/common/mod.rs b/tests/common/mod.rs index f07dfdf..4c7b497 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -15,7 +15,6 @@ pub fn config_for(dir: &Path, name: &str, relay_url: &str) -> Config { name: name.to_string(), listen: "127.0.0.1:0".to_string(), relays: vec![relay_url.to_string()], - trusted_keys: Vec::new(), responder: true, }, query: QuerySection { diff --git a/tests/conformance.rs b/tests/conformance.rs index 48888fa..4219632 100644 --- a/tests/conformance.rs +++ b/tests/conformance.rs @@ -116,7 +116,7 @@ fn envelope_field_set_is_fixed() { } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn ordering_travels_scores_dont() { +async fn scores_never_travel_and_selection_is_local() { let root = tempfile::tempdir().unwrap(); let relay_url = spawn_relay().await; let _bob = start_node_with_corpus( @@ -126,8 +126,8 @@ async fn ordering_travels_scores_dont() { true, EXPOSURE_FULL, &[ - ("strong.txt", "rust rust rust rust search engine"), - ("weak.txt", "rust notes"), + ("alpha.txt", "rust search engine document"), + ("beta.txt", "rust notes"), ], ) .await; @@ -135,20 +135,23 @@ async fn ordering_travels_scores_dont() { let http = client(); let responses = raw_query(&http, &relay_url, &alice, "rust", 700).await; assert_eq!(responses.len(), 1); - let results = responses[0] - .pointer("/body/results") - .and_then(Value::as_array) - .unwrap(); + let body = responses[0].pointer("/body").unwrap(); + let results = body.pointer("/results").and_then(Value::as_array).unwrap(); assert_eq!(results.len(), 2); - let first_url = results[0].get("url").and_then(Value::as_str).unwrap(); - assert!( - first_url.contains("strong"), - "best match should lead the ordering: {first_url}" - ); + let urls: BTreeSet<&str> = results + .iter() + .map(|result| result.get("url").and_then(Value::as_str).unwrap()) + .collect(); + assert!(urls.iter().any(|url| url.contains("alpha"))); + assert!(urls.iter().any(|url| url.contains("beta"))); for result in results { assert!(!keys(result).contains("score")); assert!(!keys(result).contains("rank")); } + assert_eq!( + body.pointer("/truncated").and_then(Value::as_bool), + Some(false) + ); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] @@ -350,15 +353,23 @@ async fn entities_are_optional_hints() { } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn trusted_keys_allowlist_filters_senders() { +async fn member_directory_filters_senders() { let root = tempfile::tempdir().unwrap(); let relay_url = spawn_relay().await; let alice = Keypair::generate(); let untrusted = Keypair::generate(); let docs = corpus_dir(root.path(), "bob", &[("doc.txt", "rust document")]); - let mut config = config_for(&root.path().join("bob"), "bob", &relay_url); - config.node.trusted_keys = vec![alice.public_hex()]; + let config = config_for(&root.path().join("bob"), "bob", &relay_url); + frxd::config::save_members( + &config.members_path(), + &[frxd::config::Member { + name: "alice".to_string(), + pubkey: alice.public_hex(), + class: frxd::config::CLASS_SOURCE.to_string(), + }], + ) + .unwrap(); { let index = LocalIndex::open(&config.index_dir()).unwrap(); index @@ -718,7 +729,7 @@ async fn aggregates_are_unicast_only_and_ignored_by_nodes() { let broadcast = Envelope::new( &sender, TYPE_AGGREGATE, - json!({"period": "2026-03", "sent": 1, "passed": 0, "cited": 0}), + json!({"period": "2026-03", "sent": 1, "passed": 0}), ); assert_eq!( publish(&http, &relay_url, &broadcast).await.status(), @@ -729,7 +740,7 @@ async fn aggregates_are_unicast_only_and_ignored_by_nodes() { let unicast_envelope = Envelope::new( &sender, TYPE_AGGREGATE, - json!({"period": "2026-03", "sent": 1, "passed": 0, "cited": 0}), + json!({"period": "2026-03", "sent": 1, "passed": 0}), ); assert_eq!( unicast(&http, &relay_url, &bob.pubkey, &unicast_envelope) @@ -874,7 +885,7 @@ fn canonical_signing_bytes_are_stable() { assert_eq!(first, second); assert_eq!( String::from_utf8(first).unwrap(), - "FRX/0.3\nquery\naaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n1700000000\n00112233445566778899aabbccddeeff\n{\"budget\":{\"max_results\":5},\"entities\":[],\"qid\":\"q1\",\"text\":\"rust\"}" + "FRX/0.4\nquery\naaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n1700000000\n00112233445566778899aabbccddeeff\n{\"budget\":{\"max_results\":5},\"entities\":[],\"qid\":\"q1\",\"text\":\"rust\"}" ); } diff --git a/tests/purges.rs b/tests/purges.rs index 9122c3c..0ea58c8 100644 --- a/tests/purges.rs +++ b/tests/purges.rs @@ -57,6 +57,70 @@ fn no_bounty_winner_selection_or_slashing() { } } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn no_in_protocol_citation_accounting_or_settlement() { + let relay_url = spawn_relay().await; + let http = client(); + let key = Keypair::generate(); + + for msg_type in ["receipt", "settlement", "citation", "invoice"] { + let envelope = Envelope::new(&key, msg_type, serde_json::json!({})); + assert_eq!( + publish(&http, &relay_url, &envelope).await.status(), + reqwest::StatusCode::BAD_REQUEST, + "relay accepted {msg_type} on the broadcast channel" + ); + assert_eq!( + unicast(&http, &relay_url, &key.public_hex(), &envelope) + .await + .status(), + reqwest::StatusCode::BAD_REQUEST, + "relay accepted unicast {msg_type}" + ); + } + + let response = serde_json::to_value(build_response("q1", Vec::new(), 0, 5)).unwrap(); + assert_absent_fields( + &response, + &[ + "receipt", + "settlement", + "citation", + "cited", + "price", + "payment", + "paid", + ], + ); + let aggregate = serde_json::to_value(AggregateBody { + period: "2026-03".to_string(), + sent: 0, + passed: 0, + }) + .unwrap(); + assert_absent_fields( + &aggregate, + &[ + "receipt", + "settlement", + "citation", + "cited", + "price", + "payment", + "paid", + ], + ); + + for route in ["/v1/receipt", "/v1/settlement", "/v1/invoice"] { + let response = http + .get(format!("{relay_url}{route}")) + .send() + .await + .unwrap(); + assert_eq!(response.status(), reqwest::StatusCode::NOT_FOUND); + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn no_protocol_query_dedup() { let relay_url = spawn_relay().await; @@ -190,7 +254,6 @@ async fn no_aggregate_appeals() { period: "2026-03".to_string(), sent: 1, passed: 1, - cited: 0, }) .unwrap(), ); @@ -205,10 +268,9 @@ async fn no_aggregate_appeals() { period: "2026-03".to_string(), sent: 1, passed: 1, - cited: 0, }) .unwrap(); - assert_exact_keys(&body, &["period", "sent", "passed", "cited"]); + assert_exact_keys(&body, &["period", "sent", "passed"]); assert_absent_fields(&body, &["appeal", "dispute", "complaint", "sanction"]); for route in ["/v1/appeal", "/v1/dispute"] {