Architecture B: Bloom Gossip + Implicit Replication
- Tag bloom filter as primary peer discovery (2048 bits x 3 hashes) - Filter table gossip for transitive peer discovery - Scoped queries for on-demand tag lookup - Liveness tracking + peer eviction (5 failures, 7-day TTL) - Topic subscriptions filter content sync at both ends - Two-theme CSS system (default minimal + kodama2) - Status bar on all pages (topics, peers, filters) - Bracketless tags, grouped moderation page, cleaner forms - Match main site heading level (h1 -> h2)
This commit is contained in:
blankie
parent
96d26f3970
commit
987317b4ac
4 changed files with 875 additions and 164 deletions
|
|
@ -4,15 +4,22 @@ import threading
|
|||
import time
|
||||
import RNS
|
||||
|
||||
from tinyweb_forum.bloom import BloomFilter
|
||||
|
||||
FORUM_APP = "tinyweb-forum"
|
||||
SYNC_INTERVAL = 300 # 5 minutes
|
||||
SYNC_INTERVAL = 300
|
||||
REQUEST_TIMEOUT = 60
|
||||
GOSSIP_FANOUT = 20 # random peers to sync per cycle
|
||||
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:
|
||||
"""Receives announces from other forum instances and auto-discovers them."""
|
||||
|
||||
aspect_filter = FORUM_APP
|
||||
receive_path_responses = False
|
||||
|
||||
|
|
@ -39,6 +46,26 @@ class ForumSync:
|
|||
self._running = False
|
||||
self._thread = None
|
||||
|
||||
def _get_subscribed_topics(self):
|
||||
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 start(self):
|
||||
self.destination = RNS.Destination(
|
||||
self.identity,
|
||||
|
|
@ -66,23 +93,48 @@ class ForumSync:
|
|||
self.fdb.set_setting("forum_auto_sync", "1" if enabled else "0")
|
||||
if enabled:
|
||||
self._start_sync_loop()
|
||||
else:
|
||||
pass # current cycle finishes, no new one starts
|
||||
|
||||
def sync_now(self):
|
||||
"""Run one sync cycle immediately. Returns count of peers synced."""
|
||||
instances = self.fdb.get_synced_instances()
|
||||
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)
|
||||
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:
|
||||
self._sync_with(inst["instance_hash"])
|
||||
count += 1
|
||||
except Exception as e:
|
||||
print(f"[forum] sync error with {inst['instance_hash'][:16]}: {e}")
|
||||
return count
|
||||
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():
|
||||
|
|
@ -120,16 +172,25 @@ class ForumSync:
|
|||
while self._running:
|
||||
try:
|
||||
instances = self.fdb.get_synced_instances()
|
||||
# Gossip: sync with random subset for scaling
|
||||
# If <= GOSSIP_FANOUT peers, sync with all (current behavior)
|
||||
# If more, sync with random FANOUT per cycle — content spreads epidemically
|
||||
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"])
|
||||
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}")
|
||||
except Exception:
|
||||
pass
|
||||
|
|
@ -141,12 +202,99 @@ 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)
|
||||
except Exception:
|
||||
pass
|
||||
for _ in range(SYNC_INTERVAL):
|
||||
if not self._running:
|
||||
return
|
||||
time.sleep(1)
|
||||
|
||||
def _sync_with(self, instance_hash):
|
||||
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):
|
||||
RNS.Transport.request_path(dest_hash)
|
||||
|
|
@ -155,10 +303,12 @@ class ForumSync:
|
|||
time.sleep(0.5)
|
||||
elapsed += 0.5
|
||||
if not RNS.Transport.has_path(dest_hash):
|
||||
self.fdb.record_sync_result(instance_hash, False)
|
||||
return
|
||||
|
||||
server_identity = RNS.Identity.recall(dest_hash)
|
||||
if server_identity is None:
|
||||
self.fdb.record_sync_result(instance_hash, False)
|
||||
return
|
||||
|
||||
destination = RNS.Destination(
|
||||
|
|
@ -175,6 +325,7 @@ class ForumSync:
|
|||
elapsed += 0.25
|
||||
|
||||
if link.status != RNS.Link.ACTIVE:
|
||||
self.fdb.record_sync_result(instance_hash, False)
|
||||
return
|
||||
|
||||
try:
|
||||
|
|
@ -186,14 +337,50 @@ class ForumSync:
|
|||
break
|
||||
|
||||
since = last_sync.replace(" ", "T") if last_sync else ""
|
||||
my_hash = self.identity.hash.hex() if self.identity else "local"
|
||||
|
||||
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]
|
||||
|
||||
# Only build content if we have a since timestamp (incremental sync)
|
||||
if since:
|
||||
ts, ps, uv = self.fdb.get_new_content(since)
|
||||
threads = [dict(r) for r in ts]
|
||||
posts = [dict(r) for r in ps]
|
||||
upvotes = [{"thread_id": tid, "instance_hash": instance_hash} for tid in uv]
|
||||
if my_topics:
|
||||
rows = self.fdb.get_threads_by_topics(my_topics, since=since)
|
||||
threads = [dict(r) for r in rows]
|
||||
tids = [r["id"] for r in rows]
|
||||
posts = [dict(p) for p in self.fdb.get_posts_by_thread_ids(tids)]
|
||||
uv_rows = self.fdb.get_new_upvotes_since(since, tids)
|
||||
upvotes = [{"thread_id": tid, "instance_hash": instance_hash} for tid in uv_rows]
|
||||
else:
|
||||
ts, ps, uv = self.fdb.get_new_content(since)
|
||||
threads = [dict(r) for r in ts]
|
||||
posts = [dict(r) for r in ps]
|
||||
upvotes = [{"thread_id": tid, "instance_hash": instance_hash} for tid in uv]
|
||||
|
||||
# Build content bloom for dedup
|
||||
existing_ids = set(t["id"] for t in threads)
|
||||
for t in self.fdb.get_threads_by_topics(my_topics) if my_topics else []:
|
||||
existing_ids.add(t["id"])
|
||||
content_bloom = BloomFilter.from_items(list(existing_ids), BLOOM_SIZE, BLOOM_HASHES)
|
||||
|
||||
my_blocks = [h.strip() for h in self.fdb.get_setting("blocked_instances", "").split(",") if h.strip()]
|
||||
my_peer_blocks = self.fdb.get_peer_block_list()
|
||||
|
|
@ -201,8 +388,12 @@ class ForumSync:
|
|||
retracted = [{"id": cid, "type": ct, "author": ai, "at": ra}
|
||||
for cid, ct, ai, ra in self.fdb.get_raw_retractions()]
|
||||
|
||||
my_hash = self.identity.hash.hex() if self.identity else "local"
|
||||
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 {},
|
||||
|
|
@ -213,6 +404,17 @@ class ForumSync:
|
|||
"blocks": {"mine": my_blocks, "peers": my_peer_blocks},
|
||||
"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,
|
||||
}
|
||||
|
||||
receipt = link.request("/forum", data=request_data, timeout=REQUEST_TIMEOUT)
|
||||
|
|
@ -229,16 +431,82 @@ class ForumSync:
|
|||
data = json.loads(resp["body"])
|
||||
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
|
||||
|
||||
# Check content bloom for dedup
|
||||
peer_bloom_data = data.get("content_bloom")
|
||||
peer_bloom = None
|
||||
if peer_bloom_data:
|
||||
peer_bloom = BloomFilter.from_bytes(
|
||||
bytes(peer_bloom_data),
|
||||
data.get("bloom_size", BLOOM_SIZE),
|
||||
data.get("bloom_hashes", BLOOM_HASHES),
|
||||
)
|
||||
|
||||
for t in data.get("threads", []):
|
||||
if t.get("author_instance", "") not in my_blocks:
|
||||
self.fdb.merge_thread(t)
|
||||
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", []):
|
||||
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)
|
||||
# Gossip blocks from peer
|
||||
|
||||
peer_blocks = data.get("blocks", {})
|
||||
for h in peer_blocks.get("mine", []):
|
||||
if h and h not in my_blocks and instance_hash:
|
||||
|
|
@ -247,18 +515,25 @@ class ForumSync:
|
|||
if h and h not in my_blocks and instance_hash:
|
||||
self.fdb.record_peer_block(instance_hash, h)
|
||||
self._apply_peer_blocks()
|
||||
# Merge incoming retractions
|
||||
|
||||
for r in data.get("retractions", []):
|
||||
if r.get("id") and r.get("type") and r.get("author") and r.get("at"):
|
||||
self.fdb.merge_retraction(r["id"], r["type"], r["author"], r["at"])
|
||||
# Discover new peers from gossip
|
||||
|
||||
for peer_hash in data.get("known_peers", []):
|
||||
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)
|
||||
else:
|
||||
pass
|
||||
self.fdb.record_sync_result(instance_hash, success)
|
||||
finally:
|
||||
link.teardown()
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue