diff --git a/README.md b/README.md index b9d45d4..733aa92 100644 --- a/README.md +++ b/README.md @@ -36,21 +36,12 @@ Enable it on TinyWeb's `/style` page under "Forum". - Threads are auto-pruned after 30 days (configurable, or set to 0 to keep everything) - Moderation is local: block authors, mute threads, keyword filters, and gossip block lists with peers (auto-block after 3 peer reports) -### Trust circle - -Content exchange is gated by a trust graph derived from your TinyWeb subscriptions: - -- Only peers you subscribe to (with forum enabled) are trusted sources of content -- Content from trusted peers propagates transitively: you see posts from their trusted peers, and theirs, and so on — no hop limit -- New identities have no trust path until a trusted peer vouches for them or you subscribe -- Trust is seeded from TinyWeb's subscription page — each subscribed peer's forum content (and their transitive trust network) enters your view automatically. - ## Moderation All moderation is local — it controls what you see: -- **Block author** — hides all content from that identity and cascades: the blocked peer's downstream trust network (peers they vouched for) is also removed from your view. If a downstream peer has an alternate trust path from another source you trust, they survive the cascade. -- **Auto-block** — when 3+ of your peers have blocked the same identity, it blocks for you too (with cascade) +- **Block author** — hides all content from that identity +- **Auto-block** — when 3+ of your peers have blocked the same identity, it blocks for you too - **Mute thread** — hides a thread from the listing - **Keyword filters** — hides threads matching keywords - **Instance sync** — choose which peers to sync with; unsync at any time @@ -80,7 +71,6 @@ All moderation is local — it controls what you see: - Retractions are voluntary — peers can ignore them - Block gossip can be gamed (requires collusion by 3+ peers on Reticulum) - Threads prune after 30 days by default -- Discovery is through the subscription graph rather than topic-based blooms - Best-effort maintenance ## Security @@ -89,4 +79,3 @@ All moderation is local — it controls what you see: - **Retractions are voluntary** — Retracting a thread or post sends a signal to peers, but any peer can ignore it. "Retract" is a polite request, not a guaranteed delete. - **Block gossip can be gamed** — Auto-block triggers after 3 peer reports. On Reticulum this requires 3+ real instances to collude, which is impractical at mesh scale, but is not cryptographically enforced. - **No rate limiting** — Forum POST endpoints have no throttling. Low risk since the HTTP port is localhost-only by default. -- **Trust circle** — Blocking a peer removes their downstream trust network from your view. A peer with multiple independent trust paths may survive a single block. diff --git a/tinyweb_forum/__init__.py b/tinyweb_forum/__init__.py index 9219d17..ffdbc9c 100644 --- a/tinyweb_forum/__init__.py +++ b/tinyweb_forum/__init__.py @@ -1,14 +1,8 @@ -import os -import sqlite3 -import threading -import time - from tinyweb_forum.db import ForumDB from tinyweb_forum.handlers import ForumHandlers from tinyweb_forum.sync import ForumSync FORUM_ENABLED_KEY = "forum_enabled" -TRUST_REFRESH_INTERVAL = 120 class ForumPlugin: @@ -22,52 +16,19 @@ class ForumPlugin: self.identity = identity self.reticulum = reticulum self._started = False - self._data_dir = data_dir - self._core_db_path = os.path.join(data_dir, "index.db") - self._trust_refresh_thread = None def is_enabled(self): return self.fdb.get_setting(FORUM_ENABLED_KEY, "0") == "1" - def _seed_trust_from_subscriptions(self): - my_hash = self.identity.hash.hex() if self.identity else "local" - if not os.path.exists(self._core_db_path): - return - try: - core = sqlite3.connect(self._core_db_path) - rows = core.execute( - "SELECT dest_hash FROM subscriptions" - ).fetchall() - core.close() - for row in rows: - self.fdb.add_trust_source(row[0], my_hash, 0) - except Exception: - pass - - def _trust_refresh_loop(self): - while self._started: - try: - self._seed_trust_from_subscriptions() - except Exception: - pass - for _ in range(TRUST_REFRESH_INTERVAL): - if not self._started: - return - time.sleep(1) - def enable(self): self.fdb.set_setting(FORUM_ENABLED_KEY, "1") if not self._started: - self._seed_trust_from_subscriptions() self.sync.start() self._started = True - self._trust_refresh_thread = threading.Thread( - target=self._trust_refresh_loop, daemon=True - ) - self._trust_refresh_thread.start() def disable(self): self.fdb.set_setting(FORUM_ENABLED_KEY, "0") + # Keep sync running so we still receive content — disable just hides UI def handle(self, method, path, query, body, cookies=None): return self.handlers.handle(method, path, query, body, cookies) diff --git a/tinyweb_forum/db.py b/tinyweb_forum/db.py index e2dc3a5..8f396d7 100644 --- a/tinyweb_forum/db.py +++ b/tinyweb_forum/db.py @@ -4,6 +4,8 @@ import threading import time from datetime import datetime, timedelta +from tinyweb_forum.bloom import BloomFilter + FORUM_DB = "forum.db" @@ -89,24 +91,20 @@ class ForumDB: " PRIMARY KEY (content_id, content_type)" ")" ) + db.execute( + "CREATE TABLE IF NOT EXISTS peer_filters (" + " peer_hash TEXT PRIMARY KEY," + " bloom_bytes BLOB," + " bloom_size INTEGER DEFAULT 2048," + " bloom_hashes INTEGER DEFAULT 3," + " tag_count INTEGER DEFAULT 0," + " last_seen REAL NOT NULL" + ")" + ) try: db.execute("ALTER TABLE synced_instances ADD COLUMN consecutive_failures INTEGER DEFAULT 0") except Exception: pass - db.execute( - "CREATE TABLE IF NOT EXISTS forum_trust_sources (" - " instance_hash TEXT NOT NULL," - " source_hash TEXT NOT NULL," - " hops INTEGER NOT NULL," - " PRIMARY KEY (instance_hash, source_hash)" - ")" - ) - db.execute( - "CREATE TABLE IF NOT EXISTS forum_trust (" - " instance_hash TEXT PRIMARY KEY," - " min_hops INTEGER NOT NULL" - ")" - ) db.commit() db.close() @@ -613,107 +611,70 @@ class ForumDB: # --- Filter Table (Architecture B: Bloom Gossip) --- - # --- Forum Trust --- - - # --- Forum Trust --- - - def add_trust_source(self, instance_hash, source_hash, hops): + def store_peer_filter(self, peer_hash, bloom_bytes, bloom_size, bloom_hashes, tag_count): db = self.get_db() try: db.execute( - "INSERT OR IGNORE INTO forum_trust_sources " - "(instance_hash, source_hash, hops) VALUES (?, ?, ?)", - (instance_hash, source_hash, hops), + "INSERT OR REPLACE INTO peer_filters " + "(peer_hash, bloom_bytes, bloom_size, bloom_hashes, tag_count, last_seen) " + "VALUES (?, ?, ?, ?, ?, ?)", + (peer_hash, bloom_bytes, bloom_size, bloom_hashes, tag_count, time.time()), ) - existing = db.execute( - "SELECT hops FROM forum_trust_sources " - "WHERE instance_hash=? AND source_hash=?", - (instance_hash, source_hash), - ).fetchone() - if existing and hops < existing["hops"]: - db.execute( - "UPDATE forum_trust_sources SET hops=? " - "WHERE instance_hash=? AND source_hash=?", - (hops, instance_hash, source_hash), - ) - self._recompute_trust_view(db) db.commit() finally: self.return_db(db) - def remove_trust_source(self, source_hash): - db = self.get_db() - try: - rows = db.execute( - "WITH RECURSIVE cascade(h) AS (" - " VALUES(?)" - " UNION ALL" - " SELECT fts.instance_hash FROM forum_trust_sources fts" - " INNER JOIN cascade c ON fts.source_hash = c.h" - ") SELECT h FROM cascade", - (source_hash,), - ).fetchall() - cascade_hashes = [r["h"] for r in rows] - placeholders = ",".join("?" for _ in cascade_hashes) - db.execute( - f"DELETE FROM forum_trust_sources WHERE source_hash IN ({placeholders})", - cascade_hashes, - ) - self._recompute_trust_view(db) - db.commit() - finally: - self.return_db(db) - - def remove_by_hash(self, instance_hash): - self.remove_trust_source(instance_hash) - db = self.get_db() - try: - db.execute( - "DELETE FROM forum_trust_sources WHERE instance_hash=?", - (instance_hash,), - ) - self._recompute_trust_view(db) - db.commit() - finally: - self.return_db(db) - - def is_trusted(self, instance_hash): + def get_peer_filter(self, peer_hash): db = self.get_db() try: row = db.execute( - "SELECT 1 FROM forum_trust WHERE instance_hash=?", (instance_hash,) + "SELECT * FROM peer_filters WHERE peer_hash = ?", (peer_hash,) ).fetchone() - return row is not None + if row: + return dict(row) + return None finally: self.return_db(db) - def get_trust_hops(self, instance_hash): + def get_all_filters(self): + """Return all stored peer filters (for filter table gossip).""" db = self.get_db() try: - row = db.execute( - "SELECT min_hops FROM forum_trust WHERE instance_hash=?", (instance_hash,) - ).fetchone() - return row["min_hops"] if row else None + return [dict(r) for r in db.execute( + "SELECT * FROM peer_filters ORDER BY last_seen DESC" + ).fetchall()] finally: self.return_db(db) - def get_all_trusted_hashes(self): + def get_filtered_peers_by_tag(self, tag): + """Return peer hashes whose bloom filter might contain the given tag.""" + tag = tag.strip().lower() db = self.get_db() try: - return [ - r["instance_hash"] for r in db.execute( - "SELECT instance_hash FROM forum_trust" - ).fetchall() - ] + matches = [] + for r in db.execute("SELECT * FROM peer_filters").fetchall(): + bf = BloomFilter.from_bytes(r["bloom_bytes"], r["bloom_size"], r["bloom_hashes"]) + if bf.might_contain(tag): + matches.append(r["peer_hash"]) + return matches finally: self.return_db(db) - def _recompute_trust_view(self, db): - db.execute("DELETE FROM forum_trust") - db.execute( - "INSERT INTO forum_trust (instance_hash, min_hops) " - "SELECT instance_hash, MIN(hops) FROM forum_trust_sources GROUP BY instance_hash" - ) + def prune_peer_filters(self, max_age_days=7): + db = self.get_db() + try: + cutoff = time.time() - max_age_days * 86400 + db.execute("DELETE FROM peer_filters WHERE last_seen < ?", (cutoff,)) + db.commit() + finally: + self.return_db(db) + + def get_peer_filter_count(self): + db = self.get_db() + try: + return db.execute("SELECT count(*) FROM peer_filters").fetchone()[0] + finally: + self.return_db(db) # --- Peer Liveness --- diff --git a/tinyweb_forum/handlers.py b/tinyweb_forum/handlers.py index cbe3097..343dd6e 100644 --- a/tinyweb_forum/handlers.py +++ b/tinyweb_forum/handlers.py @@ -190,10 +190,12 @@ class ForumHandlers: topics = self._get_subscribed_topics() topics_str = ", ".join(topics) if topics else "everything" peer_count = len(self.fdb.get_synced_instances()) + filter_count = self.fdb.get_peer_filter_count() return ( f'
' f'subscribed: {esc(topics_str)}' f' · {peer_count} peers' + f' · {filter_count} filters' f'
' ) @@ -719,7 +721,6 @@ class ForumHandlers: blocked = self._blocked_instances() blocked.add(instance) self.fdb.set_setting("blocked_instances", ",".join(blocked)) - self.fdb.remove_by_hash(instance) self._set_flash(f"Blocked {instance[:16]}...") return self._redirect("/forum/moderation") @@ -743,7 +744,6 @@ class ForumHandlers: self.fdb.clear_peer_block(instance) else: blocked.add(instance) - self.fdb.remove_by_hash(instance) self.fdb.set_setting("blocked_instances", ",".join(blocked)) auto = set(h.strip() for h in self.fdb.get_setting("auto_blocked_instances", "").split(",") if h.strip()) auto.discard(instance) @@ -839,24 +839,75 @@ class ForumHandlers: incoming_threads = data.get("threads", []) incoming_posts = data.get("posts", []) incoming_upvotes = data.get("upvotes", []) + peer_topics = data.get("my_topics", []) + peer_tag_cloud = data.get("my_tag_cloud", []) peer_bloom_data = data.get("content_bloom") peer_tag_bloom_data = data.get("tag_bloom") + peer_filter_table = data.get("filter_table", {}) + scoped_query_tag = data.get("scoped_query", "") blocked = self._blocked_instances() my_topics = self._get_subscribed_topics() from_hash = data.get("from_hash", "") - # Merge trust gossip from requester - if from_hash: - self.sync._merge_trust_gossip(data.get("trust_list", []), from_hash) + # Store peer's tag bloom filter (Architecture B discovery) + if peer_tag_bloom_data and from_hash: + self.fdb.store_peer_filter( + peer_hash=from_hash, + bloom_bytes=bytes(peer_tag_bloom_data), + bloom_size=data.get("tag_bloom_size", 2048), + bloom_hashes=data.get("tag_bloom_hashes", 3), + tag_count=len(peer_topics), + ) + + # Merge filter table gossip (transitive peer discovery) + if peer_filter_table and from_hash: + for ph, entry in peer_filter_table.items(): + if isinstance(entry, dict): + bb = entry.get("bloom_bytes") + if bb: + self.fdb.store_peer_filter( + peer_hash=ph, + bloom_bytes=bytes(bb) if isinstance(bb, list) else bb, + bloom_size=entry.get("bloom_size", 2048), + bloom_hashes=entry.get("bloom_hashes", 3), + tag_count=entry.get("tag_count", 0), + ) + + # Handle scoped query: find peers whose bloom might contain the queried tag + scoped_query_results = [] + if scoped_query_tag and from_hash: + scoped_query_results = self.fdb.get_filtered_peers_by_tag(scoped_query_tag) + + # Store peer's topics for future routing (backward compat) + if peer_topics and from_hash: + self.fdb.set_setting(f"peer_topics_{from_hash}", ",".join(peer_topics)) + if peer_tag_cloud and from_hash: + self.fdb.set_setting(f"peer_tag_cloud_{from_hash}", json.dumps(peer_tag_cloud)) + + # Build our tag bloom filter to send back + peer_tag_bs = data.get("tag_bloom_size", 2048) + peer_tag_bh = data.get("tag_bloom_hashes", 3) + my_tag_bloom = BloomFilter.from_tags(my_topics or [], peer_tag_bs, peer_tag_bh) # Build our content bloom filter for dedup our_existing = set() - peer_topics = data.get("my_topics", []) for t in self.fdb.get_threads_by_topics(peer_topics) if peer_topics else []: our_existing.add(t["id"]) our_bloom = BloomFilter.from_items(list(our_existing), data.get("bloom_size", 2048) if data else 2048, data.get("bloom_hashes", 3) if data else 3) + # Build filter table gossip from our stored filters + all_filters = self.fdb.get_all_filters() + filter_table_gossip = {} + for f in all_filters[:20]: + if f["peer_hash"] != from_hash: + filter_table_gossip[f["peer_hash"]] = { + "bloom_bytes": list(f["bloom_bytes"]), + "bloom_size": f["bloom_size"], + "bloom_hashes": f["bloom_hashes"], + "tag_count": f["tag_count"], + } + # Parse peer's content bloom for dedup peer_bloom = None if peer_bloom_data: @@ -869,10 +920,7 @@ class ForumHandlers: # Merge incoming content, filtered by bloom and topics if incoming_threads: for t in incoming_threads: - author = t.get("author_instance", "") - if author in blocked: - continue - if author and not self.fdb.is_trusted(author): + if t.get("author_instance", "") in blocked: continue if peer_bloom and peer_bloom.might_contain(t["id"]): continue @@ -884,10 +932,7 @@ class ForumHandlers: self.fdb.merge_thread(t) if incoming_posts: for p in incoming_posts: - author = p.get("author_instance", "") - if author in blocked: - continue - if author and not self.fdb.is_trusted(author): + if p.get("author_instance", "") in blocked: continue if peer_bloom and peer_bloom.might_contain(p["id"]): continue @@ -918,6 +963,8 @@ class ForumHandlers: my_blocks = list(blocked) my_peer_blocks = self.fdb.get_peer_block_list() + my_tag_cloud = self.fdb.get_tag_cloud(50) + my_tag_list = [t for t, _ in my_tag_cloud] threads, posts, upvote_threads = [], [], [] if since: @@ -934,14 +981,12 @@ class ForumHandlers: posts = [dict(r) for r in posts_list] upvote_threads = up_list - # Filter outgoing content by requester's trust list - requester_trust = set(data.get("trust_list", [])) - if requester_trust: - threads = [t for t in threads if t.get("author_instance", "") in requester_trust] - thread_ids = set(t["id"] for t in threads) - posts = [p for p in posts if p.get("thread_id") in thread_ids] - known_peers = [h for h in self.fdb.get_all_known_hashes() if h != from_hash] + peer_topics_map = {} + for ph in known_peers[:100]: + pt = self.fdb.get_setting(f"peer_topics_{ph}", "") + if pt: + peer_topics_map[ph] = [t.strip() for t in pt.split(",") if t.strip()] retracted = [{"id": cid, "type": ct, "author": ai, "at": ra} for cid, ct, ai, ra in self.fdb.get_raw_retractions()] @@ -956,10 +1001,18 @@ class ForumHandlers: "blocks": {"mine": my_blocks, "peers": my_peer_blocks}, "retractions": retracted, "known_peers": known_peers, + "peer_topics": my_tag_list, + "peer_tag_cloud": my_tag_cloud, "content_bloom": list(our_bloom.bytes), "bloom_size": data.get("bloom_size", 2048), "bloom_hashes": data.get("bloom_hashes", 3), - "trust_list": self.fdb.get_all_trusted_hashes(), + "peer_topics_map": peer_topics_map, + # Architecture B additions + "tag_bloom": list(my_tag_bloom.bytes), + "tag_bloom_size": peer_tag_bs, + "tag_bloom_hashes": peer_tag_bh, + "filter_table": filter_table_gossip, + "scoped_query_results": scoped_query_results, }), "headers": {}, } diff --git a/tinyweb_forum/sync.py b/tinyweb_forum/sync.py index e27b763..a9d544d 100644 --- a/tinyweb_forum/sync.py +++ b/tinyweb_forum/sync.py @@ -12,7 +12,11 @@ REQUEST_TIMEOUT = 60 GOSSIP_FANOUT = 20 BLOOM_SIZE = 2048 BLOOM_HASHES = 3 +TAG_BLOOM_SIZE = 2048 +TAG_BLOOM_HASHES = 3 +FILTER_TABLE_GOSSIP = 20 MAX_PEER_FAILURES = 5 +FILTER_TABLE_TTL_DAYS = 7 class _ForumAnnounceHandler: @@ -46,23 +50,21 @@ class ForumSync: raw = self.fdb.get_setting("topic_subscriptions", "") return [t.strip().lower() for t in raw.split(",") if t.strip()] - def _merge_trust_gossip(self, trust_list, source_hash): - source_hops = self.fdb.get_trust_hops(source_hash) - if source_hops is None: - return - my_hash = self.identity.hash.hex() if self.identity else "local" - blocked = set( - h.strip() - for h in self.fdb.get_setting("blocked_instances", "").split(",") - if h.strip() - ) - for peer_hash in trust_list: - if peer_hash in blocked or peer_hash == my_hash: - continue - new_hops = source_hops + 1 - existing = self.fdb.get_trust_hops(peer_hash) - if existing is None or new_hops < existing: - self.fdb.add_trust_source(peer_hash, source_hash, new_hops) + def _build_tag_bloom(self, topics=None): + if topics is None: + topics = self._get_subscribed_topics() + return BloomFilter.from_tags(topics or [], TAG_BLOOM_SIZE, TAG_BLOOM_HASHES) + + def _topics_overlap(self, my_topics, their_topics): + if not my_topics or not their_topics: + return True + return bool(set(my_topics) & set(their_topics)) + + def _topics_overlap_bloom(self, my_topics, peer_bloom): + """Check overlap using bloom filter instead of topic list.""" + if not my_topics or peer_bloom is None: + return True + return any(peer_bloom.might_contain(t) for t in my_topics) def start(self): self.destination = RNS.Destination( @@ -97,17 +99,43 @@ class ForumSync: random.shuffle(instances) count = 0 my_topics = self._get_subscribed_topics() + my_tag_bloom = self._build_tag_bloom(my_topics) for inst in instances[:GOSSIP_FANOUT]: if not self._running: break + pfilter = self.fdb.get_peer_filter(inst["instance_hash"]) + if pfilter: + pbloom = BloomFilter.from_bytes(pfilter["bloom_bytes"], pfilter["bloom_size"], pfilter["bloom_hashes"]) + if not self._topics_overlap_bloom(my_topics, pbloom): + continue + else: + peer_topics = self._peer_tag_topics(inst["instance_hash"]) + if not self._topics_overlap(my_topics, peer_topics): + continue try: - self._sync_with(inst["instance_hash"], my_topics) + self._sync_with(inst["instance_hash"], my_topics, my_tag_bloom) count += 1 except Exception as e: self.fdb.record_sync_result(inst["instance_hash"], False) print(f"[forum] sync error with {inst['instance_hash'][:16]}: {e}") return count + def scoped_query(self, tag): + """Ask known peers if they know anyone with the given tag.""" + tag = tag.strip().lower() + instances = self.fdb.get_synced_instances() + random.shuffle(instances) + results = set() + for inst in instances[:GOSSIP_FANOUT]: + if not self._running: + break + try: + matches = self._scoped_query_peer(inst["instance_hash"], tag) + results.update(matches) + except Exception: + continue + return list(results) + def _start_sync_loop(self): if self._thread and self._thread.is_alive(): return @@ -145,12 +173,22 @@ class ForumSync: try: instances = self.fdb.get_synced_instances() my_topics = self._get_subscribed_topics() + my_tag_bloom = self._build_tag_bloom(my_topics) random.shuffle(instances) for inst in instances[:GOSSIP_FANOUT]: if not self._running: break + pfilter = self.fdb.get_peer_filter(inst["instance_hash"]) + if pfilter: + pbloom = BloomFilter.from_bytes(pfilter["bloom_bytes"], pfilter["bloom_size"], pfilter["bloom_hashes"]) + if not self._topics_overlap_bloom(my_topics, pbloom): + continue + else: + peer_topics = self._peer_tag_topics(inst["instance_hash"]) + if not self._topics_overlap(my_topics, peer_topics): + continue try: - self._sync_with(inst["instance_hash"], my_topics) + self._sync_with(inst["instance_hash"], my_topics, my_tag_bloom) except Exception as e: self.fdb.record_sync_result(inst["instance_hash"], False) print(f"[forum] sync error with {inst['instance_hash'][:16]}: {e}") @@ -164,6 +202,10 @@ class ForumSync: self.fdb.prune_old_content(days) except Exception: pass + try: + self.fdb.prune_peer_filters(FILTER_TABLE_TTL_DAYS) + except Exception: + pass try: for dead in self.fdb.get_dead_peers(MAX_PEER_FAILURES): self.fdb.remove_synced_instance(dead) @@ -174,7 +216,84 @@ class ForumSync: return time.sleep(1) - def _sync_with(self, instance_hash, my_topics=None): + def _peer_tag_topics(self, instance_hash): + raw = self.fdb.get_setting(f"peer_topics_{instance_hash}", "") + return [t.strip().lower() for t in raw.split(",") if t.strip()] if raw else [] + + def _store_peer_filter(self, peer_hash, data): + tag_bloom_data = data.get("tag_bloom") + if tag_bloom_data: + self.fdb.store_peer_filter( + peer_hash=peer_hash, + bloom_bytes=bytes(tag_bloom_data), + bloom_size=data.get("tag_bloom_size", TAG_BLOOM_SIZE), + bloom_hashes=data.get("tag_bloom_hashes", TAG_BLOOM_HASHES), + tag_count=len(data.get("peer_topics", [])), + ) + + def _merge_filter_table_gossip(self, filter_table): + if not filter_table: + return + for ph, entry in filter_table.items(): + if isinstance(entry, dict): + bb = entry.get("bloom_bytes") + if bb: + self.fdb.store_peer_filter( + peer_hash=ph, + bloom_bytes=bytes(bb) if isinstance(bb, list) else bb, + bloom_size=entry.get("bloom_size", TAG_BLOOM_SIZE), + bloom_hashes=entry.get("bloom_hashes", TAG_BLOOM_HASHES), + tag_count=entry.get("tag_count", 0), + ) + + def _scoped_query_peer(self, peer_hash, tag): + dest_hash = bytes.fromhex(peer_hash) + if not RNS.Transport.has_path(dest_hash): + RNS.Transport.request_path(dest_hash) + elapsed = 0 + while not RNS.Transport.has_path(dest_hash) and elapsed < 15: + time.sleep(0.5) + elapsed += 0.5 + if not RNS.Transport.has_path(dest_hash): + return [] + server_identity = RNS.Identity.recall(dest_hash) + if server_identity is None: + return [] + destination = RNS.Destination( + server_identity, + RNS.Destination.OUT, + RNS.Destination.SINGLE, + FORUM_APP, + ) + link = RNS.Link(destination) + elapsed = 0 + while link.status == RNS.Link.PENDING and elapsed < 15: + time.sleep(0.25) + elapsed += 0.25 + if link.status != RNS.Link.ACTIVE: + return [] + try: + my_hash = self.identity.hash.hex() if self.identity else "local" + request_data = { + "scoped_query": tag, + "from_hash": my_hash, + } + receipt = link.request("/forum", data=request_data, timeout=REQUEST_TIMEOUT) + elapsed = 0 + done = (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED, RNS.RequestReceipt.FAILED) + while receipt.get_status() not in done and elapsed < REQUEST_TIMEOUT: + time.sleep(0.1) + elapsed += 0.1 + if receipt.get_status() in (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED): + resp = receipt.get_response() + if isinstance(resp, dict) and resp.get("status") == 200: + data = json.loads(resp["body"]) + return data.get("scoped_query_results", []) + return [] + finally: + link.teardown() + + def _sync_with(self, instance_hash, my_topics=None, my_tag_bloom=None): success = False dest_hash = bytes.fromhex(instance_hash) if not RNS.Transport.has_path(dest_hash): @@ -222,8 +341,25 @@ class ForumSync: if my_topics is None: my_topics = self._get_subscribed_topics() + if my_tag_bloom is None: + my_tag_bloom = self._build_tag_bloom(my_topics) - threads, posts, upvotes = [], [], [] + # Build filter table gossip: share a subset of our known peer filters + all_filters = self.fdb.get_all_filters() + filter_table_gossip = {} + for f in all_filters[:FILTER_TABLE_GOSSIP]: + if f["peer_hash"] != instance_hash and f["peer_hash"] != my_hash: + filter_table_gossip[f["peer_hash"]] = { + "bloom_bytes": list(f["bloom_bytes"]), + "bloom_size": f["bloom_size"], + "bloom_hashes": f["bloom_hashes"], + "tag_count": f["tag_count"], + } + + threads, posts = [], [] + upvotes = [] + my_tag_cloud = self.fdb.get_tag_cloud(50) + my_tag_list = [t for t, _ in my_tag_cloud] # Only build content if we have a since timestamp (incremental sync) if since: @@ -253,6 +389,11 @@ class ForumSync: for cid, ct, ai, ra in self.fdb.get_raw_retractions()] known_peers = [h for h in self.fdb.get_all_known_hashes() if h != instance_hash and h != my_hash] + peer_topics_map = {} + for ph in known_peers[:100]: + pt = self.fdb.get_setting(f"peer_topics_{ph}", "") + if pt: + peer_topics_map[ph] = [t.strip() for t in pt.split(",") if t.strip()] request_data = { "query": {"since": [since]} if since else {}, @@ -264,10 +405,16 @@ class ForumSync: "retractions": retracted, "known_peers": known_peers, "my_topics": my_topics, + "my_tag_cloud": my_tag_cloud, "content_bloom": list(content_bloom.bytes), "bloom_size": BLOOM_SIZE, "bloom_hashes": BLOOM_HASHES, - "trust_list": self.fdb.get_all_trusted_hashes(), + "peer_topics_map": peer_topics_map, + # Architecture B additions + "tag_bloom": list(my_tag_bloom.bytes), + "tag_bloom_size": TAG_BLOOM_SIZE, + "tag_bloom_hashes": TAG_BLOOM_HASHES, + "filter_table": filter_table_gossip, } receipt = link.request("/forum", data=request_data, timeout=REQUEST_TIMEOUT) @@ -285,10 +432,52 @@ class ForumSync: except (json.JSONDecodeError, KeyError): data = {} success = True + + # Store peer's tag bloom filter + self._store_peer_filter(instance_hash, data) + + # Merge filter table gossip from peer + self._merge_filter_table_gossip(data.get("filter_table", {})) + + # Store peer topics (backward compat) + peer_topics = data.get("peer_topics", []) + if peer_topics: + self.fdb.set_setting(f"peer_topics_{instance_hash}", ",".join(peer_topics)) + peer_tag_cloud = data.get("peer_tag_cloud", []) + if peer_tag_cloud: + self.fdb.set_setting(f"peer_tag_cloud_{instance_hash}", json.dumps(peer_tag_cloud)) + + # Check tag overlap using bloom filter my_blocks = set(h.strip() for h in self.fdb.get_setting("blocked_instances", "").split(",") if h.strip()) - # Merge trust gossip from peer - self._merge_trust_gossip(data.get("trust_list", []), instance_hash) + peer_tag_bloom_data = data.get("tag_bloom") + peer_tag_bloom = None + if peer_tag_bloom_data: + peer_tag_bloom = BloomFilter.from_bytes( + bytes(peer_tag_bloom_data), + data.get("tag_bloom_size", TAG_BLOOM_SIZE), + data.get("tag_bloom_hashes", TAG_BLOOM_HASHES), + ) + + # If no tag overlap, skip content sync but still store the filter + if my_topics and peer_tag_bloom is not None: + if not self._topics_overlap_bloom(my_topics, peer_tag_bloom): + success = True + now = time.strftime("%Y-%m-%dT%H:%M:%S") + self.fdb.set_last_sync(instance_hash, now) + self.fdb.record_sync_result(instance_hash, True) + return + + # Fall back to topic list overlap check if no bloom + if peer_tag_bloom is None: + incoming_topics = data.get("peer_topics", []) + if my_topics and incoming_topics: + if not self._topics_overlap(my_topics, incoming_topics): + success = True + now = time.strftime("%Y-%m-%dT%H:%M:%S") + self.fdb.set_last_sync(instance_hash, now) + self.fdb.record_sync_result(instance_hash, True) + return # Check content bloom for dedup peer_bloom_data = data.get("content_bloom") @@ -301,28 +490,20 @@ class ForumSync: ) for t in data.get("threads", []): - author = t.get("author_instance", "") - if author in my_blocks: - continue - if author and not self.fdb.is_trusted(author): - continue - if peer_bloom and peer_bloom.might_contain(t["id"]): - continue - if not my_topics: - self.fdb.merge_thread(t) - else: - t_tags = [tag.strip().lower() for tag in t.get("tags", "").split(",") if tag.strip()] - if set(my_topics) & set(t_tags): + if t.get("author_instance", "") not in my_blocks: + if peer_bloom and peer_bloom.might_contain(t["id"]): + continue + if not my_topics: self.fdb.merge_thread(t) + else: + t_tags = [tag.strip().lower() for tag in t.get("tags", "").split(",") if tag.strip()] + if set(my_topics) & set(t_tags): + self.fdb.merge_thread(t) for p in data.get("posts", []): - author = p.get("author_instance", "") - if author in my_blocks: - continue - if author and not self.fdb.is_trusted(author): - continue - if peer_bloom and peer_bloom.might_contain(p["id"]): - continue - self.fdb.merge_post(p) + if p.get("author_instance", "") not in my_blocks: + if peer_bloom and peer_bloom.might_contain(p["id"]): + continue + self.fdb.merge_post(p) for tid in data.get("upvote_threads", []): self.fdb.merge_upvote(tid, instance_hash) @@ -343,6 +524,13 @@ class ForumSync: if peer_hash and peer_hash != my_hash and peer_hash != instance_hash: self.fdb.add_known_peer(peer_hash) + my_topics_set = set(my_topics) + for ph, pt in data.get("peer_topics_map", {}).items(): + if ph and ph != my_hash and ph != instance_hash: + pt_set = set(t.lower() for t in pt) + if not my_topics or my_topics_set & pt_set: + self.fdb.add_known_peer(ph) + now = time.strftime("%Y-%m-%dT%H:%M:%S") self.fdb.set_last_sync(instance_hash, now) self.fdb.record_sync_result(instance_hash, success) @@ -358,7 +546,6 @@ class ForumSync: if h not in blocked and h not in auto_blocked and count >= 3: blocked.add(h) auto_blocked.add(h) - self.fdb.remove_by_hash(h) changed = True if changed: self.fdb.set_setting("blocked_instances", ",".join(blocked))