diff --git a/tinyweb_forum/db.py b/tinyweb_forum/db.py index a342e7f..e2dc3a5 100644 --- a/tinyweb_forum/db.py +++ b/tinyweb_forum/db.py @@ -4,8 +4,6 @@ import threading import time from datetime import datetime, timedelta -from tinyweb_forum.bloom import BloomFilter - FORUM_DB = "forum.db" @@ -91,16 +89,6 @@ 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: @@ -625,70 +613,7 @@ class ForumDB: # --- Filter Table (Architecture B: Bloom Gossip) --- - def store_peer_filter(self, peer_hash, bloom_bytes, bloom_size, bloom_hashes, tag_count): - db = self.get_db() - try: - db.execute( - "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()), - ) - db.commit() - finally: - self.return_db(db) - - def get_peer_filter(self, peer_hash): - db = self.get_db() - try: - row = db.execute( - "SELECT * FROM peer_filters WHERE peer_hash = ?", (peer_hash,) - ).fetchone() - if row: - return dict(row) - return None - finally: - self.return_db(db) - - def get_all_filters(self): - """Return all stored peer filters (for filter table gossip).""" - db = self.get_db() - try: - 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_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: - 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 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) + # --- Forum Trust --- # --- Forum Trust --- diff --git a/tinyweb_forum/handlers.py b/tinyweb_forum/handlers.py index 3b391ba..cbe3097 100644 --- a/tinyweb_forum/handlers.py +++ b/tinyweb_forum/handlers.py @@ -190,12 +190,10 @@ 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'
' ) @@ -841,79 +839,24 @@ 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", "") - # 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) - # Merge trust gossip from requester if from_hash: self.sync._merge_trust_gossip(data.get("trust_list", []), from_hash) - # 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: @@ -975,8 +918,6 @@ 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: @@ -1001,11 +942,6 @@ class ForumHandlers: 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()] @@ -1020,19 +956,10 @@ 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), - "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, "trust_list": self.fdb.get_all_trusted_hashes(), - "scoped_query_results": scoped_query_results, }), "headers": {}, } diff --git a/tinyweb_forum/sync.py b/tinyweb_forum/sync.py index dbab918..e27b763 100644 --- a/tinyweb_forum/sync.py +++ b/tinyweb_forum/sync.py @@ -12,11 +12,7 @@ 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: @@ -50,22 +46,6 @@ class ForumSync: raw = self.fdb.get_setting("topic_subscriptions", "") return [t.strip().lower() for t in raw.split(",") if t.strip()] - 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 _merge_trust_gossip(self, trust_list, source_hash): source_hops = self.fdb.get_trust_hops(source_hash) if source_hops is None: @@ -117,43 +97,17 @@ 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, my_tag_bloom) + self._sync_with(inst["instance_hash"], my_topics) 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 @@ -191,22 +145,12 @@ 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, my_tag_bloom) + self._sync_with(inst["instance_hash"], my_topics) except Exception as e: self.fdb.record_sync_result(inst["instance_hash"], False) print(f"[forum] sync error with {inst['instance_hash'][:16]}: {e}") @@ -220,10 +164,6 @@ 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) @@ -234,84 +174,7 @@ class ForumSync: return time.sleep(1) - 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): + def _sync_with(self, instance_hash, my_topics=None): success = False dest_hash = bytes.fromhex(instance_hash) if not RNS.Transport.has_path(dest_hash): @@ -359,25 +222,8 @@ 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) - # 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] + threads, posts, upvotes = [], [], [] # Only build content if we have a since timestamp (incremental sync) if since: @@ -407,11 +253,6 @@ 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 {}, @@ -423,16 +264,9 @@ 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, - "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, "trust_list": self.fdb.get_all_trusted_hashes(), } @@ -451,53 +285,8 @@ 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()) - 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 - # Merge trust gossip from peer self._merge_trust_gossip(data.get("trust_list", []), instance_hash) @@ -554,13 +343,6 @@ 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)