diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f40170..1fe0985 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 maximum-timestamp markers, immediate query and rebroadcast suppression, recipient gift-wrap cleanup, undeletable request records, and bounded restart-safe physical deletion. +- Mergeable NIP-45 HyperLogLog sketches for canonical single-target COUNT + filters, including hex, address, and arbitrary-string offset derivation. ### Fixed diff --git a/README.md b/README.md index 068c2c9..515eb7d 100644 --- a/README.md +++ b/README.md @@ -24,9 +24,9 @@ quirk. It also provides an additional Unix-domain socket transport. - **Nostr-first protocol behavior.** EVENT/REQ/CLOSE/COUNT/EOSE/OK/NOTICE/ CLOSED/AUTH and NEG-* are tested against pinned NIPs. Differential tests are migration and regression evidence, not a promise to retain upstream bugs. -- **NIP-42 AUTH, NIP-45 COUNT, NIP-50 ranked content search, NIP-62 - restart-safe Request to Vanish, NIP-70 protected events, NIP-59 gift-wrap - deletion semantics, NIP-77 negentropy set +- **NIP-42 AUTH, NIP-45 COUNT with mergeable HyperLogLog sketches, NIP-50 + ranked content search, NIP-62 restart-safe Request to Vanish, NIP-70 + protected events, NIP-59 gift-wrap deletion semantics, NIP-77 negentropy set reconciliation** (persistent LMDB B-tree, tree-backed multi-round sync sessions). - **Standards-first ephemeral delivery**: ephemeral kinds are live-only by diff --git a/crates/wok-compat/tests/e2e_transports.rs b/crates/wok-compat/tests/e2e_transports.rs index 6dfa9c3..0cd299b 100644 --- a/crates/wok-compat/tests/e2e_transports.rs +++ b/crates/wok-compat/tests/e2e_transports.rs @@ -8,6 +8,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio_tungstenite::tungstenite::Message; use wok_compat::{sign_event, sign_event_with_key}; use wok_db::{Env, EnvOptions}; +use wok_query::{HyperLogLog, NostrFilter}; use wok_relay::Config; fn now_secs() -> u64 { @@ -238,6 +239,132 @@ async fn nip62_vanish_is_immediate_and_blocks_rebroadcast() { handle.request_shutdown(); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn nip45_count_returns_mergeable_hll_for_canonical_tag_query() { + let dir = tempfile::tempdir().unwrap(); + let env = Env::open(dir.path(), EnvOptions::default()).unwrap(); + env.ensure_initialized().unwrap(); + let cfg = test_cfg(dir.path()); + let handle = wok_relay::start(env, cfg).unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let relay = handle.clone(); + tokio::spawn(async move { + let _ = wok_ws::serve_listener(relay, listener).await; + }); + tokio::time::sleep(Duration::from_millis(50)).await; + + let (mut ws, _) = tokio_tungstenite::connect_async(format!("ws://{addr}/")) + .await + .unwrap(); + let keys: Vec = { + let mut rng = rand::thread_rng(); + (0..4).map(|_| Keypair::new(SECP256K1, &mut rng)).collect() + }; + let target = sign_event(json!({ + "created_at": now_secs() - 20, + "kind": 1, + "tags": [], + "content": "hll-target", + })); + let target_id = target["id"].as_str().unwrap().to_string(); + ws.send(Message::Text(json!(["EVENT", target]).to_string().into())) + .await + .unwrap(); + let _ = recv_until(&mut ws, |text| text.contains("\"OK\"")).await; + + for (index, key) in keys.iter().enumerate() { + let reaction = sign_event_with_key( + json!({ + "created_at": now_secs() - 10 + index as u64, + "kind": 7, + "tags": [["e", target_id]], + "content": "+", + }), + key, + ); + ws.send(Message::Text(json!(["EVENT", reaction]).to_string().into())) + .await + .unwrap(); + let accepted = recv_until(&mut ws, |text| text.contains("\"OK\"")).await; + assert!(accepted.iter().any(|text| text.contains("true"))); + } + let repeated_author = sign_event_with_key( + json!({ + "created_at": now_secs(), + "kind": 7, + "tags": [["e", target_id]], + "content": "second reaction from one author", + }), + &keys[0], + ); + ws.send(Message::Text( + json!(["EVENT", repeated_author]).to_string().into(), + )) + .await + .unwrap(); + let accepted = recv_until(&mut ws, |text| text.contains("\"OK\"")).await; + assert!(accepted.iter().any(|text| text.contains("true"))); + + let count_filter = json!({"#e":[target_id], "kinds":[7]}); + ws.send(Message::Text( + json!(["COUNT", "hll", count_filter]).to_string().into(), + )) + .await + .unwrap(); + let response = recv_until(&mut ws, |text| text.contains("\"COUNT\"")).await; + let body = response + .iter() + .find_map(|text| { + serde_json::from_str::(text) + .ok() + .filter(|value| value[0] == "COUNT") + .map(|value| value[2].clone()) + }) + .expect("COUNT body"); + assert_eq!(body["count"], 5); + let actual = body["hll"].as_str().expect("HLL response"); + assert_eq!(actual.len(), 512); + + let parsed_filter = NostrFilter::parse(&count_filter, 500, 3).unwrap(); + let mut expected = HyperLogLog::for_filter(&parsed_filter).unwrap(); + for key in &keys { + let (pubkey, _) = key.x_only_public_key(); + expected.add_pubkey(&pubkey.serialize()); + } + assert_eq!(actual, expected.encode_hex()); + + let empty_target = "00".repeat(32); + ws.send(Message::Text( + json!(["COUNT", "empty-hll", {"#e":[empty_target], "kinds":[7]}]) + .to_string() + .into(), + )) + .await + .unwrap(); + let empty = recv_until(&mut ws, |text| text.contains("empty-hll")).await; + assert!(empty + .iter() + .any(|text| text.contains(&format!("\"hll\":\"{}\"", "00".repeat(256))))); + + ws.send(Message::Text( + json!(["COUNT", "ambiguous-hll", { + "#e":[target_id], "#p":["11".repeat(32)], "kinds":[7] + }]) + .to_string() + .into(), + )) + .await + .unwrap(); + let ambiguous = recv_until(&mut ws, |text| text.contains("ambiguous-hll")).await; + assert!(ambiguous + .iter() + .filter(|text| text.contains("\"COUNT\"")) + .all(|text| !text.contains("\"hll\""))); + + handle.request_shutdown(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn nip50_ranked_historical_and_live_search() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/wok-compat/tests/nip_conformance.rs b/crates/wok-compat/tests/nip_conformance.rs index 90d4842..25a86d6 100644 --- a/crates/wok-compat/tests/nip_conformance.rs +++ b/crates/wok-compat/tests/nip_conformance.rs @@ -134,10 +134,21 @@ fn nip45_count_encoding() { sub_id: "c".into(), count: 3, limited: true, + hll: Some("00".repeat(256)), } .to_json(); assert!(s.contains("\"count\":3")); assert!(s.contains("\"limited\":true")); + assert!(!s.contains("\"hll\"")); + + let s = RelayMessage::Count { + sub_id: "c".into(), + count: 0, + limited: false, + hll: Some("00".repeat(256)), + } + .to_json(); + assert!(s.contains("\"hll\"")); } #[test] diff --git a/crates/wok-query/src/hll.rs b/crates/wok-query/src/hll.rs new file mode 100644 index 0000000..2934aba --- /dev/null +++ b/crates/wok-query/src/hll.rs @@ -0,0 +1,145 @@ +//! NIP-45 HyperLogLog registers for mergeable COUNT responses. + +use crate::NostrFilter; + +/// NIP-45 fixes the precision at 8 bits: 256 one-byte registers. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HyperLogLog { + offset: usize, + registers: [u8; 256], +} + +impl HyperLogLog { + pub fn for_filter(filter: &NostrFilter) -> Option { + Some(Self { + offset: offset_for_filter(filter)?, + registers: [0; 256], + }) + } + + pub fn add_pubkey(&mut self, pubkey: &[u8]) { + let Some(window) = pubkey + .get(self.offset..self.offset.saturating_add(8)) + .filter(|window| window.len() == 8) + else { + return; + }; + let register = window[0] as usize; + let mut value = 0u64; + for byte in &window[1..] { + value = (value << 8) | u64::from(*byte); + } + // `value` occupies the low 56 bits. Remove the eight padding zeroes + // counted by u64::leading_zeros, then add one as specified by NIP-45. + let rank = value.leading_zeros().saturating_sub(8).saturating_add(1) as u8; + self.registers[register] = self.registers[register].max(rank); + } + + pub fn encode_hex(&self) -> String { + const HEX: &[u8; 16] = b"0123456789abcdef"; + let mut encoded = String::with_capacity(512); + for register in self.registers { + encoded.push(HEX[(register >> 4) as usize] as char); + encoded.push(HEX[(register & 0x0f) as usize] as char); + } + encoded + } + + #[cfg(test)] + fn registers(&self) -> &[u8; 256] { + &self.registers + } +} + +/// Canonical HLL requests count one target. Multiple tag names or target +/// values have ambiguous merge semantics, so callers deliberately omit HLL +/// for those shapes. +pub fn offset_for_filter(filter: &NostrFilter) -> Option { + if filter.tags.len() != 1 { + return None; + } + let (tag, values) = filter.tags.first_key_value()?; + if values.size() != 1 { + return None; + } + let value = values.at(0); + let seed: [u8; 32] = if matches!(tag, 'e' | 'p') { + value.try_into().ok()? + } else if value.len() == 64 { + match std::str::from_utf8(value) + .ok() + .and_then(|hex| wok_event::from_lower_hex_exact(hex).ok()) + { + Some(bytes) => bytes.try_into().ok()?, + None => wok_event::sha256(value), + } + } else if let Some(pubkey) = std::str::from_utf8(value) + .ok() + .and_then(|address| wok_event::parse_a_tag(address).ok()) + .map(|(_, pubkey, _)| pubkey) + { + pubkey + } else { + wok_event::sha256(value) + }; + Some(usize::from(seed[16] >> 4) + 8) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn filter(value: serde_json::Value) -> NostrFilter { + NostrFilter::parse(&value, 500, 3).unwrap() + } + + #[test] + fn offset_covers_hex_address_and_hashed_values() { + let hex = format!("{}f{}", "0".repeat(32), "0".repeat(31)); + assert_eq!(offset_for_filter(&filter(json!({"#e":[hex]}))), Some(23)); + + let pubkey = format!("{}a{}", "0".repeat(32), "0".repeat(31)); + let address = format!("30023:{pubkey}:profile"); + assert_eq!( + offset_for_filter(&filter(json!({"#a":[address]}))), + Some(18) + ); + + let seed = wok_event::sha256(b"arbitrary-target"); + assert_eq!( + offset_for_filter(&filter(json!({"#t":["arbitrary-target"]}))), + Some(usize::from(seed[16] >> 4) + 8) + ); + } + + #[test] + fn sketch_matches_the_fixed_precision_bit_rules() { + let mut hll = HyperLogLog { + offset: 8, + registers: [0; 256], + }; + let mut pubkey = [0u8; 32]; + hll.add_pubkey(&pubkey); + assert_eq!(hll.registers()[0], 57); + + pubkey[8] = 7; + pubkey[9] = 0x10; + hll.add_pubkey(&pubkey); + assert_eq!(hll.registers()[7], 4); + assert_eq!(hll.encode_hex().len(), 512); + } + + #[test] + fn ambiguous_filter_shapes_do_not_get_a_sketch() { + assert!(offset_for_filter(&filter(json!({"kinds":[7]}))).is_none()); + assert!( + offset_for_filter(&filter(json!({"#e":["00".repeat(32), "11".repeat(32)]}))).is_none() + ); + assert!(offset_for_filter(&filter(json!({ + "#e":["00".repeat(32)], + "#p":["11".repeat(32)] + }))) + .is_none()); + } +} diff --git a/crates/wok-query/src/lib.rs b/crates/wok-query/src/lib.rs index b23d386..3acd6b2 100644 --- a/crates/wok-query/src/lib.rs +++ b/crates/wok-query/src/lib.rs @@ -1,10 +1,12 @@ pub mod filter; +pub mod hll; pub mod monitor; pub mod scan; pub mod scheduler; pub mod subid; pub use filter::{dumb_match, FilterValidator, NostrFilter, NostrFilterGroup}; +pub use hll::{offset_for_filter as nip45_hll_offset, HyperLogLog}; pub use monitor::{ActiveMonitors, Recipient}; pub use scan::{foreach_by_filter, DbQuery, DbScan}; pub use scheduler::QueryScheduler; diff --git a/crates/wok-query/src/scan.rs b/crates/wok-query/src/scan.rs index d1c9895..21bb8b4 100644 --- a/crates/wok-query/src/scan.rs +++ b/crates/wok-query/src/scan.rs @@ -11,6 +11,8 @@ use wok_db::{ }; use wok_event::PackedEventView; +use crate::HyperLogLog; + #[derive(Clone, Debug)] struct CandidateEvent { packed: u64, @@ -679,11 +681,17 @@ pub struct DbQuery { sent_events_curr: HashSet, last_work_checked: u64, max_total_events: u64, + hll: Option, } impl DbQuery { pub fn new(sub: Subscription, max_total_events: u64) -> Self { let max_total_events = if sub.count_only { 0 } else { max_total_events }; + let hll = if sub.count_only && sub.filter_group.filters.len() == 1 { + HyperLogLog::for_filter(&sub.filter_group.filters[0]) + } else { + None + }; let all_filters_search = !sub.filter_group.filters.is_empty() && sub .filter_group @@ -704,6 +712,7 @@ impl DbQuery { // max_filter_limit_count. The request-wide delivery ceiling is // for EVENT responses. max_total_events, + hll, } } @@ -711,12 +720,24 @@ impl DbQuery { self.sent_events_full.len() as u64 } - fn event_is_visible(txn: &RoTxn<'_>, lev_id: u64) -> Result { + pub fn hll_hex(&self) -> Option { + self.hll.as_ref().map(HyperLogLog::encode_hex) + } + + fn visible_event_pubkey( + txn: &RoTxn<'_>, + lev_id: u64, + ) -> Result, wok_db::DbError> { let Some(raw) = txn.get_u64(txn.env().dbis().event, lev_id)? else { - return Ok(false); + return Ok(None); }; let packed = PackedEventView::new(raw)?; - Ok(!is_event_vanished_ro(txn, packed)?) + if is_event_vanished_ro(txn, packed)? { + return Ok(None); + } + let mut pubkey = [0u8; 32]; + pubkey.copy_from_slice(packed.pubkey()); + Ok(Some(pubkey)) } /// Returns true when the scan is complete. @@ -743,18 +764,18 @@ impl DbQuery { self.sub.latest_event_id, time_budget_us, |lev_id| { - match Self::event_is_visible(txn, lev_id) { - Ok(true) => {} - Ok(false) => return, + let pubkey = match Self::visible_event_pubkey(txn, lev_id) { + Ok(Some(pubkey)) => pubkey, + Ok(None) => return, Err(error) => { visibility_error = Some(error); return; } - } + }; if (self.max_total_events == 0 || (sent.len() as u64) < self.max_total_events) && sent.insert(lev_id) { - hits.push(lev_id); + hits.push((lev_id, pubkey)); } }, )?; @@ -762,7 +783,10 @@ impl DbQuery { return Err(error); } self.sent_events_full = sent; - for lev_id in hits { + for (lev_id, pubkey) in hits { + if let Some(hll) = &mut self.hll { + hll.add_pubkey(&pubkey); + } cb(&self.sub, lev_id); } if self.max_total_events != 0 @@ -789,7 +813,7 @@ impl DbQuery { let mut sent_full = std::mem::take(&mut self.sent_events_full); let mut sent_curr = std::mem::take(&mut self.sent_events_curr); let mut last_work = self.last_work_checked; - let mut hits: Vec = Vec::new(); + let mut hits: Vec<(u64, [u8; 32])> = Vec::new(); let mut visibility_error = None; let mut handle = |lev_id| { if f.limit == 0 { @@ -798,16 +822,16 @@ impl DbQuery { if lev_id > latest { return false; } - match Self::event_is_visible(txn, lev_id) { - Ok(true) => {} - Ok(false) => return false, + let pubkey = match Self::visible_event_pubkey(txn, lev_id) { + Ok(Some(pubkey)) => pubkey, + Ok(None) => return false, Err(error) => { visibility_error = Some(error); return true; } - } + }; if sent_full.insert(lev_id) { - hits.push(lev_id); + hits.push((lev_id, pubkey)); } sent_curr.insert(lev_id); sent_curr.len() as u64 >= f.limit @@ -835,7 +859,10 @@ impl DbQuery { self.sent_events_full = sent_full; self.sent_events_curr = sent_curr; self.last_work_checked = last_work; - for lev in hits { + for (lev, pubkey) in hits { + if let Some(hll) = &mut self.hll { + hll.add_pubkey(&pubkey); + } cb(&self.sub, lev); } if self.max_total_events != 0 diff --git a/crates/wok-query/src/scheduler.rs b/crates/wok-query/src/scheduler.rs index 413e1b0..bd18d13 100644 --- a/crates/wok-query/src/scheduler.rs +++ b/crates/wok-query/src/scheduler.rs @@ -86,7 +86,7 @@ impl QueryScheduler { ) -> Result<(), wok_db::DbError> where F: FnMut(&Subscription, u64, Option<&[u8]>), - C: FnMut(&Subscription, u64), + C: FnMut(&Subscription, u64, Option), { let Some(idx) = self.running.pop_front() else { return Ok(()); @@ -131,7 +131,7 @@ impl QueryScheduler { let q = self.queries[idx].take().unwrap(); self.free.push(idx); self.remove_sub(q.sub.conn_id, &q.sub.sub_id); - on_complete(&q.sub, q.sent_count()); + on_complete(&q.sub, q.sent_count(), q.hll_hex()); } else { self.running.push_back(idx); } diff --git a/crates/wok-query/tests/scan_kinds.rs b/crates/wok-query/tests/scan_kinds.rs index 740de31..36a01d8 100644 --- a/crates/wok-query/tests/scan_kinds.rs +++ b/crates/wok-query/tests/scan_kinds.rs @@ -179,7 +179,7 @@ fn deep_author_kind_pages_are_complete_and_non_overlapping() { &txn, 10_000, |_, lev, _| scheduled.push(lev), - |_, total| completions.push(total), + |_, total, _hll| completions.push(total), ) .unwrap(); } diff --git a/crates/wok-relay/src/protocol.rs b/crates/wok-relay/src/protocol.rs index af9e058..cc341e7 100644 --- a/crates/wok-relay/src/protocol.rs +++ b/crates/wok-relay/src/protocol.rs @@ -55,6 +55,7 @@ pub enum RelayMessage { sub_id: String, count: u64, limited: bool, + hll: Option, }, Notice { message: String, @@ -221,11 +222,15 @@ impl RelayMessage { sub_id, count, limited, + hll, } => { let mut body = json!({ "count": count }); if *limited { body["limited"] = json!(true); } + if let (false, Some(hll)) = (*limited, hll) { + body["hll"] = json!(hll); + } json!(["COUNT", sub_id, body]).to_string() } Self::Notice { message } => json!(["NOTICE", message]).to_string(), diff --git a/crates/wok-relay/src/server.rs b/crates/wok-relay/src/server.rs index 7eadf33..9ff5e07 100644 --- a/crates/wok-relay/src/server.rs +++ b/crates/wok-relay/src/server.rs @@ -1920,7 +1920,7 @@ fn run_req_worker( } } } - let mut completed: Vec<(Subscription, u64)> = Vec::new(); + let mut completed: Vec<(Subscription, u64, Option)> = Vec::new(); // Events are framed and delivered inside the scan callback: no // per-event Subscription clone, no intermediate collection, and the // payload JSON is copied exactly once (into the frame). @@ -1946,12 +1946,12 @@ fn run_req_worker( } } }, - |sub, total| { - completed.push((sub.clone(), total)); + |sub, total, hll| { + completed.push((sub.clone(), total, hll)); }, ); drop(txn); - for (sub, total) in completed { + for (sub, total, hll) in completed { if sub.count_only { let mut count = total; let mut limited = false; @@ -1965,6 +1965,7 @@ fn run_req_worker( sub_id: sub.sub_id.to_string(), count, limited, + hll: if limited { None } else { hll }, }, &metrics, ); @@ -2368,7 +2369,7 @@ fn run_negentropy( |sub, lev, _| { lev_hits.push((sub.clone(), lev)); }, - |sub, total| { + |sub, total, _hll| { done.push((sub.clone(), total)); }, ); diff --git a/docs/known-differences.md b/docs/known-differences.md index f57280b..8cd1ddb 100644 --- a/docs/known-differences.md +++ b/docs/known-differences.md @@ -23,6 +23,9 @@ it is not a promise to reproduce upstream bugs. - NIP-11 reports Wok's repository as the software implementation. - Wok implements NIP-50 ranked content search using a Wok-owned derived LMDB index. The pinned strfry revision has no NIP-50 implementation. +- Wok returns mergeable NIP-45 HyperLogLog sketches for canonical + single-target COUNT filters, including address and hashed-string offsets; + this is not present in the pinned strfry revision. ## Intentional protocol and operational differences diff --git a/docs/nips.md b/docs/nips.md index d08e30a..fcc8321 100644 --- a/docs/nips.md +++ b/docs/nips.md @@ -15,7 +15,7 @@ arbitrary list. | 13 | Proof of work | leading-zero validation + NIP-11 minimum | relay tests | `relay.abuse.enabled` and `min_pow_difficulty > 0` | | 40 | Expiration | packed expiration + cron | `nip_conformance.rs` | always | | 42 | AUTH | ingest AUTH | unit + e2e when serviceUrl set | AUTH enabled and serviceUrl set | -| 45 | COUNT | REQ worker | `nip_conformance.rs` | `maxFilterLimitCount > 0` | +| 45 | COUNT + mergeable HyperLogLog | REQ worker + `wok-query` HLL | `nip_conformance.rs`, `e2e_transports.rs`, HLL unit vectors | `maxFilterLimitCount > 0` | | 50 | Search capability | transactional LMDB term/bigram index + ranked query scanner | `nip_conformance.rs`, `search.rs`, `e2e_transports.rs` | always | | 59 | Gift wrap | recipient-only restricted reads, recipient-authorized deletion, and live-only kind 21059 | restrict + DB/live tests | usable AUTH, restricted kind 1059 with involved-pubkey enforcement, and `events.ephemeral_persistence = "live_only"` | | 62 | Request to Vanish | persistent maximum-timestamp markers, immediate query/rebroadcast suppression, gift-wrap recipient cleanup, and bounded physical deletion | `nip62_vanish.rs`, relay e2e | `relay.nip62.enabled` | @@ -34,6 +34,12 @@ them with every other supplied filter field, ranks before applying `limit`, and supports matching live events after EOSE. See [NIP-50 search](nip50-search.md) for exact query and scoring semantics. +NIP-45 responses include a 512-character HLL register value for a single +filter containing exactly one tag attribute with one target. Offset derivation +implements all specified target forms: raw event/pubkey hex, an address's +pubkey, or SHA-256 of any other string. Multi-filter, multi-target, and limited +counts omit HLL because their sketches would be ambiguous or incomplete. + NIP-62 accepts a signed kind 62 request containing either a matching `["relay", ""]` tag or `["relay", "ALL_RELAYS"]`. The relay immediately suppresses qualifying authored events and gift wraps for the