Pipeline Workers¶

The system runs four long-lived background workers plus a producer that enqueues work. Workers are independent processes consuming from RabbitMQ queues.

Producer¶

Located in workers/producer.py (generic) or workers/producers/<site>.py (custom).

The producer is responsible for enqueuing URLs to scrape. It connects to the pipeline, creates a job cursor, and publishes URL messages to url.queue.

graph LR
    P[Producer] -->|"generates URLs"| UQ[url.queue]
    UQ --> DW[dedup_worker]

Generic producer — used for simple sites where URLs are template-based. The urls or url_template in the profile defines the target URLs.

Custom producers — for complex sites requiring browser bootstrap sessions, dynamic pagination, or API token capture:

Producer	Custom Logic
`idealista.py`	Dynamic pagination, location variants, inline detail follows
`g2.py`	Category-based listing URLs
`hermes.py`	Category navigation
`trustpilot.py`	Business/review page generation
`booking_com.py`	Browser bootstrap → GraphQL token capture
`facebook_marketplace.py`	Browser bootstrap → session cookie capture

Dedup Worker¶

(workers/dedup_worker.py)

Consumes from url.queue, deduplicates URLs using RedisBloom, and publishes unique URLs to fetch.queue.

# Dedup uses a Bloom filter for memory-efficient set membership
# False positives are possible but rare — trade-off for O(1) check
bloom = BloomDedup(redis_client, "dedup:my_job")
if not await bloom.check(url):
    await bloom.add(url)
    await publish(channel, "fetch.queue", message)

Fetch Worker¶

(workers/fetch_worker.py)

Consumes from fetch.queue, fetches URLs using the configured fetcher plugin, and publishes raw responses to raw.queue.

Key responsibilities: - Browser pool management — creates and reuses CloakBrowser sessions - Anti-bot evasion — patched Chromium fingerprints, residential proxies, warmup pages, automatic session rotation on block detection - Rate limiting — per-stage rate limiter to avoid triggering rate limits - Retry logic — configurable retry with backoff for transient failures - Block detection — heuristics to detect challenge/block pages and rotate IPs

graph LR
    FQ[fetch.queue] --> FW[fetch_worker]
    FW --> BP[Browser Pool]
    BP -->|"session"| CB[CloakBrowser / wreq]
    CB -->|"HTML/JSON"| RQ[raw.queue]

Two fetch worker replicas run in production, with per-stage concurrency and pool_size automatically divided by the replica count to keep the total browser fleet within configured limits.

Parse Worker¶

(workers/parse_worker.py)

Consumes from raw.queue, parses HTML/JSON using the configured parser plugin, and publishes structured items to parsed.queue.

graph LR
    RQ[raw.queue] --> PW[parse_worker]
    PW -->|"selectolax / jsonpath"| Items[structured items]
    Items --> PQ[parsed.queue]

The parse worker: 1. Looks up the stage's parser plugin from the profile registry 2. Calls parser.parse(raw_response) to extract structured data 3. Applies field validation (required fields, drop/flag strategy) 4. Handles inline fan-out (single page → multiple items) 5. Publishes each item batch to parsed.queue

Store Worker¶

(workers/store_worker.py)

Consumes from parsed.queue and persists items to configured storage backends.

graph LR
    PQ[parsed.queue] --> SW[store_worker]
    SW --> S3[Garage S3]
    SW --> PG[PostgreSQL]
    SW --> J[JSON / CSV / Markdown]

Features: - Batch writing — items are buffered and flushed on batch size or idle timeout - Multi-backend — items go to all configured storage targets simultaneously - Dedup at storage — source_url is the primary key in Postgres - Parquet format — S3 storage uses PyArrow for columnar output

Worker Communication Pattern¶

Workers communicate only through queues. No worker talks to another worker directly:

Producer ──▶ url.queue ──▶ Dedup ──▶ fetch.queue ──▶ Fetch ──▶ raw.queue ──▶ Parse ──▶ parsed.queue ──▶ Store

This means: - Any worker can be restarted independently - Workers can be scaled horizontally (fetch has 2 replicas) - Backpressure is natural — if a downstream worker is slow, its input queue fills up - Dead-letter queue catches messages that can't be processed