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
ce6d31c357
commit
497079c7a7
4 changed files with 875 additions and 164 deletions
|
|
@ -1,8 +1,11 @@
|
|||
import sqlite3
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from tinyweb_forum.bloom import BloomFilter
|
||||
|
||||
FORUM_DB = "forum.db"
|
||||
|
||||
|
||||
|
|
@ -88,6 +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.commit()
|
||||
db.close()
|
||||
|
||||
|
|
@ -509,22 +526,190 @@ class ForumDB:
|
|||
db = self.get_db()
|
||||
try:
|
||||
cutoff = (datetime.utcnow() - timedelta(days=retention_days)).strftime("%Y-%m-%dT%H:%M:%S")
|
||||
# Delete posts in old threads
|
||||
db.execute(
|
||||
"DELETE FROM posts WHERE thread_id IN "
|
||||
"(SELECT id FROM threads WHERE updated_at < ?)",
|
||||
(cutoff,),
|
||||
)
|
||||
# Delete orphaned posts (thread already deleted)
|
||||
db.execute(
|
||||
"DELETE FROM posts WHERE thread_id NOT IN (SELECT id FROM threads)"
|
||||
)
|
||||
# Delete old threads
|
||||
db.execute("DELETE FROM threads WHERE updated_at < ?", (cutoff,))
|
||||
# Clean up orphaned upvotes
|
||||
db.execute(
|
||||
"DELETE FROM upvotes WHERE thread_id NOT IN (SELECT id FROM threads)"
|
||||
)
|
||||
db.commit()
|
||||
finally:
|
||||
self.return_db(db)
|
||||
|
||||
def get_tag_cloud(self, limit=50):
|
||||
db = self.get_db()
|
||||
try:
|
||||
rows = db.execute("SELECT tags FROM threads").fetchall()
|
||||
counts = {}
|
||||
for r in rows:
|
||||
if r["tags"]:
|
||||
for t in r["tags"].split(","):
|
||||
tag = t.strip().lower()
|
||||
if tag:
|
||||
counts[tag] = counts.get(tag, 0) + 1
|
||||
return sorted(counts.items(), key=lambda x: -x[1])[:limit]
|
||||
finally:
|
||||
self.return_db(db)
|
||||
|
||||
def get_threads_by_topics(self, topics, since="", limit=200):
|
||||
db = self.get_db()
|
||||
try:
|
||||
params = []
|
||||
where = []
|
||||
if topics:
|
||||
clauses = []
|
||||
for t in topics:
|
||||
clauses.append("t.tags LIKE ?")
|
||||
params.append(f"%{t}%")
|
||||
where.append("(" + " OR ".join(clauses) + ")")
|
||||
if since:
|
||||
where.append("t.updated_at > ?")
|
||||
params.append(since)
|
||||
where_clause = (" WHERE " + " AND ".join(where)) if where else ""
|
||||
return db.execute(
|
||||
"SELECT t.*, (SELECT count(*) FROM posts p WHERE p.thread_id = t.id) AS reply_count "
|
||||
f"FROM threads t{where_clause} ORDER BY t.updated_at DESC LIMIT ?",
|
||||
params + [limit],
|
||||
).fetchall()
|
||||
finally:
|
||||
self.return_db(db)
|
||||
|
||||
def get_posts_by_thread_ids(self, thread_ids):
|
||||
if not thread_ids:
|
||||
return []
|
||||
db = self.get_db()
|
||||
try:
|
||||
placeholders = ",".join("?" for _ in thread_ids)
|
||||
return db.execute(
|
||||
f"SELECT * FROM posts WHERE thread_id IN ({placeholders}) ORDER BY created_at ASC",
|
||||
thread_ids,
|
||||
).fetchall()
|
||||
finally:
|
||||
self.return_db(db)
|
||||
|
||||
def get_new_upvotes_since(self, since, thread_ids=None):
|
||||
db = self.get_db()
|
||||
try:
|
||||
query = (
|
||||
"SELECT thread_id FROM upvotes u "
|
||||
"WHERE NOT EXISTS (SELECT 1 FROM threads t WHERE t.id = u.thread_id AND t.updated_at > ?)"
|
||||
)
|
||||
params = [since]
|
||||
if thread_ids:
|
||||
placeholders = ",".join("?" for _ in thread_ids)
|
||||
query += f" AND u.thread_id IN ({placeholders})"
|
||||
params.extend(thread_ids)
|
||||
return [r["thread_id"] for r in db.execute(query, params).fetchall()]
|
||||
finally:
|
||||
self.return_db(db)
|
||||
|
||||
# --- 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)
|
||||
|
||||
# --- Peer Liveness ---
|
||||
|
||||
def record_sync_result(self, peer_hash, success):
|
||||
db = self.get_db()
|
||||
try:
|
||||
existing = db.execute(
|
||||
"SELECT consecutive_failures FROM synced_instances WHERE instance_hash = ?",
|
||||
(peer_hash,),
|
||||
).fetchone()
|
||||
if existing is not None:
|
||||
new_failures = 0 if success else (existing["consecutive_failures"] + 1)
|
||||
db.execute(
|
||||
"UPDATE synced_instances SET consecutive_failures = ? WHERE instance_hash = ?",
|
||||
(new_failures, peer_hash),
|
||||
)
|
||||
if success:
|
||||
db.execute(
|
||||
"UPDATE synced_instances SET status = 'active' WHERE instance_hash = ?",
|
||||
(peer_hash,),
|
||||
)
|
||||
db.commit()
|
||||
finally:
|
||||
self.return_db(db)
|
||||
|
||||
def get_dead_peers(self, max_failures=5):
|
||||
"""Return list of peer hashes with too many consecutive failures."""
|
||||
db = self.get_db()
|
||||
try:
|
||||
return [
|
||||
r["instance_hash"] for r in db.execute(
|
||||
"SELECT instance_hash FROM synced_instances "
|
||||
"WHERE consecutive_failures >= ?",
|
||||
(max_failures,),
|
||||
).fetchall()
|
||||
]
|
||||
finally:
|
||||
self.return_db(db)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue