# Crawl core — discovery, pipeline states, scheduling, failure policies, repair ``` companies (pending) ─▶ discovery.discover_company ─▶ sensors (pending, staggered next_run_at) scheduler tick ─▶ claim due sensors (SKIP LOCKED) ─▶ pipeline.run_sensor ─▶ observation ─▶ snapshot ─▶ change (pending) ─▶ intelligence ``` ## Discovery (`services/discovery.py`, `catlas discover|onboard`) Per company, within a ~90 s budget and ≤ ~25 requests: homepage (website variants, redirects; off-domain redirect → `domains` row `kind=redirect`, canonical domain updated only when unambiguous and unique) → generic-html extraction of the homepage (nav/footer/main links classified by `urls.classify_url`, `link rel=alternate` feeds, JSON-LD `sameAs`) → `robots.txt` `Sitemap:` lines (else `/sitemap.xml`; index + ≤ 4 children, bounded by `discovery_max_sitemap_urls`) → ATS detection (links, embedded `boards.greenhouse.io/embed/job_board?for=…`, `jobs.lever.co/…`, Ashby, Workable, Workday, Recruitee, Personio, Teamtailor, SmartRecruiters in the homepage + best careers page; the API endpoint is verified with one request) → ≤ 12 common-path probes for missing high-value surfaces (soft-404 and redirect-to-home rejected) → subdomain probes (`careers. jobs. news. blog. docs. developer(s). status. investors. ir. shop.` — DNS first, ≤ 6 GETs; `status.` tries `/api/v2/summary.json`). Selection: one sensor per surface (best `confidence × method weight × verified × depth`), plus `jobs_board` next to `careers` and `feed` next to `blog`/`newsroom`, capped by `discovery_max_sensors_per_company` in surface-importance order. Sensor seed values: `discovery_confidence`, `discovery_method` (nav | link | sitemap | robots | probe | subdomain | ats | feed), `quality_score = 100 × confidence × (0.5 + 0.5 × importance) × reliability`, `base_interval_s = SURFACE_BASE_INTERVAL_S × tier factor (1: ×0.5, 2: ×0.75, 3: ×1, 4: ×1.5)` clamped to `[min_interval_s, max_interval_s]`, `tier` letter from `tier_for_interval`, `next_run_at` random within the first interval (or now with `--fetch-now`), `priority = 0.3 + 0.5 × company importance + 0.2 × surface importance`. Company → `onboarding_status` active / failed (+`onboarding_error`) / no_website; `companies.stats.discovery` and `source_meta.same_as` filled. `onboard_pending` claims `queue_jobs(kind='discover')` with `FOR UPDATE SKIP LOCKED` (attempts / dead after `max_attempts`, 30 min × attempts backoff) and then `companies.onboarding_status='pending'` (marked `discovering` while claimed). ## Pipeline states (`services/pipeline.py`, `catlas run-sensor`) | outcome | rows | sensor | |---|---|---| | `not_modified` (HTTP 304 on etag / last-modified) | observation `not_modified=true` | `consecutive_unchanged++`, interval grows | | `unchanged` (same `normalized_hash` **and** `structured_hash`) | observation `changed=false` | same as above, validators refreshed | | `ok` (first snapshot) | observation, snapshot v1, entities inserted (first_seen) | baseline, no change row | | `changed` | observation, snapshot vN (`previous_snapshot_id`), entity reconciliation, `changes` row | counters, `last_change_at`, burst on meaningful+ | | `failed` | observation with `failure_class`, `failures` row | policy backoff, status transitions, review item | | `redirected` (final registrable domain ≠ sensor domain) | observation `REDIRECT` | `status=redirected`, `config.redirect_url`, `review_queue(sensor_migration)` | | `skipped` | — | domain `blocked_until` / daily budget exhausted → rescheduled | Objects: raw bytes → `archive.put_bytes` (`observations.object_key`), normalized text → `text_key`, blocks JSON → `blocks_key`; `snapshots.extracted = Extraction.structured_payload()` bounded (300 items per list, ≤ 900 KB), `extracted_summary` counters. Change classification: `sdk.diff.compare(previous blocks, new blocks, surface, texts, structured_delta, history)` → `significance` → `taxonomy.change_kind` with `settings.*_threshold` → `changes.kind`; `status='pending'` for meaningful / major / critical (consumed by `services/events`), `'archived'` for minor. Pure noise (no typed delta, `keep_noise_snapshots=false`) writes neither snapshot nor change but updates the sensor hashes so the same noise is not re-detected. Entity reconciliation (scoped to the sensor): jobs by fingerprint (insert / refresh `last_seen_at` / `no_longer_listed` + `removed_at`), people / products / locations by `name_norm`, pricing plans versioned (`superseded` + `valid_to`, `removed`), news by canonical URL. Removals happen only when the extraction is trustworthy (≥ 1 entity extracted or a structured response) and never when an HTML listing drops > 70 % of ≥ 10 entities (parse degradation guard → note in `structured_delta.notes`). `structured_delta` shape is documented in `sdk/models.py`. ## Scheduling formula ``` unchanged / 304: current < base ? min(base, current × burst_decay) : min(max, current × stability_growth) meaningful+: burst_interval_s (then decays back to base, one step per unchanged run) minor: min(current, base) failure: min(max, current × FAILURE_POLICY[class].multiplier) next_run_at: now + interval × U(0.9, 1.1) quality_score: EMA(α = 0.15) towards 100 × extraction confidence (success) or 0 (failure) ``` Scheduler (`catlas schedule`): every `scheduler_tick_s` claim `status in (active, failing, pending) and next_run_at <= now()` not claimed in the last 15 min, ordered by `priority desc, next_run_at`, `limit scheduler_claim_batch`, `FOR UPDATE SKIP LOCKED`; run under `asyncio.Semaphore(fetch_concurrency)` (per-domain spacing / concurrency in `fetch.DomainGovernor`); `crawl_runs` row per tick (`claimed, ok, changed, meaningful, failed, not_modified, unchanged, skipped, redirected, tick_ms`), heartbeat in `settings_kv['scheduler:heartbeat']` (+ per-worker key); onboarding batch every 4 ticks unless `--no-onboarding`; periodic tasks from `services/periodic` (interval + cron in `settings.tz`, APScheduler); SIGTERM/SIGINT release claims. Several schedulers on different machines share one database safely (claims, idempotent runs, no node assumptions). `--once` runs a single tick. ## Failure policies (`taxonomy.FAILURE_POLICY`: multiplier, failures-before-failing) DNS 3.0/3 · TIMEOUT 1.5/4 · HTTP_4XX 2.0/3 · HTTP_5XX 1.5/5 · BOT_CHALLENGE 4.0/2 · PARSING 2.0/3 · SCHEMA 2.0/3 · REDIRECT 2.0/2 · PAGE_REMOVED 4.0/2 · RATE_LIMIT 3.0/4 · ROBOTS 8.0/1 · BLOCKED_DESTINATION 8.0/1 · TOO_LARGE 4.0/2 · UNKNOWN 2.0/3. Status transitions: `failing` at the class threshold → `stale` at `stale_after_failures` (6) → `retired` (+`retired_at`) at `retire_after_failures` (30). `ROBOTS` → `blocked` immediately; `ROBOTS` / repeated `BOT_CHALLENGE` open a `review_queue(kind='blocked_source')` item (never bypassed). Domain budgets (`domain_budgets`: `daily_budget`, `blocked_until`) are honoured before any request; `used_today` resets per day; `cost_ledger` counts fetch units per company / connector and storage. ## Repair (`services/repair.py`, `catlas repair`, periodic every 30 min) For `failing` / `stale` / `redirected` / repeated `PAGE_REMOVED` sensors (≤ 6 requests each): retry the URL (recovered → active, `next_run_at=now`) → candidates for the same surface from the redirect target, the sitemap and the homepage navigation → content-identity check against the last snapshot text (simhash Hamming ≤ 12 or fuzzy ratio ≥ 0.6) → **migrate** (old sensor `retired` with `config.successor_id`, new sensor `pending` with `config.predecessor_id`, `discovery_method='repair'`; history is never deleted) or `review_queue(kind='sensor_migration')` when uncertain. Each attempt is stamped in `config.last_repair_at`. ## Operations cheat-sheet ```bash catlas connectors # registry + connectors table catlas discover https://stripe.com # dry-run surface table (no DB writes) catlas onboard --company --fetch-now catlas schedule --once # one tick (tests / cron); catlas schedule for the daemon catlas run-sensor [--file fixture] [--force] catlas sensors --company [--status failing] [--surface careers] catlas repair --dry-run catlas stats-crawl ```