"""STRUDEL engine — announce listener + adaptive crawler + indexer for the Reticulum
search engine. Single-threaded scheduler loop; RNS callbacks only push events
onto a thread-safe queue (they never touch the DB). SQLite (WAL) is the sole
shared state; this process is the single writer.
Stdlib + rns only. Policy and source are published on the STRUDEL node.
"""
import os
import re
import sys
import time
import math
import random
import queue
import sqlite3
import hashlib
import threading
from dataclasses import dataclass
import posixpath
import json
import traceback
import RNS
DB_DIR = "/data/db"
DB_PATH = os.path.join(DB_DIR, "roogle.db")
HEARTBEAT = os.path.join(DB_DIR, "heartbeat")
NODE_ASPECT = "nomadnetwork.node"
TICK = 5 # scheduler tick, seconds
STATS_INTERVAL = 900 # recompute uptime + counters
PRUNE_INTERVAL = 86400 # prune old probes daily
PROBE_RETENTION = 90 * 86400
DAY = 86400
POLICY_VERSION = "2026-09-08.1"
POLICY_PATH = "/page/robots.txt"
PATH_TIMEOUT = 20
LINK_TIMEOUT = 30
REQ_TIMEOUT = 60
CRAWL_MAX_DEPTH = 2
CRAWL_MAX_PAGES = 40
CATALOGUE_LIMIT = 200
MAX_RESPONSE = 64 * 1024
POLICY_BYTES = 4096
SLOW_SECONDS = 20
SLOW_RATE = 1024
DAILY_VISITS = 512
DAILY_REQUESTS = 4096
DAILY_BYTES = 16 * 1024 * 1024
CRAWLER_IDENTITY = None
def log(*a):
print(time.strftime("%H:%M:%S"), *a, flush=True)
def jitter(v, frac=0.10):
return v * (1.0 + random.uniform(-frac, frac))
def touch_heartbeat():
try:
with open(HEARTBEAT, "w") as f:
f.write(str(time.time()))
except Exception:
pass
_LINK_TARGET_RE = re.compile(r"[[^]]*(/[^]s]+)")
_NAME_CTRL_RE = re.compile(r"[x00-x1fx7f]")
def clean_name(raw):
"""Node name from announce app_data: strip backticks + control chars,
cap 64 chars."""
if raw is None:
return None
s = raw.replace(", "")
s = _NAME_CTRL_RE.sub("", s).strip()
return s[:64] if s else None
def strip_micron(text):
"""Best-effort reduction of micron markup to plain text."""
# Lines starting with # are micron comments (incl. #! directives) — the
# parser never renders them, so they must not enter the index.
text = re.sub(r"(?m)^#.*$", "", text)
# Escaped backtick -> placeholder so we don't treat it as a control char.
text = text.replace("\, "x00")
# Links: label -> label
text = re.sub(r"[([^]]*)^\]*]", r"1", text)
# Links with no explicit target: label
text = re.sub(r"[([^]]*)]", r"1", text)
# Input fields with a prefilled value: -> prefill
text = re.sub(r"]*|[^]*[^>]*)>", r"1", text)
# Bare input fields: or
text = re.sub(r"]*>", "", text)
# Colour codes: xxx / xxx (3 hex) and Trrggbb / Trrggbb (24-bit).
# Without the T branch the 24-bit form matches nothing here and survives
# to the lone-backtick sweep below, which strips the and leaves the
# payload fused to adjacent text (rueTb4b4b4-> rueFTb4b4b4.
text = re.sub(r"FB(?:T[0-9a-fA-F]{6}|[0-9a-fA-F]{3})", "", text)
# Single-char format toggles / resets: etc.
text = re.sub(r"fb_!*clraie=<>", "", text)
# Double-backtick full reset
text = text.replace("", "")
# Section headers / dividers at line start
text = re.sub(r"(?m)^s*>+s*", "", text)
text = re.sub(r"(?m)^s*-{2,}.*$", "", text)
# Any remaining lone backticks
text = text.replace(", "")
text = text.replace("x00", ")
return text
def first_title(stripped, path):
for line in stripped.splitlines():
s = line.strip()
if s:
return s[:200]
return path
def safe_path(path):
# Never submit variables/actions or crawl generated histories and files.
if not isinstance(path, str) or len(path) > 256:
return None
if any(c in path for c in "?%#|") or ".." in path.split("/"):
return None
path = posixpath.normpath(path)
if not path.startswith("/page/") or not path.endswith(".mu"):
return None
if re.search(r"(?:^|[/_.-])(commit|commits|tree|blob|raw|diff|history|archive|download|search|calendar|file|files|feed)(?:[/_.-]|$)", path, re.I):
return None
return path
def extract_links(raw):
return list(dict.fromkeys(p for m in _LINK_TARGET_RE.finditer(raw)
if (p := safe_path(m.group(1)))))[:200]
SCHEMA = """
CREATE TABLE IF NOT EXISTS nodes(
hash TEXT PRIMARY KEY,
name TEXT,
first_seen REAL, last_announce REAL,
last_probe REAL, last_probe_ok INTEGER,
last_rtt_ms REAL,
next_probe REAL,
last_crawl REAL, next_crawl REAL,
crawl_status TEXT,
page_count INTEGER DEFAULT 0,
up24h REAL, up7d REAL, up30d REAL,
announces_seen INTEGER DEFAULT 0,
crawl_interval REAL,
last_content_change REAL,
seeded INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS probes(node_hash TEXT, ts REAL, ok INTEGER, rtt_ms REAL);
CREATE INDEX IF NOT EXISTS probes_node_ts ON probes(node_hash, ts);
CREATE TABLE IF NOT EXISTS pages(
node_hash TEXT, path TEXT, title TEXT,
content TEXT,
content_hash TEXT, fetched_at REAL, size INTEGER,
last_changed REAL, change_count INTEGER DEFAULT 0,
PRIMARY KEY(node_hash, path)
);
CREATE VIRTUAL TABLE IF NOT EXISTS pages_fts USING fts5(
title, content, content='pages', content_rowid='rowid');
CREATE TRIGGER IF NOT EXISTS pages_ai AFTER INSERT ON pages BEGIN
INSERT INTO pages_fts(rowid, title, content) VALUES (new.rowid, new.title, new.content);
END;
CREATE TRIGGER IF NOT EXISTS pages_ad AFTER DELETE ON pages BEGIN
INSERT INTO pages_fts(pages_fts, rowid, title, content) VALUES('delete', old.rowid, old.title, old.content);
END;
CREATE TRIGGER IF NOT EXISTS pages_au AFTER UPDATE ON pages BEGIN
INSERT INTO pages_fts(pages_fts, rowid, title, content) VALUES('delete', old.rowid, old.title, old.content);
INSERT INTO pages_fts(rowid, title, content) VALUES (new.rowid, new.title, new.content);
END;
CREATE TABLE IF NOT EXISTS meta(k TEXT PRIMARY KEY, v TEXT);
"""
def db_open():
conn = sqlite3.connect(DB_PATH, timeout=30)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA busy_timeout=30000")
conn.executescript(SCHEMA)
# Migration: announces_seen was added after first deploy; CREATE IF NOT
# EXISTS does not extend existing tables.
cols = [r[1] for r in conn.execute("PRAGMA table_info(nodes)").fetchall()]
if "announces_seen" not in cols:
conn.execute("ALTER TABLE nodes ADD COLUMN announces_seen INTEGER DEFAULT 0")
if "crawl_interval" not in cols:
conn.execute("ALTER TABLE nodes ADD COLUMN crawl_interval REAL")
if "last_content_change" not in cols:
conn.execute("ALTER TABLE nodes ADD COLUMN last_content_change REAL")
if "noindex" not in cols:
# Directory/search-engine nodes: kept in the node directory (probed,
# front page crawled for a description) but their page content is
# excluded from search results — meta-search pollution.
conn.execute("ALTER TABLE nodes ADD COLUMN noindex INTEGER DEFAULT 0")
if "seeded" not in cols:
# Nodes bulk-imported from The Nomad Index at launch (2026-07-29).
# Their sites pre-date strudel, so an insert-time first_seen would be a
# lie; the migration that introduced this column (2026-07-31) also
# NULLed first_seen for that cohort. first_seen is only meaningful
# where seeded=0 — i.e. nodes genuinely discovered by the engine.
conn.execute("ALTER TABLE nodes ADD COLUMN seeded INTEGER DEFAULT 0")
pcols = [r[1] for r in conn.execute("PRAGMA table_info(pages)").fetchall()]
if "last_changed" not in pcols:
conn.execute("ALTER TABLE pages ADD COLUMN last_changed REAL")
if "change_count" not in pcols:
conn.execute("ALTER TABLE pages ADD COLUMN change_count INTEGER DEFAULT 0")
migrate(conn)
conn.commit()
return conn
def meta_set(conn, k, v):
conn.execute("INSERT INTO meta(k, v) VALUES(?, ?) ON CONFLICT(k) DO UPDATE SET v=excluded.v",
(k, str(v)))
def upsert_announce(conn, h, name):
now = time.time()
row = conn.execute("SELECT hash FROM nodes WHERE hash=?", (h,)).fetchone()
if row:
if name:
conn.execute(
"UPDATE nodes SET last_announce=?, name=?, "
"announces_seen=COALESCE(announces_seen,0)+1 WHERE hash=?",
(now, name, h))
else:
conn.execute(
"UPDATE nodes SET last_announce=?, "
"announces_seen=COALESCE(announces_seen,0)+1 WHERE hash=?",
(now, h))
else:
conn.execute(
"INSERT INTO nodes(hash, name, first_seen, last_announce, next_probe, "
"next_crawl, crawl_status, announces_seen) VALUES(?,?,?,?,?,?, 'never', 1)",
(h, name, now, now, None, now + DAY + int(h[:8], 16) % (2 * DAY)))
# Global "announces seen" counter (survives node deletions).
conn.execute(
"INSERT INTO meta(k, v) VALUES('announces_total', '1') "
"ON CONFLICT(k) DO UPDATE SET v=CAST(CAST(v AS INTEGER)+1 AS TEXT)")
conn.commit()
def migrate(conn):
additions = {
'nodes': {'operator_days': 'INTEGER DEFAULT 0', 'slow_gate': 'INTEGER DEFAULT 0',
'slow_visits': 'INTEGER DEFAULT 0', 'fast_visits': 'INTEGER DEFAULT 0',
'last_slow': 'REAL', 'batch_size': 'INTEGER DEFAULT 5',
'successful_visits': 'INTEGER DEFAULT 0', 'failure_count': 'INTEGER DEFAULT 0',
'policy_checked': 'REAL', 'policy_status': 'TEXT', 'last_load_seconds': 'REAL'},
'pages': {'crawl_depth': 'INTEGER DEFAULT 1', 'interval_days': 'INTEGER DEFAULT 3',
'next_fetch': 'REAL DEFAULT 0', 'load_seconds': 'REAL',
'slow_samples': 'INTEGER DEFAULT 0', 'fast_samples': 'INTEGER DEFAULT 0',
'last_slow': 'REAL'},
}
for table, cols in additions.items():
existing = {r[1] for r in conn.execute('PRAGMA table_info(%s)' % table)}
for col, decl in cols.items():
if col not in existing:
conn.execute('ALTER TABLE %s ADD COLUMN %s %s' % (table, col, decl))
conn.executescript('''
CREATE TABLE IF NOT EXISTS frontier(
node_hash TEXT, path TEXT, depth INTEGER, added REAL,
PRIMARY KEY(node_hash,path));
CREATE TABLE IF NOT EXISTS crawl_visits(
id INTEGER PRIMARY KEY, node_hash TEXT, ts REAL, finished REAL,
status TEXT DEFAULT 'interrupted', pages INTEGER DEFAULT 0);
CREATE INDEX IF NOT EXISTS visits_ts ON crawl_visits(ts);
CREATE TABLE IF NOT EXISTS crawl_requests(
id INTEGER PRIMARY KEY, visit_id INTEGER, node_hash TEXT, path TEXT,
ts REAL, charge INTEGER, received INTEGER DEFAULT 0,
transferred INTEGER DEFAULT 0, elapsed REAL, status TEXT DEFAULT 'reserved');
CREATE INDEX IF NOT EXISTS requests_ts ON crawl_requests(ts);
''')
if not conn.execute("SELECT 1 FROM meta WHERE k='adaptive_migration'").fetchone():
now = time.time()
for n in conn.execute('SELECT * FROM nodes').fetchall():
interval = min(7 * DAY, max(DAY, n['crawl_interval'] or 3 * DAY))
spread = int(n['hash'][:8], 16) % (DAY if n['last_crawl'] else 7 * DAY)
nxt = max(now + spread, (n['last_crawl'] or 0) + interval)
conn.execute('UPDATE nodes SET next_probe=NULL, next_crawl=?, crawl_interval=?, '
'up24h=NULL,up7d=NULL,up30d=NULL WHERE hash=?',
(nxt, interval, n['hash']))
conn.execute("UPDATE pages SET next_fetch=COALESCE(fetched_at,0)+3*86400")
meta_set(conn, 'adaptive_migration', now)
meta_set(conn, 'policy_version', POLICY_VERSION)
meta_set(conn, 'limits', json.dumps({'daily_visits': DAILY_VISITS, 'daily_requests': DAILY_REQUESTS,
'daily_response_bytes': DAILY_BYTES, 'max_pages': CRAWL_MAX_PAGES,
'standard_bytes': 256*1024, 'slow_bytes': 32*1024, 'grey_bytes': 96*1024,
'max_response': MAX_RESPONSE, 'policy_bytes': POLICY_BYTES,
'standard_seconds': 600, 'slow_seconds': 90, 'grey_seconds': 180}))
def parse_policy(raw):
"""None keeps the previous preference; 0 explicitly restores automatic mode."""
groups, agents, values, directives = [], [], [], False
for line in raw.splitlines():
line = line.split('#', 1)[0].strip()
if ':' not in line:
continue
key, value = (part.strip() for part in line.split(':', 1))
if key.lower() == 'user-agent':
if directives:
groups.append((agents, values))
agents, values, directives = [], [], False
agents.append(value.lower())
elif agents:
directives = True
if key.lower() == 'recrawl-interval':
values.append(value.lower())
groups.append((agents, values))
selected = [v for a, v in groups if 'strudel' in a]
if not selected:
selected = [v for a, v in groups if '*' in a]
numbers, auto = [], False
for values in selected:
for value in values:
if value == 'auto':
auto = True
elif re.fullmatch(r'(?:[1-9]|1[0-4])d', value):
numbers.append(int(value[:-1]))
return max(numbers) if numbers else (0 if auto else None)
def next_interval(days, changed):
ladder = (1, 2, 3, 4, 7)
days = min(7, max(1, int(days or 3)))
if changed:
return max((d for d in ladder if d < days), default=1)
return min((d for d in ladder if d > days), default=7)
def effective_days(node, content_days):
failure = min(7, 2 ** min(3, node['failure_count'])) if node['failure_count'] else 1
return max(min(7, max(1, content_days)), node['operator_days'],
7 if node['slow_gate'] else 1, failure)
def limits(node):
if not node['successful_visits']:
pages, size, seconds = 1, 32*1024, 90
else:
pages, size, seconds = max(5, min(40, node['batch_size'])), 256*1024, 600
delay = 10
if node['operator_days'] > 7:
pages, size, seconds = min(pages, 10), min(size, 96*1024), min(seconds, 180)
if node['slow_gate']:
pages, size, seconds, delay = min(pages, 3), min(size, 32*1024), min(seconds, 90), 30
if node['fast_visits'] == 0:
pages = 1
return pages, size, seconds, delay
def pause(seconds):
end = time.monotonic() + max(0, seconds)
while time.monotonic() < end:
touch_heartbeat()
time.sleep(min(1, max(0, end - time.monotonic())))
def open_link(h, deadline):
started = time.monotonic()
link = None
try:
dh = bytes.fromhex(h)
if not RNS.Transport.has_path(dh):
RNS.Transport.request_path(dh)
until = min(deadline, started + PATH_TIMEOUT)
while not RNS.Transport.has_path(dh):
if time.monotonic() >= until:
return None, None
pause(.2)
ident = RNS.Identity.recall(dh)
if ident is None or time.monotonic() >= deadline:
return None, None
dest = RNS.Destination(ident, RNS.Destination.OUT, RNS.Destination.SINGLE,
'nomadnetwork', 'node')
link = RNS.Link(dest)
until = min(deadline, time.monotonic() + LINK_TIMEOUT)
while link.status != RNS.Link.ACTIVE:
if link.status == RNS.Link.CLOSED or time.monotonic() >= until:
link.teardown()
return None, None
pause(.1)
link.identify(CRAWLER_IDENTITY)
# Lower the decompression ceiling before response resources transfer.
link._strudel_limit = POLICY_BYTES
def resource_started(resource):
resource.max_decompressed_size = link._strudel_limit
link.set_resource_started_callback(resource_started)
return link, (time.monotonic() - started) * 1000
except Exception:
if link is not None:
link.teardown()
raise
@dataclass
class Response:
text: str | None
elapsed: float
size: int = 0
transferred: int = 0
status: str = 'ok'
def request_page(link, path, cap, deadline):
started = time.monotonic()
remaining = min(REQ_TIMEOUT, deadline - started)
if path == POLICY_PATH:
# No-handler requests receive no response in NomadNet. Bound the
# preference lookup using the established link, then use cached rules.
remaining = min(remaining, max(4, min(12, (getattr(link, 'rtt', 1) or 1)*4+2)))
if remaining <= 0:
return Response(None, 0, status='deadline')
done, state = threading.Event(), {}
link._strudel_limit = cap
def response(rr):
state['receipt'] = rr
done.set()
def failed(rr):
state['receipt'] = rr
state['failed'] = True
done.set()
rr = link.request(path, data=None, response_callback=response, failed_callback=failed,
timeout=remaining, max_response_size=cap)
if not rr:
return Response(None, time.monotonic() - started, status='failed')
end = min(deadline, started + remaining)
while not done.wait(.2):
touch_heartbeat()
if time.monotonic() >= end or link.status == RNS.Link.CLOSED:
rr.status = RNS.RequestReceipt.FAILED
for resource in list(getattr(link, 'incoming_resources', [])):
if getattr(resource, 'request_id', None) == rr.request_id:
resource.cancel()
return Response(None, time.monotonic()-started, status='timeout')
elapsed = time.monotonic() - started
received = state.get('receipt', rr)
transfer = getattr(received, 'response_transfer_size', 0) or 0
if state.get('failed'):
return Response(None, elapsed, transferred=transfer, status='failed')
data = getattr(received, 'response', None)
if data is None:
return Response(None, elapsed, status='missing')
if not isinstance(data, (bytes, bytearray, str)):
return Response(None, elapsed, transferred=transfer, status='nontext')
raw = data.encode('utf-8') if isinstance(data, str) else bytes(data)
if len(raw) > cap:
return Response(None, elapsed, len(raw), transfer, 'oversized')
text = raw.decode('utf-8', 'replace')
if 'x00' in text or (text and text.count('ufffd') > len(text) / 100):
return Response(None, elapsed, len(raw), transfer, 'nontext')
if 'Request Not Allowed' in text or 'not authorised to carry out the request' in text:
return Response(None, elapsed, len(raw), transfer, 'denied')
return Response(text, elapsed, len(raw), transfer)
def costly(result):
# Rate includes response latency: this is cost as seen here, not radio bitrate.
return result.elapsed > SLOW_SECONDS or (
result.size >= 4096 and result.elapsed >= 4 and
(result.transferred or result.size) / result.elapsed < SLOW_RATE)
def inexpensive(result):
return result.text is not None and result.elapsed < 8 and (
result.size < 4096 or (result.transferred or result.size) / max(.001, result.elapsed) > 4096)
def start_visit(conn, node):
now = time.time()
# A simultaneous slowdown across independent nodes throttles the whole
# crawler rather than repeatedly testing the same congested route.
hold = conn.execute("SELECT v FROM meta WHERE k='network_backoff_until'").fetchone()
if hold and float(hold[0]) > now:
return None
# Reserve the visit and next eligibility before any packet can leave.
conn.execute('BEGIN IMMEDIATE')
if conn.execute('SELECT COUNT(*) FROM crawl_visits WHERE ts>?', (now-DAY,)).fetchone()[0] >= DAILY_VISITS:
conn.rollback()
return None
used = conn.execute('SELECT COUNT(*),COALESCE(SUM(charge),0) FROM crawl_requests WHERE ts>?', (now-DAY,)).fetchone()
if used[0] >= DAILY_REQUESTS or used[1] + POLICY_BYTES > DAILY_BYTES:
conn.rollback()
return None
cur = conn.execute('INSERT INTO crawl_visits(node_hash,ts) VALUES(?,?)', (node['hash'], now))
days = effective_days(node, (node['crawl_interval'] or 3*DAY)/DAY)
conn.execute('UPDATE nodes SET last_crawl=?,next_crawl=?,crawl_status=?,next_probe=NULL WHERE hash=?',
(now, now+days*DAY, 'interrupted', node['hash']))
conn.commit()
return cur.lastrowid
def fetch_budgeted(conn, visit_id, h, link, path, cap, deadline):
now = time.time()
conn.execute('BEGIN IMMEDIATE')
used = conn.execute('SELECT COUNT(*),COALESCE(SUM(charge),0) FROM crawl_requests WHERE ts>?', (now-DAY,)).fetchone()
if used[0] >= DAILY_REQUESTS or used[1]+cap > DAILY_BYTES:
conn.rollback()
return None
cur = conn.execute('INSERT INTO crawl_requests(visit_id,node_hash,path,ts,charge) VALUES(?,?,?,?,?)',
(visit_id, h, path, now, cap))
conn.commit()
# On crash, the full reservation survives; retries cannot reclaim it.
result = request_page(link, path, cap, deadline)
charge = cap if result.text is None else min(cap, max(result.size, result.transferred))
conn.execute('UPDATE crawl_requests SET charge=?,received=?,transferred=?,elapsed=?,status=? WHERE id=?',
(charge,result.size,result.transferred,result.elapsed,result.status,cur.lastrowid))
conn.commit()
return result, charge
def enqueue(conn, h, path, depth):
if safe_path(path) is None or depth > CRAWL_MAX_DEPTH:
return
count = conn.execute('SELECT COUNT(*) FROM frontier WHERE node_hash=?', (h,)).fetchone()[0]
if count < CATALOGUE_LIMIT:
conn.execute('INSERT OR IGNORE INTO frontier VALUES(?,?,?,?)', (h,path,depth,time.time()))
def page_candidates(conn, h, attempted):
# Due pages first; never-fetched candidates follow. Old checks cannot starve.
rows = conn.execute('''SELECT f.path,f.depth,p.content_hash,p.interval_days,p.fetched_at
FROM frontier f LEFT JOIN pages p ON p.node_hash=f.node_hash AND p.path=f.path
WHERE f.node_hash=? AND (p.next_fetch IS NULL OR p.next_fetch<=?)
ORDER BY CASE WHEN f.path='/page/index.mu' AND p.path IS NULL THEN 0 ELSE 1 END,
COALESCE(p.next_fetch,f.added),f.path LIMIT 200''', (h,time.time())).fetchall()
return [r for r in rows if r['path'] not in attempted]
def store_page(conn, h, path, depth, result):
now = time.time()
text = strip_micron(result.text)
# Whitespace-only formatting changes do not accelerate a site.
digest = hashlib.sha256(re.sub(r's+', ' ', text).strip().encode()).hexdigest()
old = conn.execute('SELECT * FROM pages WHERE node_hash=? AND path=?', (h,path)).fetchone()
changed = old is not None and re.sub(r's+', ' ', old['content']).strip() != re.sub(r's+', ' ', text).strip()
interval = next_interval(old['interval_days'], changed) if old else 3
slow = costly(result)
old_slow = old['slow_samples'] if old and (old['last_slow'] or 0) > now-60*DAY else 0
slow_count = old_slow + 1 if slow else old_slow
fast_count = (old['fast_samples'] if old else 0) + 1 if inexpensive(result) else 0
if fast_count >= 2:
slow_count = 0
last_slow = now if slow else (old['last_slow'] if old else None)
conn.execute('''INSERT INTO pages(node_hash,path,title,content,content_hash,fetched_at,size,
last_changed,change_count,crawl_depth,interval_days,next_fetch,load_seconds,slow_samples,fast_samples,last_slow)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(node_hash,path) DO UPDATE SET title=excluded.title,content=excluded.content,
content_hash=excluded.content_hash,fetched_at=excluded.fetched_at,size=excluded.size,
last_changed=excluded.last_changed,change_count=excluded.change_count,
crawl_depth=excluded.crawl_depth,interval_days=excluded.interval_days,next_fetch=excluded.next_fetch,
load_seconds=excluded.load_seconds,slow_samples=excluded.slow_samples,
fast_samples=excluded.fast_samples,last_slow=excluded.last_slow''',
(h,path,first_title(text,path),text,digest,now,result.size,
now if changed else (old['last_changed'] if old else None),
(old['change_count'] or 0)+int(changed) if old else 0,depth,interval,now+interval*DAY,
result.elapsed,slow_count,fast_count,last_slow))
return changed, old is None
def do_crawl(conn, original):
node = dict(original)
h = node['hash']
visit_id = start_visit(conn, node)
if visit_id is None:
return False
started = time.monotonic()
link = None
status, fetched, changed_any, useful = 'failed', 0, False, False
visit_slow, successful_slow, fast = False, False, True
content_attempts, charged = set(), 0
max_pages, byte_limit, seconds, delay = limits(node)
deadline = started + seconds
try:
link, setup = open_link(h, deadline)
conn.execute('UPDATE nodes SET last_probe=?,last_probe_ok=?,last_rtt_ms=? WHERE hash=?',
(time.time(), int(link is not None), setup, h))
conn.commit()
if link is None:
return True
policy = fetch_budgeted(conn,visit_id,h,link,POLICY_PATH,POLICY_BYTES,deadline)
if policy is None:
status = 'budget'
return True
result, cost = policy
charged += cost
conn.execute('UPDATE nodes SET policy_checked=?,policy_status=? WHERE hash=?',
(time.time(),result.status,h))
if result.text is not None:
preference = parse_policy(result.text)
if preference is not None:
node['operator_days'] = preference
conn.execute('UPDATE nodes SET operator_days=? WHERE hash=?', (preference,h))
max_pages, byte_limit, seconds, delay = limits(node)
deadline = started + seconds
if preference > original['operator_days'] and original['last_crawl'] and time.time()-original['last_crawl'] < preference*DAY:
status = 'preference updated'
conn.commit()
return True
conn.commit()
if result.status in ('denied','oversized','nontext') or costly(result):
visit_slow = costly(result)
status = 'slow' if visit_slow else result.status
return True
if result.text is None:
# Missing handlers are silent in NomadNet. An actual long timeout
# is costly and stops above; a quick empty/failure gets one page.
if link.status != RNS.Link.ACTIVE:
return True
# A silent missing handler does not reduce an otherwise proven
# node to one page forever. First contact is already limited to one.
enqueue(conn,h,'/page/index.mu',0)
for old in conn.execute('SELECT path,crawl_depth FROM pages WHERE node_hash=? LIMIT 200',(h,)).fetchall():
enqueue(conn,h,old['path'],old['crawl_depth'])
conn.commit()
status = 'ok'
last_request_end = time.monotonic()
while len(content_attempts) < max_pages:
candidates = page_candidates(conn,h,content_attempts)
if not candidates:
break
candidate = candidates[0]
if node.get('noindex') and candidate['path'] != '/page/index.mu':
content_attempts.add(candidate['path'])
continue
cap = min(16*1024 if node['slow_gate'] or not node['successful_visits'] else MAX_RESPONSE, byte_limit-charged)
if cap < 512 or time.monotonic()+delay >= deadline:
break
pause(max(0, last_request_end+delay-time.monotonic()))
content_attempts.add(candidate['path'])
response = fetch_budgeted(conn,visit_id,h,link,candidate['path'],cap,deadline)
last_request_end = time.monotonic()
if response is None:
status = 'budget'
break
result, cost = response
charged += cost
if result.text is None:
status = result.status
visit_slow = costly(result) or result.status == 'timeout'
break
if not result.text.strip():
# Empty responses do not prove deletion.
continue
changed, new = store_page(conn,h,candidate['path'],candidate['depth'],result)
fetched += 1
changed_any |= changed
useful |= changed or new
fast &= inexpensive(result)
conn.execute('UPDATE nodes SET last_load_seconds=? WHERE hash=?',(result.elapsed,h))
if costly(result):
visit_slow = successful_slow = True
conn.commit()
status = 'slow'
break
if candidate['depth'] < CRAWL_MAX_DEPTH and not node.get('noindex'):
for path in extract_links(result.text):
# Do not expand exact duplicate content across endless paths.
duplicate = conn.execute('SELECT COUNT(*) FROM pages WHERE node_hash=? AND content_hash=(SELECT content_hash FROM pages WHERE node_hash=? AND path=?)',
(h,h,candidate['path'])).fetchone()[0]
if duplicate < 3:
enqueue(conn,h,path,candidate['depth']+1)
conn.commit()
except Exception:
status = 'error'
log('crawl error',h,traceback.format_exc())
finally:
if link is not None:
link.teardown()
now = time.time()
if visit_slow:
node['slow_gate'] = 1
node['fast_visits'] = 0
if successful_slow:
node['slow_visits'] += 1
node['last_slow'] = now
elif fetched and fast:
node['fast_visits'] += 1
if node['fast_visits'] >= 2:
node['slow_gate'] = 0
node['slow_visits'] = 0
else:
node['fast_visits'] = 0
failed = status not in ('ok','slow','budget','preference updated')
node['failure_count'] = min(3,node['failure_count']+1) if failed else 0
days = next_interval((node['crawl_interval'] or 3*DAY)/DAY,changed_any) if fetched else (node['crawl_interval'] or 3*DAY)/DAY
days = effective_days(node,days)
batch = node['batch_size']
if fetched and useful and fast and not visit_slow:
batch = min((b for b in (5,10,20,40) if b>batch),default=40)
elif visit_slow or failed or charged >= byte_limit:
batch = max(5,batch//2)
count = conn.execute('SELECT COUNT(*) FROM pages WHERE node_hash=?',(h,)).fetchone()[0]
conn.execute('''UPDATE nodes SET next_crawl=?,crawl_interval=?,crawl_status=?,page_count=?,
operator_days=?,slow_gate=?,slow_visits=?,fast_visits=?,last_slow=?,batch_size=?,
successful_visits=successful_visits+?,failure_count=?,
last_content_change=CASE WHEN ? THEN ? ELSE last_content_change END WHERE hash=?''',
(now+days*DAY,min(7,days)*DAY,status,count,node['operator_days'],node['slow_gate'],
node['slow_visits'],node['fast_visits'],node['last_slow'],batch,int(fetched>0),
node['failure_count'],int(changed_any),now,h))
conn.execute('UPDATE crawl_visits SET finished=?,status=?,pages=? WHERE id=?',(now,status,fetched,visit_id))
if visit_slow:
recent = conn.execute("SELECT COUNT(DISTINCT node_hash) FROM crawl_visits WHERE ts>? AND status='slow'", (now-1800,)).fetchone()[0]
if recent >= 3:
meta_set(conn,'network_backoff_until',now+3600)
conn.commit()
touch_heartbeat()
log('visit',h,status,'pages='+str(fetched),'charge='+str(charged),'next_days='+str(days))
return True
class NodeAnnounceHandler:
aspect_filter = NODE_ASPECT
def __init__(self, events):
self.events = events
def received_announce(self, destination_hash, announced_identity, app_data):
name = clean_name(app_data.decode('utf-8','replace')) if app_data else None
try:
self.events.put_nowait((destination_hash.hex(),name))
except queue.Full:
pass
def drain_events(conn, events):
for _ in range(500):
try:
h,name = events.get_nowait()
except queue.Empty:
break
upsert_announce(conn,h,name)
def main():
global CRAWLER_IDENTITY
os.makedirs(DB_DIR,exist_ok=True)
# Process-wide fallback also bounds decompression before a resource callback.
RNS.Resource.AUTO_COMPRESS_MAX_SIZE = MAX_RESPONSE
reticulum = RNS.Reticulum()
conn = db_open()
identity_path = os.path.join(DB_DIR,'crawler.identity')
if os.path.exists(identity_path):
CRAWLER_IDENTITY = RNS.Identity.from_file(identity_path)
if CRAWLER_IDENTITY is None:
raise RuntimeError('Cannot read persistent crawler identity; refusing to rotate it')
else:
CRAWLER_IDENTITY = RNS.Identity()
CRAWLER_IDENTITY.to_file(identity_path)
os.chmod(identity_path,0o600)
meta_set(conn,'crawler_identity',CRAWLER_IDENTITY.hash.hex())
meta_set(conn,'engine_started',time.time())
conn.commit()
events = queue.Queue(maxsize=10000)
handler = NodeAnnounceHandler(events)
RNS.Transport.register_announce_handler(handler)
log('STRUDEL adaptive crawler',POLICY_VERSION,'identity',CRAWLER_IDENTITY.hash.hex())
last_stats = 0
while True:
touch_heartbeat()
drain_events(conn,events)
node = conn.execute('SELECT * FROM nodes WHERE next_crawl<=? ORDER BY next_crawl LIMIT 1', (time.time(),)).fetchone()
if node is not None:
do_crawl(conn,node)
if time.time()-last_stats > STATS_INTERVAL:
now=time.time()
meta_set(conn,'stats_updated',now)
meta_set(conn,'pages_indexed',conn.execute('SELECT COUNT(*) FROM pages').fetchone()[0])
conn.execute('DELETE FROM crawl_requests WHERE ts<?',(now-90*DAY,))
conn.execute('DELETE FROM crawl_visits WHERE ts<?',(now-90*DAY,))
conn.commit()
last_stats=now
pause(TICK)
if __name__ == '__main__':
main()