#!/usr/bin/env python3 """ JonasSearch Crawler Gamma 0.1 PostgreSQL-based parallel crawler. Features: - PostgreSQL queue with row-level locking (FOR UPDATE SKIP LOCKED) - Multiple crawler workers can safely share one queue - Stores title, headings, cleaned text and links - Search vector is maintained by the PostgreSQL trigger - URL deduplication via UNIQUE(url) - Deadlock-safe parallel queue inserts (deterministic ordering, bounded retries with rollback + backoff) - Smart priority queue: prefers never-crawled URLs, new domains and new hosts, with a spread bonus for little-explored territory - Per-host crawl delay - robots.txt support with in-memory cache - Graceful shutdown - Basic crawler statistics - Generation-based rolling updates: the crawler remembers the deployment generation it started with, checks it before every relevant commit (safe point) and exits with exit code 42 when a new generation requires an update - the open transaction is rolled back, never half-committed - Launcher support: --version, --self-update N (verify a staged release) and --update-deployment N (bump generation + required_version) Expected PostgreSQL schema: pages crawl_queue crawler_stats Connection defaults can be overridden with environment variables: JONASSEARCH_DB_HOST JONASSEARCH_DB_PORT JONASSEARCH_DB_NAME JONASSEARCH_DB_USER JONASSEARCH_DB_PASSWORD JONASSEARCH_UPDATE_CHECK_SECONDS JONASSEARCH_RELEASES_DIR Example: python3 crawler.py """ import os import re import sys import math import signal import time import logging import argparse import subprocess import threading import multiprocessing as mp from collections import defaultdict from html import unescape from pathlib import Path from urllib.parse import parse_qsl, urljoin, urlparse, urldefrag from urllib.robotparser import RobotFileParser from dotenv import load_dotenv import psycopg2 from psycopg2.extras import execute_values from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError try: from bs4 import BeautifulSoup except ImportError: BeautifulSoup = None def _load_env_files(): """ .env zuerst aus dem Arbeitsverzeichnis laden (Standard), zusätzlich aus dem Projektordner – so findet der Prozess die Konfiguration auch dann, wenn er von woanders gestartet wird (z. B. via systemd). Eine nicht lesbare .env darf den Crawler nicht abstürzen lassen (z. B. falscher Benutzer unter systemd): dann laufen die bereits gesetzten Umgebungsvariablen unverändert weiter. """ for dotenv_path in (None, Path(__file__).resolve().parent / ".env"): try: load_dotenv(dotenv_path) except OSError as exc: logging.warning( "[CONFIG] %s not readable (%s); continuing with the " "current environment.", dotenv_path or ".env (working directory)", exc, ) _load_env_files() # ============================================================ # CONFIG # ============================================================ DB_HOST = os.getenv("JONASSEARCH_DB_HOST", "127.0.0.1") DB_PORT = int(os.getenv("JONASSEARCH_DB_PORT", "25432")) DB_NAME = os.getenv("JONASSEARCH_DB_NAME", "jonassearch") DB_USER = os.getenv("JONASSEARCH_DB_USER", "jonassearch") DB_PASSWORD = os.getenv("JONASSEARCH_DB_PASSWORD", "") START_URL = os.getenv("JONASSEARCH_START_URL", "https://github.com/") WORKERS = int(os.getenv("JONASSEARCH_WORKERS", "3")) # Minimum delay between requests to the same host, per worker. # robots.txt can specify a larger crawl delay. DEFAULT_DELAY = float(os.getenv("JONASSEARCH_DELAY", "3.0")) ROBOTS_CACHE_SECONDS = int( os.getenv("JONASSEARCH_ROBOTS_CACHE_SECONDS", str(24 * 60 * 60)) ) REQUEST_TIMEOUT = float(os.getenv("JONASSEARCH_TIMEOUT", "20")) MAX_CONTENT_LENGTH = int( os.getenv("JONASSEARCH_MAX_CONTENT_LENGTH", str(5 * 1024 * 1024)) ) USER_AGENT = os.getenv( "JONASSEARCH_USER_AGENT", "JonasSearchBot/0.1 (+https://jonassearch.local/bot)" ) # ============================================================ # DEPLOYMENT / UPDATE CONFIG # ============================================================ # Version of THIS crawler code. Increased with every released crawler # change; completely independent from the deployment generation (the # generation is a global deployment counter in PostgreSQL, the version # identifies the crawler code). CRAWLER_VERSION = int(os.getenv("JONASSEARCH_CRAWLER_VERSION", "1")) # Exit codes. 42 is reserved exclusively for "I must be updated" so # the launcher can distinguish update, crash and normal exit. EXIT_OK = 0 EXIT_ERROR = 1 EXIT_UPDATE_REQUIRED = 42 # How often the background checker re-reads the generation. It only # sets an internal flag; the actual exit happens at the next safe # commit point, never in the middle of a transaction. UPDATE_CHECK_INTERVAL_SECONDS = float( os.getenv("JONASSEARCH_UPDATE_CHECK_SECONDS", "30") ) # Staging directory for updates: the launcher stages a new version in # releases//crawler.py, verifies it with --self-update and # only then activates it. Relative paths resolve against the crawler # file's directory, so they work from any working directory. _RELEASES_RAW = os.getenv("JONASSEARCH_RELEASES_DIR", "releases") CRAWLER_RELEASES_DIR = Path(_RELEASES_RAW) if not CRAWLER_RELEASES_DIR.is_absolute(): CRAWLER_RELEASES_DIR = ( Path(__file__).resolve().parent / CRAWLER_RELEASES_DIR ) # How long the launcher waits before restarting the crawler after it # exited with EXIT_UPDATE_REQUIRED. LAUNCHER_RESTART_DELAY = float( os.getenv("JONASSEARCH_LAUNCHER_RESTART_DELAY", "2") ) # Only crawl normal HTTP(S) pages. ALLOWED_SCHEMES = {"http", "https"} # Ignore obvious non-HTML resources. BLOCKED_EXTENSIONS = { ".7z", ".avi", ".bin", ".bmp", ".bz2", ".class", ".csv", ".doc", ".docx", ".gif", ".gz", ".ico", ".iso", ".jpeg", ".jpg", ".m4a", ".mkv", ".mov", ".mp3", ".mp4", ".mpeg", ".pdf", ".png", ".ppt", ".pptx", ".rar", ".svg", ".tar", ".tgz", ".tif", ".tiff", ".webm", ".webp", ".xls", ".xlsx", ".xml", ".zip" } # Do not follow these URL schemes. BLOCKED_PREFIXES = ( "mailto:", "javascript:", "tel:", "data:", "ftp:", "file:", ) # Pages with no search-index value that only burn bandwidth, CPU and # index space: pkg.go.dev's reverse-dependency tabs are megabyte-sized # link lists (10k+ links of pure package names). These URLs are # skipped before fetching; the queue item is marked done so it is not # retried endlessly. SKIP_URL_PATTERNS = tuple( re.compile(pattern) for pattern in ( # Query part (not fragment), then the tab value, then only a # fragment or the end may follow. r"^https?://pkg\.go\.dev/[^#]*[?&]tab=importedby(#|$)", ) ) # Keep generated pages reasonably useful for search. MAX_TITLE_LENGTH = 1000 MAX_HEADINGS = 200 MAX_HEADING_LENGTH = 1000 MAX_TEXT_LENGTH = 2_000_000 MAX_LINKS = 10_000 # Cap for the text that feeds the full-text index. PostgreSQL # hard-limits a single tsvector value to 1,048,575 bytes (not # configurable), and the trigger pages_search_vector_update() builds # the vector from the content column. Content near MAX_TEXT_LENGTH # would overflow that limit and abort the whole INSERT with # ProgramLimitExceeded. 200,000 characters are always safe: even with # 4-byte UTF-8 characters this stays well below 1,048,575 bytes. CONTENT_INDEX_MAX_CHARS = 200_000 # Maximum size of a URL (UTF-8 encoded) that may be inserted into the # crawl queue. PostgreSQL B-tree index entries are limited to about 2704 # bytes, so extremely long URLs would abort the INSERT with # ProgramLimitExceeded and crash the worker. Filtering them out before # the INSERT keeps the crawl running without touching the DB schema. MAX_QUEUE_URL_BYTES = 2000 # Rows per INSERT statement when filling the queue. Short statements # keep the lock window on the UNIQUE index small. QUEUE_INSERT_PAGE_SIZE = 500 # Deadlock handling for queue inserts: several independent crawler # instances write overlapping URL batches into the same crawl_queue. # Depending on insert order, PostgreSQL may then abort one transaction # with DeadlockDetected (SQLSTATE 40P01). Instead of letting the worker # crash, the insert is retried a few times with a short, growing # backoff. The deterministic ordering in queue_urls() makes such # deadlocks rare in the first place - without requiring any change to # the database schema or to the existing index. QUEUE_INSERT_MAX_ATTEMPTS = 3 QUEUE_INSERT_RETRY_BASE_DELAY = 0.5 # ============================================================ # URL PRIORITY CONFIG # ============================================================ # The queue is consumed with ORDER BY priority DESC, but so far every # URL was inserted with priority 0, so the sort never differentiated. # The constants below feed calculate_url_priority() (see the "URL # PRIORITY" section): the score prefers the unknown - never-crawled # URLs, new domains, new hosts - and gives hosts with little presence # in the crawl so far a spread bonus, WITHOUT ever blocking a domain. # A heavily crawled host merely loses priority, so crawl capacity # flows to new territory first. # Base score every URL starts with (the old "everything equal" level). PRIORITY_BASE = 100 # URL is not yet present in crawl_queue/pages (strongest single signal). PRIORITY_BONUS_NEW_URL = 200 # Registrable domain has no pages and no queue entries yet. PRIORITY_BONUS_NEW_DOMAIN = 120 # Host is unknown although the domain may already be known (subdomain # or www variant of an existing site). PRIORITY_BONUS_NEW_HOST = 60 # Spread bonus: log2-falloff boost for hosts with a small presence in # the crawl, so many different domains/hosts are reached before the # crawl digs deeper into a single ecosystem. 0 for hosts that do not # exist in the DB yet (those get the explicit new-host bonuses above). PRIORITY_SPREAD_BONUS_MAX = 80 PRIORITY_SPREAD_BONUS_DIVISOR = 20.0 # Source preference, integrated into the new system instead of # replacing it: links out of little-explored territory are promising, # links from a heavily crawled source are pulled slightly back. PRIORITY_SOURCE_NOVELTY_BONUS = 20 PRIORITY_SOURCE_PENALTY_DIVISOR = 10.0 # Same-host link floods (big navigation lists, archives) are flattened, # not blocked: a host with this many known URLs gets the penalty. PRIORITY_CLUSTER_HOST_THRESHOLD = 25 PRIORITY_CLUSTER_PENALTY = 60 # URL variants of already known pages (trailing slash, tracking, # session or pagination parameters) lose against genuinely new pages. PRIORITY_VARIANT_PENALTY = 30 # How many known URLs are sampled for the variant check, so batches # with many known URLs cannot turn into a runtime blow-up. VARIANT_CHECK_SAMPLE_LIMIT = 200 # Query-string keys that mark trivial variants of the same page. VARIANT_QUERY_KEYS = { "session", "sessionid", "sid", "phpsessid", "jsessionid", "aspsessionid", "utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content", "fbclid", "gclid", "ref", "referer", "page", "p", "start", "offset", "from", } # How long a crawler process reuses its aggregate host/domain counts # before reloading them from the database. Each process keeps its own # view; slightly stale counts only shift priorities a little, never # correctness, so no cross-process locking is needed. HOST_STATS_CACHE_SECONDS = 30 # Row cap for the statistics queries in _load_host_stats(). These are # arbitrary samples over tables without host columns (the schema must # not be extended), not complete counts - they only steer priorities, # never block URLs. Missed hosts are treated as unexplored, which errs # in favor of spread. HOST_STATS_SAMPLE_LIMIT = 20000 # ============================================================ # DEPLOYMENT / UPDATE (generation-based rolling updates) # ============================================================ # # The deployment_state table is the global switch for rolling # updates (exactly one row, id = 1): # # CREATE TABLE IF NOT EXISTS deployment_state ( # id INTEGER PRIMARY KEY, # generation BIGINT NOT NULL, # required_version INTEGER NOT NULL # ) # # - generation: global deployment counter. The launcher bumps # it when a new crawler version goes live. # - required_version: the crawler version this generation requires. # # Every crawler instance reads the generation ONCE at startup ("my # generation is N"). Before every relevant PostgreSQL commit # (safe_commit) and additionally in a periodic background check it # compares the current generation with its own: # # - identical -> everything fine, continue. # - different -> the open transaction is rolled back cleanly and the # process exits with EXIT_UPDATE_REQUIRED (42). The # launcher recognizes that exit code, provisions the # new version and starts it; the new process reads the # new generation at startup and runs on. # # No process is ever killed mid-transaction: exits happen only at the # crawler's own safe commit points. # # CRAWLER_VERSION (CONFIG section) is the version of THIS crawler # code. It is NOT the generation and must be increased with every # released crawler change. class UpdateRequired(Exception): """ Raised at a safe point (before a commit) when a new deployment generation is active. safe_commit() rolls the open transaction back BEFORE raising, so no work is ever committed half-done. """ class UpdateSignal: """ Per-process update state (every crawler worker process owns one). generation: the deployment generation this process started with. pending: True once a check noticed a new generation; the process stops at the next safe commit point. stop_checker: tells the background checker thread to finish. """ def __init__(self): self.generation = None self.pending = False self.stop_checker = False def check(self): """Raise UpdateRequired if an update is pending.""" if self.pending: raise UpdateRequired( "new deployment generation active; update required" ) update_signal = UpdateSignal() def ensure_deployment_state_table(conn): """ Idempotent bootstrap for deployment_state (fresh databases): creates the table if needed and makes sure exactly one row with id = 1 exists. Safe to run from several crawler instances at once (CREATE TABLE IF NOT EXISTS + ON CONFLICT DO NOTHING). """ with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS deployment_state ( id INTEGER PRIMARY KEY, generation BIGINT NOT NULL, required_version INTEGER NOT NULL ) """) cur.execute(""" INSERT INTO deployment_state (id, generation, required_version) VALUES (1, 1, %s) ON CONFLICT (id) DO NOTHING """, (CRAWLER_VERSION,)) conn.commit() def _fetch_generation_row(conn): """ Read the deployment_state row (id = 1). If the table does not exist yet, it is bootstrapped (see ensure_deployment_state_table) and the row is read again. Returns the row or None (table without row). """ try: with conn.cursor() as cur: cur.execute( "SELECT generation FROM deployment_state WHERE id = 1" ) row = cur.fetchone() conn.rollback() # read-only; leave no open transaction behind return row except psycopg2.errors.UndefinedTable: conn.rollback() logging.warning( "deployment_state table missing; creating it with " "generation 1 and required_version %d.", CRAWLER_VERSION, ) ensure_deployment_state_table(conn) with conn.cursor() as cur: cur.execute( "SELECT generation FROM deployment_state WHERE id = 1" ) row = cur.fetchone() conn.rollback() return row def get_current_generation(conn=None): """ Read the current deployment generation. Called once per process at startup: this becomes "my generation" (update_signal.generation). """ own_conn = conn is None if own_conn: conn = db_connect() try: row = _fetch_generation_row(conn) if row is None: logging.warning( "deployment_state has no row with id = 1; inserting " "generation 1 with required_version %d.", CRAWLER_VERSION, ) ensure_deployment_state_table(conn) row = _fetch_generation_row(conn) return int(row[0]) finally: if own_conn: conn.close() def periodic_update_check(): """ Regular update check OUTSIDE any transaction, so instances that currently do not commit still notice a new generation. Runs as a daemon thread per worker and uses its own short-lived connection, so it can never interfere with the worker's open transaction. A new generation only sets update_signal.pending; the actual exit happens at the next safe commit point. A failed check (e.g. short DB hiccup) is simply retried next interval. """ while not ( STOP.is_set() or update_signal.pending or update_signal.stop_checker ): time.sleep(UPDATE_CHECK_INTERVAL_SECONDS) if STOP.is_set() or update_signal.pending or update_signal.stop_checker: break try: conn = db_connect() except Exception: continue try: with conn.cursor() as cur: cur.execute( "SELECT generation FROM deployment_state WHERE id = 1" ) row = cur.fetchone() conn.rollback() except Exception: try: conn.rollback() except Exception: pass continue finally: try: conn.close() except Exception: pass if row is None: continue try: current = int(row[0]) except (TypeError, ValueError): continue if ( not update_signal.pending and update_signal.generation is not None and current != update_signal.generation ): update_signal.pending = True logging.warning( "[UPDATE] Deployment generation changed: %d -> %d. " "This worker (version %d) stops at the next safe " "commit point.", update_signal.generation, current, CRAWLER_VERSION, ) def safe_commit(conn, operation=None): """ The ONLY commit path for crawling work. 1. If an update is pending, roll the open transaction back and raise UpdateRequired: the caller's writes are discarded cleanly, nothing is committed half-done. 2. Otherwise run the writes (operation) and commit them. 3. If the operation hits ProgramLimitExceeded (e.g. a tsvector overflow - PostgreSQL hard-limits one tsvector value to 1,048,575 bytes), roll the aborted transaction back and re-raise, so the caller can skip the work item cleanly instead of dying on a poisoned connection. Every crawler commit that does real work goes through here (page saves, queue inserts, queue claim/finish, stats), so a deployment switch can never cut a transaction in half. The one-off startup commits (seeding, stats table) are routed through this function as well. """ if update_signal.pending: try: conn.rollback() except Exception: logging.exception("Rollback before update exit failed.") raise UpdateRequired( "new deployment generation active; update required" ) try: result = operation() if operation is not None else None except psycopg2.errors.ProgramLimitExceeded: # The failed transaction has poisoned the connection ("current # transaction is aborted, commands ignored until end of # transaction block"); roll it back so the connection stays # usable, then let the caller decide (worker: skip the page, # queue_urls: retry the batch). Without the rollback the very # next statement on this connection would fail as well. try: conn.rollback() except Exception: logging.exception( "Rollback after ProgramLimitExceeded failed." ) raise conn.commit() return result # ============================================================ # LOGGING / SHUTDOWN # ============================================================ logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(processName)s] %(levelname)s %(message)s", datefmt="%H:%M:%S", ) STOP = mp.Event() def handle_signal(signum, _frame): logging.info("Shutdown signal received.") STOP.set() signal.signal(signal.SIGINT, handle_signal) signal.signal(signal.SIGTERM, handle_signal) # ============================================================ # DATABASE # ============================================================ def db_connect(): return psycopg2.connect( host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASSWORD, connect_timeout=10, application_name="jonassearch-crawler", ) def ensure_stats_table(conn): """Create crawler_stats. Writes only - commit happens in safe_commit().""" with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS crawler_stats ( id BIGSERIAL PRIMARY KEY, recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(), crawled INTEGER NOT NULL DEFAULT 0, queued INTEGER NOT NULL DEFAULT 0, failed INTEGER NOT NULL DEFAULT 0 ) """) def write_stats(conn, crawled, queued, failed): """Insert one stats row. Writes only - commit happens in safe_commit().""" with conn.cursor() as cur: cur.execute(""" INSERT INTO crawler_stats (crawled, queued, failed) VALUES (%s, %s, %s) """, (crawled, queued, failed)) def get_next_queue_item(conn): """ Atomically claim one pending URL. SKIP LOCKED is the important part: several crawler processes can consume the same PostgreSQL queue without waiting for each other. Writes only - the commit happens in safe_commit(). """ with conn.cursor() as cur: cur.execute(""" WITH next_item AS ( SELECT id FROM crawl_queue WHERE status = 'pending' ORDER BY priority DESC, discovered_at ASC, id ASC FOR UPDATE SKIP LOCKED LIMIT 1 ) UPDATE crawl_queue q SET status = 'processing', attempts = attempts + 1, last_attempt_at = now() FROM next_item WHERE q.id = next_item.id RETURNING q.id, q.url, q.source_url """) row = cur.fetchone() return row def finish_queue_item(conn, queue_id, success=True): """Mark a queue item done/failed. Writes only - commit happens in safe_commit().""" with conn.cursor() as cur: cur.execute(""" UPDATE crawl_queue SET status = %s WHERE id = %s """, ("done" if success else "failed", queue_id)) def requeue_stale_items(conn, older_than_seconds=3600): """ Recover URLs left in 'processing' after a crash. Writes only - commit happens in safe_commit(). """ with conn.cursor() as cur: cur.execute(""" UPDATE crawl_queue SET status = 'pending' WHERE status = 'processing' AND last_attempt_at < now() - (%s * interval '1 second') """, (older_than_seconds,)) changed = cur.rowcount if changed: logging.info("Requeued %d stale processing URLs.", changed) def _insert_queue_rows(conn, rows): """ Insert pre-ordered queue rows and return the number of newly inserted rows. All pages run inside ONE transaction (committed by safe_commit() after this function returns), so the batch stays atomic as before. rowcount is accumulated per page because execute_values() does not aggregate rowcount across pages. """ inserted = 0 with conn.cursor() as cur: for start in range(0, len(rows), QUEUE_INSERT_PAGE_SIZE): page = rows[start:start + QUEUE_INSERT_PAGE_SIZE] execute_values( cur, """ INSERT INTO crawl_queue (url, source_url, priority, status) VALUES %s ON CONFLICT (url) DO NOTHING """, page, template="(%s, %s, %s, %s)", page_size=QUEUE_INSERT_PAGE_SIZE, ) inserted += max(cur.rowcount, 0) return inserted def queue_urls(conn, links, source_url): """ Queue links for crawling, deadlock-safe for parallel crawlers that share one crawl_queue and its UNIQUE index crawl_queue_url_key. Strategy (no DB schema or index changes required): - Clean and deduplicate the links within the batch. - Compute priorities for all URLs (novelty + spread + source preference - repetition penalties) with batched DB lookups only. - Sort by priority, then deterministically by URL, so that concurrent crawlers insert the same URLs in the same order into the UNIQUE index. Lock acquisition on the index then happens in a consistent order, which prevents most lock inversions (deadlocks) at the source. - ON CONFLICT (url) DO NOTHING keeps re-inserts harmless. - If PostgreSQL still detects a deadlock, roll back the aborted transaction and retry the whole batch a few times with a short growing backoff instead of crashing the worker. Any other database error is handled the same way. """ if not links: return 0 # Link URLs also come from the web and can carry NUL bytes. safe_links = clean_text_list(links) safe_source = clean_text(source_url) # Skip URLs that are too large for the UNIQUE B-tree index on # crawl_queue.url (btree entries are limited to ~2704 bytes). filtered_links = [ url for url in safe_links if len(url.encode("utf-8")) <= MAX_QUEUE_URL_BYTES ] skipped = len(safe_links) - len(filtered_links) if skipped: logging.warning( "[QUEUE] %d URLs skipped because they exceed %d bytes", skipped, MAX_QUEUE_URL_BYTES, ) if not filtered_links: return 0 # Priorities are computed BEFORE the deterministic URL sort: the # priority sort decides the crawl order, the final URL ordering # keeps index lock acquisition consistent between crawlers. try: rows, stats = build_queue_rows(conn, filtered_links, safe_source) except Exception: # Priority computation must never take the worker down: fall # back to the previous behavior (everything priority 0) and # still queue the links. logging.exception( "[QUEUE] Priority computation failed; queuing without priorities." ) try: conn.rollback() except Exception: pass unique_links = sorted(dict.fromkeys(filtered_links)) rows = [(url, safe_source, 0, "pending") for url in unique_links] stats = {} if stats: log_queue_stats(safe_source, stats) for attempt in range(1, QUEUE_INSERT_MAX_ATTEMPTS + 1): try: inserted = safe_commit( conn, lambda: _insert_queue_rows(conn, rows) ) if attempt > 1: logging.info( "[QUEUE] Insert succeeded on retry %d/%d.", attempt, QUEUE_INSERT_MAX_ATTEMPTS, ) return inserted except psycopg2.errors.DeadlockDetected as exc: # The failed transaction has poisoned the connection's # current transaction; roll it back so the connection stays # usable for the retry and for everything after it. try: conn.rollback() except Exception: logging.exception("[QUEUE] Rollback after deadlock failed.") if attempt < QUEUE_INSERT_MAX_ATTEMPTS: delay = QUEUE_INSERT_RETRY_BASE_DELAY * (2 ** (attempt - 1)) logging.warning( "[QUEUE] Deadlock while inserting %d URLs " "(attempt %d/%d), retrying in %.1fs: %s", len(rows), attempt, QUEUE_INSERT_MAX_ATTEMPTS, delay, exc, ) time.sleep(delay) except psycopg2.Error as exc: # Any other database error (connection loss, constraint # trouble, ...) must not crash the worker either. Same # controlled retry with backoff as for deadlocks. try: conn.rollback() except Exception: logging.exception("[QUEUE] Rollback after DB error failed.") if attempt < QUEUE_INSERT_MAX_ATTEMPTS: delay = QUEUE_INSERT_RETRY_BASE_DELAY * (2 ** (attempt - 1)) logging.warning( "[QUEUE] DB error while inserting %d URLs " "(attempt %d/%d), retrying in %.1fs: %s", len(rows), attempt, QUEUE_INSERT_MAX_ATTEMPTS, delay, exc, ) time.sleep(delay) # All retries exhausted. Log this clearly and keep the worker # alive; the skipped URLs can be rediscovered from other pages. logging.error( "[QUEUE] Insert of %d URLs from %s failed after %d attempts. " "Sample skipped URLs: %s", len(rows), safe_source or "-", QUEUE_INSERT_MAX_ATTEMPTS, ", ".join(sorted(url for url, _s, _p, _st in rows)[:5]) + ("..." if len(rows) > 5 else ""), ) return 0 # ============================================================ # URL PRIORITY # ============================================================ # Hosts known to serve near-infinite URL spaces (URL floods). Not a # block list: links to these hosts are still queued, they just do not # get preferential treatment for sheer quantity. DIVERSITY_SOFT_HOSTS = {"github.com"} # Cache for aggregate host/domain counts, per process. Multiple # crawler processes/instances each maintain their own view; counts are # advisory only (they shape priorities, never block), so a slightly # stale view between reloads is harmless by design. _HOST_STATS = { "loaded_at": 0.0, "host_counts": {}, # registrable domain -> crawled pages (pages) "host_page_counts": {}, # hostname -> crawled pages (pages) "queue_counts": {}, # hostname -> pending queue entries (crawl_queue) } def _registerable_domain(hostname): """ Registrable domain of a hostname, approximated with the last two labels (example.com, sub.example.org -> example.com). Public Suffix List handling would be more precise but is intentionally out of scope here, matching the existing simple TLD logic in parse_html. """ if not hostname: return "" labels = [label for label in hostname.split(".") if label] if len(labels) >= 2: return ".".join(labels[-2:]) return "".join(labels) def _split_url(url): """ Split a normalized URL into (hostname, registrable_domain) using the same URL processing as the rest of the crawler (urlparse). Falls back to the empty hostname rather than raising so that one malformed URL can never break a batch. """ try: hostname = (urlparse(url).hostname or "").lower().strip(".") except ValueError: return "", "" return hostname, _registerable_domain(hostname) def _strip_tracking_query(url): """ Remove tracking/session-style query parameters; returns None if the URL has no such parameters at all. """ try: parsed = urlparse(url) except ValueError: return None if not parsed.query: return None query_pairs = [ (key, value) for key, value in parse_qsl(parsed.query, keep_blank_values=True) if key.lower() not in VARIANT_QUERY_KEYS ] if len(query_pairs) == len(parse_qsl(parsed.query)): return None query = "&".join( f"{key}={value}" for key, value in query_pairs ) netloc = parsed.netloc stripped = f"{parsed.scheme}://{netloc}{parsed.path or '/'}" if query: stripped += "?" + query return stripped def _is_probable_variant(url, known_urls): """ Heuristic for "URL looks like a trivial variant of an already known page": same address after dropping tracking/session/pagination parameters, or a trailing-slash twin of a known URL. Purely local computation - known_urls comes from a single batched DB lookup. """ if url in known_urls: return True stripped = _strip_tracking_query(url) if stripped is not None and stripped in known_urls: return True if url.endswith("/") and url[:-1] in known_urls: return True if not url.endswith("/") and (url + "/") in known_urls: return True return False def _load_host_stats(conn): """ Refresh the aggregate host/domain view if it is stale. Two read-only grouped queries in ONE transaction, so this cannot interact with concurrent queue inserts: - pages.domain stores the crawl-time hostname; page counts are folded to their registrable domain (and kept per host) in Python. - crawl_queue hosts are extracted from the URL, so the queue counts stay in sync with what is actually pending. """ now = time.monotonic() if now - _HOST_STATS["loaded_at"] < HOST_STATS_CACHE_SECONDS: return try: with conn.cursor() as cur: cur.execute( "SELECT domain FROM pages " "ORDER BY domain LIMIT %s", (HOST_STATS_SAMPLE_LIMIT,), ) domain_counts = defaultdict(int) host_page_counts = defaultdict(int) for (host,) in cur.fetchall(): hostname = (host or "").lower().strip(".") if not hostname: continue host_page_counts[hostname] += 1 domain_counts[_registerable_domain(hostname)] += 1 cur.execute( "SELECT url FROM crawl_queue " "WHERE status = 'pending' " "ORDER BY url LIMIT %s", (HOST_STATS_SAMPLE_LIMIT,), ) queue_counts = defaultdict(int) for (queued_url,) in cur.fetchall(): hostname, _domain = _split_url(queued_url) if hostname: queue_counts[hostname] += 1 except Exception: # Statistics are an optimization. On failure keep the last view # (or an empty one); the crawl continues with slightly worse # priorities instead of losing URLs. logging.exception("[QUEUE] Host statistics could not be loaded.") try: conn.rollback() except Exception: pass _HOST_STATS["loaded_at"] = now return _HOST_STATS["host_counts"] = domain_counts _HOST_STATS["host_page_counts"] = host_page_counts _HOST_STATS["queue_counts"] = queue_counts _HOST_STATS["loaded_at"] = now def calculate_url_priority( url, *, url_known, domain_page_counts, host_page_counts, host_queued, source_domain_pages, batch_host_known, ): """ Central priority computation for one URL. Concept (see the URL PRIORITY CONFIG constants): priority = base + novelty bonuses for unknown URL / domain / host + spread bonus for little-represented hosts + source preference from the previously used rules - repetition/clustering penalties - variant penalty for near-duplicates of known pages All inputs are plain integers looked up beforehand; this function does no I/O of its own so it stays cheap and easy to extend. """ hostname, domain = _split_url(url) priority = PRIORITY_BASE # --- Novelty bonuses -------------------------------------- if not url_known: priority += PRIORITY_BONUS_NEW_URL domain_pages = domain_page_counts.get(domain, 0) host_pages = host_page_counts.get(hostname, 0) host_known = host_pages + host_queued if domain_pages == 0: # Never-crawled domain: the strongest exploration signal. priority += PRIORITY_BONUS_NEW_DOMAIN priority += PRIORITY_BONUS_NEW_HOST elif host_pages == 0: # New host within an already known domain (subdomain or www # variant of an existing site). priority += PRIORITY_BONUS_NEW_HOST # --- Spread bonus ----------------------------------------- # Log2 falloff: a host with little presence in the crawl gets a # strong boost, a huge host none at all. Unseen hosts do not get a # spread bonus (they already receive the explicit new bonuses). if host_known > 0: spread = PRIORITY_SPREAD_BONUS_MAX - ( math.log2(1.0 + host_known / PRIORITY_SPREAD_BONUS_DIVISOR) * PRIORITY_SPREAD_BONUS_MAX / math.log2(1.0 + PRIORITY_SPREAD_BONUS_DIVISOR) ) priority += max(0.0, min(PRIORITY_SPREAD_BONUS_MAX, spread)) # --- Source preference ------------------------------------ # Integrated from the previous behavior: links out of little- # explored territory are promising, links from an exhausted source # are pulled back slightly. if source_domain_pages == 0: priority += PRIORITY_SOURCE_NOVELTY_BONUS else: priority -= ( math.log2(1.0 + source_domain_pages) * PRIORITY_SOURCE_PENALTY_DIVISOR ) # --- Repetition / clustering penalties --------------------- if hostname in DIVERSITY_SOFT_HOSTS: priority -= PRIORITY_CLUSTER_PENALTY elif batch_host_known >= PRIORITY_CLUSTER_HOST_THRESHOLD: # A flood of links to the same host within one page: flatten it # instead of letting one host dominate the next queue window. priority -= PRIORITY_CLUSTER_PENALTY if url_known: priority -= PRIORITY_VARIANT_PENALTY # Keep the value in a sane integer range for the existing column. return int(max(0, min(priority, 2 ** 31 - 1))) def _find_known_urls(conn, urls): """ Look up which of the batch URLs already exist, in ONE query per table using ANY(%s) instead of per-URL queries. Returns the set of URLs that are already known (queued or crawled). """ known = set() sample = urls[:VARIANT_CHECK_SAMPLE_LIMIT] if not sample: return known try: with conn.cursor() as cur: cur.execute( "SELECT url FROM crawl_queue WHERE url = ANY(%s)", (sample,), ) known.update(row[0] for row in cur.fetchall()) cur.execute( "SELECT url FROM pages WHERE url = ANY(%s)", (sample,), ) known.update(row[0] for row in cur.fetchall()) except Exception: logging.exception( "[QUEUE] Known-URL lookup failed; assuming all URLs are new." ) try: conn.rollback() except Exception: pass return known return known def build_queue_rows(conn, links, source_url): """ Turn sanitized, deduplicated links into queue rows ordered by the computed priority (highest first). Returns (rows, stats). Stats is a small dict for batch logging: new_domains / new_hosts / known_hosts / spread_applied / min/max. """ unique_links = sorted(dict.fromkeys(links)) if not unique_links: return [], {} _load_host_stats(conn) domain_page_counts = _HOST_STATS["host_counts"] host_page_counts = _HOST_STATS["host_page_counts"] queue_counts = _HOST_STATS["queue_counts"] # Batched lookups only, never per-URL queries: one known-URL # sample for variant detection plus per-batch host tallies. known_urls = _find_known_urls(conn, unique_links) host_known_in_batch = defaultdict(int) for url in known_urls: hostname, _domain = _split_url(url) if hostname: host_known_in_batch[hostname] += 1 _source_host, source_domain = _split_url(source_url or "") source_domain_pages = domain_page_counts.get(source_domain, 0) rows = [] new_domains = 0 new_hosts = 0 known_hosts = 0 spread_applied = 0 for url in unique_links: hostname, domain = _split_url(url) url_known = url in known_urls or _is_probable_variant(url, known_urls) priority = calculate_url_priority( url, url_known=url_known, domain_page_counts=domain_page_counts, host_page_counts=host_page_counts, host_queued=queue_counts.get(hostname, 0), source_domain_pages=source_domain_pages, batch_host_known=host_known_in_batch.get(hostname, 0), ) # Batch statistics for the [QUEUE] log line. domain_known = bool(domain_page_counts.get(domain, 0)) host_known = bool( host_page_counts.get(hostname, 0) or queue_counts.get(hostname, 0) ) if domain and not domain_known: new_domains += 1 if hostname: if not host_known: new_hosts += 1 elif not url_known: known_hosts += 1 if priority > PRIORITY_BASE: spread_applied += 1 rows.append((url, source_url, priority, "pending")) # Highest priority first, URL as deterministic tiebreaker. The # ordering also keeps concurrent batches inserting in a stable # order, which is what keeps UNIQUE-index deadlocks rare. rows.sort(key=lambda row: (-row[2], row[0])) stats = { "count": len(unique_links), "new_domains": new_domains, "new_hosts": new_hosts, "known_hosts": known_hosts, "spread_applied": spread_applied, "priority_min": min((row[2] for row in rows), default=0), "priority_max": max((row[2] for row in rows), default=0), } return rows, stats def log_queue_stats(source_url, stats): """ One compact log line per batch so the journal later shows whether the crawl actually spreads across more domains/hosts. """ if not stats: return logging.info( "[QUEUE] %d URLs | new domains: %d | new hosts: %d | " "known hosts: %d | priority range: %d-%d | " "spread bonus applied: %d | source: %s", stats.get("count", 0), stats.get("new_domains", 0), stats.get("new_hosts", 0), stats.get("known_hosts", 0), stats.get("priority_min", 0), stats.get("priority_max", 0), stats.get("spread_applied", 0), source_url or "-", ) # ============================================================ # TEXT SANITIZATION # ============================================================ def clean_text(value): """ Remove NUL characters from a string. Web content can contain a literal "\x00" anywhere in the extracted text. PostgreSQL rejects NUL bytes inside ordinary text columns, so a single dirty string would abort the INSERT and crash the worker. Sanitizing centrally before every DB write keeps the crawl running. """ if isinstance(value, str): return value.replace("\x00", "") return value def clean_text_list(values): """Sanitize every string of a list (headings, links, ...).""" return [clean_text(v) for v in values] # ============================================================ # URL NORMALIZATION # ============================================================ def normalize_url(url, base_url=None): if not url: return None url = unescape(url).strip() if base_url: url = urljoin(base_url, url) if any(url.lower().startswith(prefix) for prefix in BLOCKED_PREFIXES): return None url, _fragment = urldefrag(url) parsed = urlparse(url) if parsed.scheme.lower() not in ALLOWED_SCHEMES: return None if not parsed.netloc: return None # Lowercase scheme and hostname. scheme = parsed.scheme.lower() hostname = (parsed.hostname or "").lower() if not hostname: return None # Drop default ports. port = parsed.port if (scheme == "http" and port == 80) or (scheme == "https" and port == 443): port = None if port: netloc = f"{hostname}:{port}" else: netloc = hostname # Preserve path/query, but remove empty query marker. path = parsed.path or "/" query = parsed.query normalized = f"{scheme}://{netloc}{path}" if query: normalized += "?" + query # Avoid obvious files. path_lower = path.lower() for ext in BLOCKED_EXTENSIONS: if path_lower.endswith(ext): return None return normalized # ============================================================ # ROBOTS # ============================================================ class RobotsCache: def __init__(self): self.cache = {} self.last_checked = {} def allowed(self, url): parsed = urlparse(url) host = parsed.netloc.lower() now = time.monotonic() if ( host not in self.cache or now - self.last_checked.get(host, 0) > ROBOTS_CACHE_SECONDS ): self._load(host, parsed.scheme) rp = self.cache[host] try: return rp.can_fetch(USER_AGENT, url) except Exception: return True def delay(self, url): parsed = urlparse(url) host = parsed.netloc.lower() rp = self.cache.get(host) if not rp: return DEFAULT_DELAY # RobotFileParser does not expose Crawl-delay consistently across # all Python versions, so use it when available. try: delay = rp.crawl_delay(USER_AGENT) if delay is None: delay = rp.crawl_delay("*") if delay is not None: return max(DEFAULT_DELAY, float(delay)) except Exception: pass return DEFAULT_DELAY def _load(self, host, scheme): robots_url = f"{scheme}://{host}/robots.txt" rp = RobotFileParser() rp.set_url(robots_url) try: req = Request( robots_url, headers={"User-Agent": USER_AGENT} ) with urlopen(req, timeout=REQUEST_TIMEOUT) as response: data = response.read(512 * 1024).decode( "utf-8", errors="replace" ) rp.parse(data.splitlines()) except Exception: # If robots.txt cannot be retrieved, default to allowing the URL. # This is deliberately conservative with respect to availability. rp.parse([]) self.cache[host] = rp self.last_checked[host] = time.monotonic() # ============================================================ # HTTP # ============================================================ class RateLimiter: def __init__(self): self.last_request = defaultdict(float) def wait(self, url, delay): host = urlparse(url).netloc.lower() now = time.monotonic() wait_for = delay - (now - self.last_request[host]) if wait_for > 0: logging.info("[RATE] %s: %.1fs", host, wait_for) time.sleep(wait_for) self.last_request[host] = time.monotonic() def is_skipped_url(url): """ True for URLs whose pages bring no search-index value but burn significant crawl resources (bandwidth, parsing time, index size). See SKIP_URL_PATTERNS. """ return any(pattern.match(url) for pattern in SKIP_URL_PATTERNS) def fetch(url, limiter, robots): parsed = urlparse(url) if not robots.allowed(url): return None, None, "robots" delay = robots.delay(url) limiter.wait(url, delay) req = Request( url, headers={ "User-Agent": USER_AGENT, "Accept": ( "text/html,application/xhtml+xml," "application/xml;q=0.9,*/*;q=0.1" ), "Accept-Language": "en,de;q=0.8", }, ) try: with urlopen(req, timeout=REQUEST_TIMEOUT) as response: status = getattr(response, "status", 200) content_type = response.headers.get("Content-Type", "") content_length = response.headers.get("Content-Length") if content_length: try: if int(content_length) > MAX_CONTENT_LENGTH: return status, None, "too_large" except ValueError: pass # We only parse HTML/XHTML. ctype = content_type.lower() if ctype and not ( "text/html" in ctype or "application/xhtml+xml" in ctype ): return status, None, "not_html" data = response.read(MAX_CONTENT_LENGTH + 1) if len(data) > MAX_CONTENT_LENGTH: return status, None, "too_large" charset = response.headers.get_content_charset() or "utf-8" text = data.decode(charset, errors="replace") return status, text, None except HTTPError as e: return e.code, None, f"http_{e.code}" except (URLError, TimeoutError, OSError) as e: return None, None, f"network:{e}" except Exception as e: return None, None, f"error:{e}" # ============================================================ # HTML PARSING # ============================================================ def clean_text_bs4(soup): for tag in soup([ "script", "style", "noscript", "template", "svg", "canvas", "iframe", "object", "embed", "form" ]): tag.decompose() # Remove navigation/boilerplate-ish elements where possible. for tag in soup.find_all([ "nav", "footer", "header", "aside" ]): tag.decompose() text = soup.get_text(" ", strip=True) # Normalize whitespace. text = re.sub(r"\s+", " ", text) return text[:MAX_TEXT_LENGTH] def parse_html(html, base_url): if BeautifulSoup is None: raise RuntimeError( "BeautifulSoup fehlt. Installiere python3-bs4." ) soup = BeautifulSoup(html, "html.parser") title = "" if soup.title: title = soup.title.get_text(" ", strip=True) title = re.sub(r"\s+", " ", title)[:MAX_TITLE_LENGTH] headings = [] for tag in soup.find_all(re.compile(r"^h[1-6]$")): value = re.sub(r"\s+", " ", tag.get_text(" ", strip=True)) if value: headings.append(value[:MAX_HEADING_LENGTH]) if len(headings) >= MAX_HEADINGS: break links = [] seen = set() for tag in soup.find_all("a", href=True): normalized = normalize_url(tag.get("href"), base_url) if not normalized: continue if normalized in seen: continue seen.add(normalized) links.append(normalized) if len(links) >= MAX_LINKS: break content = clean_text_bs4(soup) domain = urlparse(base_url).hostname or "" domain = domain.lower() # TLD is intentionally simple here; the DB can later be extended with # Public Suffix List handling if needed. tld = "" if "." in domain: tld = "." + domain.rsplit(".", 1)[-1] return { "title": title, "headings": headings, "content": content, "links": links, "domain": domain, "tld": tld, } # ============================================================ # PAGE STORAGE # ============================================================ def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms): """ Store/refresh one page. Writes only - commit happens in safe_commit(). """ # Sanitize everything that came from the web before it reaches # PostgreSQL. NUL bytes are legal in HTML but illegal in text columns. url = clean_text(url) source_url = clean_text(source_url) content_type = clean_text(content_type) safe_parsed = { "title": clean_text(parsed["title"]), "headings": clean_text_list(parsed["headings"]), "content": clean_text(parsed["content"]), "links": clean_text_list(parsed["links"]), "domain": clean_text(parsed["domain"]), "tld": clean_text(parsed["tld"]), } # Cap the text that feeds the full-text index. The DB trigger # pages_search_vector_update() builds a tsvector from the content # column and PostgreSQL hard-limits one tsvector value to # 1,048,575 bytes (not configurable). Pages near MAX_TEXT_LENGTH # overflow it and abort the whole INSERT (ProgramLimitExceeded), # which used to kill the worker. Capping here keeps oversize # pages saveable (with a truncated index); the trigger-side cap # (fix_pages_search_vector_trigger.sql) adds the same protection # for every other insert path, independent of this code. index_content = safe_parsed["content"][:CONTENT_INDEX_MAX_CHARS] if len(index_content) < len(safe_parsed["content"]): logging.warning( "[SAVE ] %s: content capped for the search index " "(%d -> %d chars).", url, len(safe_parsed["content"]), len(index_content), ) with conn.cursor() as cur: cur.execute(""" INSERT INTO pages ( url, title, headings, content, links, domain, tld, source_url, http_status, content_type, discovered_at, crawled_at, crawl_time_ms ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, COALESCE( (SELECT discovered_at FROM crawl_queue WHERE url = %s), now() ), now(), %s ) ON CONFLICT (url) DO UPDATE SET title = EXCLUDED.title, headings = EXCLUDED.headings, content = EXCLUDED.content, links = EXCLUDED.links, domain = EXCLUDED.domain, tld = EXCLUDED.tld, source_url = EXCLUDED.source_url, http_status = EXCLUDED.http_status, content_type = EXCLUDED.content_type, crawled_at = now(), crawl_time_ms = EXCLUDED.crawl_time_ms """, ( url, safe_parsed["title"], safe_parsed["headings"], index_content, safe_parsed["links"], safe_parsed["domain"], safe_parsed["tld"], source_url, status, content_type, url, crawl_ms, )) # ============================================================ # WORKER # ============================================================ def worker(worker_id): process_name = f"crawler-{worker_id}" mp.current_process().name = process_name conn = None robots = RobotsCache() limiter = RateLimiter() crawled = 0 failed = 0 last_stats = time.monotonic() current_queue_id = None update_requested = False try: conn = db_connect() # Remember the generation this instance started with. Read ONCE # per process: five crawler instances all remember their own # generation (here: the same one, until a deployment bump). update_signal.generation = get_current_generation(conn) logging.info( "Started. DB=%s:%s/%s version=%d generation=%d", DB_HOST, DB_PORT, DB_NAME, CRAWLER_VERSION, update_signal.generation, ) # Regular update check OUTSIDE the transactions (own short-lived # connection, daemon thread). It only sets update_signal.pending; # the exit itself happens at the next safe commit point, never # in the middle of a transaction. checker_thread = threading.Thread( target=periodic_update_check, name=f"{process_name}-update-check", daemon=True, ) checker_thread.start() while not STOP.is_set(): # Safe point between page cycles: if the periodic check # noticed a new generation, stop before starting new work. if update_signal.pending: update_requested = True break item = safe_commit(conn, lambda: get_next_queue_item(conn)) if not item: logging.info("Queue empty. Worker waiting...") time.sleep(5) continue queue_id, url, source_url = item current_queue_id = queue_id logging.info("=" * 70) logging.info("[CRAWL] %s", url) logging.info("[FROM ] %s", source_url or "-") # Known no-value pages (e.g. pkg.go.dev "?tab=importedby": # megabyte-sized reverse-dependency link lists) are # skipped before fetching. Marked 'done' so they are not # retried endlessly. if is_skipped_url(url): logging.info("[SKIP ] %s (no search value)", url) safe_commit( conn, lambda: finish_queue_item( conn, queue_id, success=True ), ) continue started = time.monotonic() status, html, error = fetch(url, limiter, robots) if error: logging.info("[HTTP ] %s -> %s", url, error) safe_commit( conn, lambda: finish_queue_item( conn, queue_id, success=False ), ) failed += 1 continue logging.info("[HTTP ] %s -> HTTP %s", url, status) # A redirect normally arrives as the final URL in urllib. # The requested URL remains the canonical queue identity. if html is None: safe_commit( conn, lambda: finish_queue_item( conn, queue_id, success=False ), ) failed += 1 continue try: parsed = parse_html(html, url) except Exception as exc: logging.exception("[PARSE] %s: %s", url, exc) safe_commit( conn, lambda: finish_queue_item( conn, queue_id, success=False ), ) failed += 1 continue crawl_ms = int((time.monotonic() - started) * 1000) logging.info("[TITLE] %s", parsed["title"][:160]) logging.info("[DOMAIN] %s", parsed["domain"]) logging.info("[HEADS ] %d", len(parsed["headings"])) logging.info("[LINKS ] %d gefunden", len(parsed["links"])) try: safe_commit( conn, lambda: save_page( conn=conn, url=url, source_url=source_url, status=status, content_type="text/html", parsed=parsed, crawl_ms=crawl_ms, ), ) except psycopg2.errors.ProgramLimitExceeded as exc: # Defense-in-depth: save_page() caps the content before # the insert, so this can only fire if the DB-side # trigger overflows anyway (e.g. because # fix_pages_search_vector_trigger.sql has not been # applied yet). safe_commit() has already rolled the # transaction back; mark the queue item failed so the # page is not retried endlessly, and keep the worker # alive. logging.warning( "[SAVE ] %s: not indexable, skipped (%s)", url, exc, ) safe_commit( conn, lambda: finish_queue_item( conn, queue_id, success=False ), ) failed += 1 continue inserted = queue_urls( conn, parsed["links"], url ) logging.info( "[SORT ] %s -> %d Links, %d neue URLs", url, len(parsed["links"]), inserted, ) safe_commit( conn, lambda: finish_queue_item(conn, queue_id, success=True), ) crawled += 1 current_queue_id = None # Periodic stats; not on every page to avoid unnecessary DB load. if time.monotonic() - last_stats >= 30: try: with conn.cursor() as cur: cur.execute( "SELECT COUNT(*) FROM crawl_queue WHERE status='pending'" ) queued = cur.fetchone()[0] safe_commit( conn, lambda: write_stats(conn, crawled, queued, failed), ) last_stats = time.monotonic() except UpdateRequired: raise except Exception: conn.rollback() except UpdateRequired: # A safe point (safe_commit) noticed the new generation and # rolled back the open transaction BEFORE raising. Nothing was # committed half-done. update_requested = True # Best effort: give the claimed item back to the queue so the # next version can crawl it immediately. (Should this fail, # requeue_stale_items() recovers it later anyway.) This commit # is wanted shutdown bookkeeping, not crawled content. try: if current_queue_id is not None: with conn.cursor() as cur: cur.execute( "UPDATE crawl_queue " "SET status = 'pending' " "WHERE id = %s AND status = 'processing'", (current_queue_id,), ) conn.commit() logging.info( "[UPDATE] Queue item %d requeued for the next version.", current_queue_id, ) except Exception: try: conn.rollback() except Exception: pass logging.warning( "[UPDATE] %s stops for update (started with generation %s, " "version %d).", process_name, update_signal.generation, CRAWLER_VERSION, ) except Exception: logging.exception("Worker crashed.") raise finally: update_signal.stop_checker = True if conn is not None: conn.close() logging.info( "Stopped. crawled=%d failed=%d", crawled, failed ) if update_requested: # Reserved exit code, exclusively for "update required". sys.exit(EXIT_UPDATE_REQUIRED) # ============================================================ # STARTUP # ============================================================ def main(): print("=" * 70) print(" JonasSearch Crawler Gamma 0.1") print("=" * 70) print(f"Start: {START_URL}") print(f"Workers: {WORKERS}") print(f"Delay: {DEFAULT_DELAY:.1f} Sekunden") print(f"Robots-Cache: {ROBOTS_CACHE_SECONDS // 3600} Stunden") print(f"DB: {DB_HOST}:{DB_PORT}/{DB_NAME}") print(f"Version: {CRAWLER_VERSION}") print() conn = db_connect() ensure_deployment_state_table(conn) ensure_stats_table(conn) safe_commit(conn) safe_commit(conn, lambda: requeue_stale_items(conn)) # Seed URL. ON CONFLICT makes restarting harmless. queue_urls(conn, [normalize_url(START_URL)], None) conn.close() workers = [] for i in range(WORKERS): p = mp.Process(target=worker, args=(i,), name=f"crawler-{i}") p.start() workers.append(p) update_requested = False worker_crashed = False try: while any(p.is_alive() for p in workers): time.sleep(1) for p in workers: if p.exitcode == EXIT_UPDATE_REQUIRED: # This worker reached its safe point and wants the # update. Stop the remaining workers as well (they # also exit at their own safe points, never # mid-transaction) and propagate exit code 42. if not update_requested: logging.warning( "[UPDATE] Worker %s requests update; the " "other workers stop at their safe points.", p.name, ) update_requested = True STOP.set() elif p.exitcode not in (None, 0): logging.error( "Worker %s exited with code %s.", p.name, p.exitcode, ) worker_crashed = True STOP.set() except KeyboardInterrupt: STOP.set() finally: STOP.set() for p in workers: p.join(timeout=10) for p in workers: if p.is_alive(): p.terminate() p.join() if update_requested: # Reserved exit code: "update required". Only the launcher may # interpret this code; normal errors use EXIT_ERROR (1). print("Crawler stopped: update required.") sys.exit(EXIT_UPDATE_REQUIRED) if worker_crashed: # A normal error must NOT look like a normal end (exit 0) or an # update (42): the launcher restarts on this code unchanged. print("Crawler stopped: worker error.") sys.exit(EXIT_ERROR) print("Crawler stopped.") # ============================================================ # CLI (version / deployment / update checks) # ============================================================ def build_arg_parser(): parser = argparse.ArgumentParser( prog="crawler.py", description="JonasSearch PostgreSQL crawler with " "generation-based rolling updates.", ) parser.add_argument( "--version", action="store_true", help="print the crawler version (just the number) and exit", ) parser.add_argument( "--self-update", type=int, metavar="VERSION", help="verify a staged crawler release in " "%s//crawler.py without activating it " "(exit 0 = startable, 1 = broken/missing)" % CRAWLER_RELEASES_DIR, ) parser.add_argument( "--update-deployment", type=int, metavar="VERSION", help="bump the deployment generation and set required_version " "to VERSION. Run ONLY after the new version has been " "provisioned and verified - from this moment all running " "crawlers stop at their next safe commit point.", ) parser.add_argument( "--init-deployment", action="store_true", help="create deployment_state if needed and seed the initial " "row (id=1, generation 1, required_version = this " "crawler's version)", ) return parser def cmd_version(): # Deliberately machine-readable: the launcher parses this output. print(CRAWLER_VERSION) return EXIT_OK def _basic_self_check(): """ Minimal smoke test used by --self-update when no releases directory exists yet: this file must at least compile. """ try: with open(__file__, "r", encoding="utf-8") as fh: compile(fh.read(), __file__, "exec") except Exception as exc: logging.error("[UPDATE] Self-check failed: %s", exc) return EXIT_ERROR print("[UPDATE] Basic self-check passed.") return EXIT_OK def cmd_self_update(version): """ Verify that a staged release for `version` exists and is startable - WITHOUT activating it. The launcher runs this with the CURRENT (old) interpreter before swapping anything, so a broken release can never become active (fallback guarantee). Checks: 1. releases//crawler.py exists, 2. `python3 releases//crawler.py --version` prints exactly `version` and exits with EXIT_OK (imports, syntax and the version constant of the new file are all checked). """ release_file = CRAWLER_RELEASES_DIR / str(version) / "crawler.py" if not CRAWLER_RELEASES_DIR.is_dir(): logging.warning( "[UPDATE] No releases directory (%s); running the basic " "self-check only.", CRAWLER_RELEASES_DIR, ) return _basic_self_check() if not release_file.is_file(): logging.error( "[UPDATE] Staged release missing: %s", release_file ) return EXIT_ERROR print(f"[UPDATE] Checking staged release: {release_file}") try: result = subprocess.run( [sys.executable, str(release_file), "--version"], capture_output=True, text=True, timeout=60, ) except Exception as exc: logging.error("[UPDATE] Release check could not run: %s", exc) return EXIT_ERROR printed = (result.stdout or "").strip() if result.returncode != EXIT_OK or printed != str(version): logging.error( "[UPDATE] Staged release check FAILED (exit code %d, " "version output %r, stderr: %s). The release is NOT " "activated; the old version stays in place.", result.returncode, printed, (result.stderr or "").strip()[-500:], ) return EXIT_ERROR print(f"[UPDATE] Release {version} verified and startable.") return EXIT_OK def cmd_update_deployment(required_version): """ The global switch point: atomically raise the deployment generation and set the required crawler version in ONE statement. Run only AFTER the new version has been provisioned and verified on the machines (see --self-update). From this moment every running old crawler sees the new generation at its next safe commit point, rolls its open transaction back and exits with EXIT_UPDATE_REQUIRED - no process is ever killed mid-transaction. """ conn = None try: conn = db_connect() ensure_deployment_state_table(conn) with conn.cursor() as cur: cur.execute( """ UPDATE deployment_state SET generation = generation + 1, required_version = %s WHERE id = 1 RETURNING generation, required_version """, (required_version,), ) row = cur.fetchone() if row is None: logging.error( "deployment_state row (id = 1) missing; nothing updated." ) conn.rollback() return EXIT_ERROR conn.commit() generation, activated_version = row print( f"[UPDATE] Deployment generation -> {generation}, " f"required_version -> {activated_version}. Running crawlers " "stop at their next safe commit point with exit code " f"{EXIT_UPDATE_REQUIRED}." ) return EXIT_OK except Exception: logging.exception("Deployment update failed.") if conn is not None: try: conn.rollback() except Exception: pass return EXIT_ERROR finally: if conn is not None: conn.close() def cmd_init_deployment(): """ One-time initialization of deployment_state: create the table and the single current-state row (id = 1) with generation 1 and the currently running crawler version. Restarting is harmless. """ conn = None try: conn = db_connect() ensure_deployment_state_table(conn) row = _fetch_generation_row(conn) generation = int(row[0]) if row else 1 print( f"[UPDATE] deployment_state ready: generation {generation}, " f"required_version {CRAWLER_VERSION}." ) return EXIT_OK except Exception: logging.exception("Initializing deployment_state failed.") return EXIT_ERROR finally: if conn is not None: conn.close() def main_cli(argv=None): args = build_arg_parser().parse_args(argv) if args.version: return cmd_version() if args.self_update is not None: return cmd_self_update(args.self_update) if args.update_deployment is not None: return cmd_update_deployment(args.update_deployment) if args.init_deployment: return cmd_init_deployment() main() return EXIT_OK if __name__ == "__main__": mp.set_start_method("spawn", force=True) sys.exit(main_cli())