lichenblankie/tinyweb-forum
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

remove dead Architecture B code (topic bloom, filter_table, peer_filters)

Browse source
This commit is contained in:
blankie 2026-06-17 20:11:19 +00:00
parent f680931c8c
commit d63e0c19be
3 changed files with 6 additions and 372 deletions
Download patch file Download diff file Expand all files Collapse all files

77
tinyweb_forum/db.py
View file

@ -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 ---

75
tinyweb_forum/handlers.py
View file

@ -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'<div style="font-size:0.85rem;color:#606060">'
f'subscribed: {esc(topics_str)}'
f' · {peer_count} peers'
f' · {filter_count} filters'
f'</div>'
)
@ -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": {},
}

226
tinyweb_forum/sync.py
View file

@ -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)