diff --git a/crawler.py b/crawler.py index 0996ea9..1bcd4d2 100644 --- a/crawler.py +++ b/crawler.py @@ -10,7 +10,10 @@ Features: - Stores title, headings, cleaned text and links - Search vector is maintained by the PostgreSQL trigger - URL deduplication via UNIQUE(url) -- Priority queue +- 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 @@ -34,6 +37,7 @@ Example: import os import re +import math import signal import time import logging @@ -41,7 +45,7 @@ import multiprocessing as mp from collections import defaultdict from html import unescape from pathlib import Path -from urllib.parse import urljoin, urlparse, urldefrag +from urllib.parse import parse_qsl, urljoin, urlparse, urldefrag from urllib.robotparser import RobotFileParser from dotenv import load_dotenv @@ -131,6 +135,93 @@ MAX_LINKS = 10_000 # 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 + # ============================================================ # LOGGING / SHUTDOWN @@ -250,7 +341,56 @@ def requeue_stale_items(conn, older_than_seconds=3600): 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 (commit only at the end), 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) + + conn.commit() + 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 @@ -276,24 +416,484 @@ def queue_urls(conn, links, source_url): if not filtered_links: return 0 - rows = [(url, safe_source, 0, "pending") for url in filtered_links] - - with conn.cursor() as cur: - execute_values( - cur, - """ - INSERT INTO crawl_queue (url, source_url, priority, status) - VALUES %s - ON CONFLICT (url) DO NOTHING - """, - rows, - template="(%s, %s, %s, %s)", - page_size=500, + # 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." ) - inserted = cur.rowcount + 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 = {} - conn.commit() - return inserted + if stats: + log_queue_stats(safe_source, stats) + + for attempt in range(1, QUEUE_INSERT_MAX_ATTEMPTS + 1): + try: + inserted = _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 "-", + ) # ============================================================