haiku.rag/docs/ingester.md

Ingester

The ingester is a long-running service that watches sources for changes and feeds documents into haiku.rag's LanceDB. It runs as a separate process (haiku-ingester serve), owns its own SQLite job queue, and exposes a small HTTP control plane for operations.

Use the ingester when:

• you have a corpus you want to keep in sync continuously
• documents arrive over time from filesystem, S3, or HTTP sources
• you want retry + dead-letter behavior, not "fire and forget"

For one-off ingestion, the haiku-rag add-src CLI is enough — see CLI → Add Documents.

Install

The ingester ships behind an optional extra:

pip install 'haiku.rag-slim[ingester]'
# or, for the full package:
pip install 'haiku.rag[ingester]'

That pulls fastapi, uvicorn, aiosqlite, and the [s3] extra. The production binary is haiku-ingester.

Configure sources

Add an ingester: block to your haiku.rag.yaml. The minimum is a single source:

ingester:
  sources:
    - type: fs
      id: local-docs
      root: /Users/you/docs
      delete_orphans: true

Filesystem

ingester:
  sources:
    - type: fs
      id: local-docs                          # optional; auto-derives from root
      root: /Users/you/docs
      poll_interval_s: 300
      delete_orphans: true
      ignore_patterns: ["**/.git/**", "**/node_modules/**"]
      include_patterns: ["*.md", "*.pdf"]    # optional whitelist

Uses watchfiles for push events plus a periodic sweep that catches anything the OS dropped between starts. Patterns follow gitignore syntax.

S3 / object storage

ingester:
  sources:
    - type: s3
      id: corp-docs
      uri: s3://my-bucket/incoming/
      poll_interval_s: 300
      delete_orphans: true
      ignore_patterns: ["draft*"]
      include_patterns: ["*.pdf", "*.md"]
      storage_options:
        endpoint: http://seaweed:8333         # omit for AWS default chain
        aws_access_key_id: ${AWS_KEY}
        aws_secret_access_key: ${AWS_SECRET}
        region: us-east-1
        allow_http: "true"

ETags are the cheap-skip key. Each sweep lists the prefix, compares the listed ETag against the document's stored metadata["source_revision"], and only fetches keys whose ETag has changed. If the bytes turn out to match the stored MD5 (multipart re-upload landing a new ETag on the same content), only the revision is refreshed — no re-chunk.

storage_options follows the same convention as lancedb.storage_options — the dict is passed straight to obstore (the Rust object_store library LanceDB uses internally), so credentials configured for the LanceDB backend can be copy-pasted here.

HTTP

ingester:
  sources:
    - type: http
      id: arxiv
      urls:
        - https://arxiv.org/pdf/2301.12345.pdf
      headers:
        Authorization: Bearer ${SOME_TOKEN}
      poll_interval_s: 86400

HTTP is pull-based with HEAD-driven change detection. A 410 Gone response from a configured URL triggers a delete event; other failure statuses fall through to UPSERT-with-no-revision so the worker can GET and decide.

WebDAV

ingester:
  sources:
    - type: webdav
      id: nextcloud
      base_url: https://nextcloud.example.com/remote.php/dav/files/alice/Documents/
      username: alice
      password: ${NEXTCLOUD_APP_PASSWORD}
      ignore_patterns: ["**/Trash/**"]
      poll_interval_s: 600

Each sweep issues one PROPFIND with Depth: infinity against base_url and parses the multistatus response. Files (non-collection resources) are emitted as UPSERT / UNCHANGED based on the getetag property (falling back to getlastmodified if the server omits it); URIs that were in the previous snapshot but no longer appear under the collection are emitted as DELETE.

Fetches are plain HTTP GETs — any WebDAV server already supports them.

Bearer-token auth can replace HTTP Basic via the standard headers map:

    - type: webdav
      id: kdrive
      base_url: https://kdrive.infomaniak.com/app/drive/123/
      headers:
        Authorization: Bearer ${KDRIVE_TOKEN}

Workers and retry

ingester:
  workers:
    worker_count: 4
    max_concurrent: 4
    poll_idle_interval_s: 1.0
    claim_timeout_s: 1800
    reaper_interval_s: 60
    shutdown_grace_s: 60            # SIGTERM drains in-flight up to this long
    retry:
      max_attempts: 5
      base_delay_s: 2.0
      max_delay_s: 300.0
      jitter: 0.25                  # ±25%

The worker pool runs worker_count async workers behind a shared max_concurrent semaphore. Jobs that hit a TransientError are rescheduled with exponential backoff plus jitter, up to max_attempts, then land in the dead-letter queue. PermanentError (unsupported extension, 4xx HTTP except 408/429, etc.) skips retry entirely.

A reaper task resets jobs whose claimed_at is older than claim_timeout_s so a crashed worker doesn't strand its job.

Backpressure. Each poller skips its periodic sweep when its source already has queued or claimed jobs in the queue. The unique-index dedup would coalesce a re-sweep anyway; the skip saves the listing round-trip (PROPFIND / S3 LIST / FS walk). FS push events from watchfiles still flow during a skipped sweep, so new files aren't lost.

Graceful shutdown. On SIGINT / SIGTERM, pollers stop immediately and workers are given shutdown_grace_s to finish in-flight jobs. Jobs still running after the grace window are cancelled — they stay claimed in the queue and are reset by the reaper on the next start once claim_timeout_s elapses.

Per-source override. A source can opt out of the global retry policy:

ingester:
  sources:
    - type: http
      id: flaky-api
      urls: [...]
      retry:
        max_attempts: 10
        base_delay_s: 10

Circuit breaker

After N consecutive discover() failures, a source's circuit breaker opens and polling pauses for a cooldown. Other sources keep running.

ingester:
  sources:
    - type: http
      id: rate-limited
      urls: [...]
      circuit_breaker:
        failure_threshold: 5
        cooldown_s: 600

Run it

haiku-ingester serve                          # workers + pollers + API
haiku-ingester serve --no-api                 # workers + pollers only
haiku-ingester serve --db /path.lancedb       # explicit DB
haiku-ingester serve --host 0.0.0.0           # bind API on all interfaces
haiku-ingester serve --port 9000              # override API port

--host and --port are CLI overrides for ingester.api.host and ingester.api.port in haiku.rag.yaml. Both default to the YAML value (which itself defaults to 127.0.0.1:8765 — loopback only).

The service blocks until SIGINT or SIGTERM. Shutdown drains the API server, then pollers, then in-flight workers.

Single-writer constraint

LanceDB supports exactly one writer + N readers per database URI. Run exactly one haiku-ingester serve against a given LanceDB. Multiple MCP servers or read-only consumers against the same DB are fine.

HTTP control plane

By default the ingester exposes a FastAPI control plane on 127.0.0.1:8765. Set ingester.api.auth_token to require a Bearer token; without one the API stays open and the service logs a warning.

Method	Path	Purpose
GET	/health	liveness + queue counts
GET	/sources	configured pollers + last-poll time + breaker state
POST	/sources/{id}/refresh	force an out-of-band sweep
GET	/jobs	filtered list (status, source_id, uri, limit, offset)
GET	/jobs/{id}	one job
POST	/jobs/{id}/retry	reset attempts to 0, status to queued
DELETE	/jobs/{id}	cancel a queued/claimed job
GET	/dlq	dead jobs
POST	/dlq/{id}/retry	resurrect from DLQ

OpenAPI docs at http://localhost:8765/docs.

ingester:
  api:
    enabled: true
    host: 127.0.0.1
    port: 8765
    auth_token: ${INGESTER_TOKEN}             # null → unauthenticated

Operating

Smoke-test a single URI

run-once bypasses the queue and runs a single Job through the pipeline. Useful for sanity-checking a source before starting the service.

haiku-ingester run-once /path/to/test.pdf
haiku-ingester run-once https://example.com/spec.pdf
haiku-ingester run-once s3://my-bucket/key.pdf

Exit codes: 0 success, 1 transient error, 2 permanent error.

The queue

The ingester's SQLite queue lives at ~/Library/Application Support/haiku.rag/ingester.db on macOS (platform user data dir; configurable via ingester.queue.path). It's created automatically by serve.

For ops setup you can pre-create it:

haiku-ingester queue init             # create the DB and schema
haiku-ingester queue migrate          # apply pending schema changes

Logs

The service logs via Python logging to stderr through a Rich handler. A typical run looks like:

INFO     Ingester running: 4 worker(s), 1 source(s)
INFO     API listening on 127.0.0.1:8765
INFO     Swept local-docs: 142 upsert, 0 delete, 8 unchanged
INFO     Processing upsert file:///.../a.md (job 5d9a...)
INFO     Job 5d9a... succeeded in 0.34s: file:///.../a.md

When LOGFIRE_TOKEN is set, spans are also shipped to Logfire.

Operating against the API

TOKEN=$INGESTER_TOKEN   # omit -H entirely if no token configured

curl http://localhost:8765/health
curl -H "Authorization: Bearer $TOKEN" http://localhost:8765/sources
curl -H "Authorization: Bearer $TOKEN" 'http://localhost:8765/jobs?status=dead'

# Force a poll now
curl -H "Authorization: Bearer $TOKEN" -X POST \
    http://localhost:8765/sources/local-docs/refresh

# Resurrect a dead job
curl -H "Authorization: Bearer $TOKEN" -X POST \
    http://localhost:8765/jobs/<id>/retry