Phase-1 decentralized catalog crawler (onix plugin + module) - #894
Draft
ameersohel45 wants to merge 33 commits into
Draft
Phase-1 decentralized catalog crawler (onix plugin + module)#894ameersohel45 wants to merge 33 commits into
ameersohel45 wants to merge 33 commits into
Conversation
… onix plugin) Add the Phase-1 catalog crawler as a framework-agnostic engine (pkg/catalogcrawler) plus a thin onix plugin that adapts it, per the locked design/plan. Engine (pkg/catalogcrawler, no core/module|pkg/plugin imports): - change detection by version cursor (sync / skip-unchanged / rollback / retire) - selection: public + configured networkIds -> publishDirectives.visibleTo - resolve full catalog from baseline + change files (reuses pkg/catalogfile) - push builder: publishDirectives updateMode=FULL + visibleTo - retry backoff + sync_status rollup (ok/partial/failed) - Postgres state: crawler_index / crawler_queue / crawler_catalog + FOR UPDATE SKIP LOCKED claim + coalescing enqueue + transactional complete - source resolution: config / registry / on-demand - two scheduled jobs (index + catalog) linked by the queue - HTTP client: digest verify, SSRF guard, size cap - config from env; standalone driver cmd/catalog-crawler onix integration (thin plugin over the engine): - Crawler plugin interface -> Start/Stop/CrawlNow - plugin adapter builds+starts the engine, injects schemav2validator - /crawl handler (202 on-demand trigger), replacing the inline catalog/pull - crawl module wiring: handler type, manager, config, module map, yaml block Phase-1 decisions reflected in code: signature verification kept but flag-off; resolve-full -> FULL replace (removals handled by replace); crawler-side schema validation of the /push body against the catalog/publish action; queue keyed by catalog_id (file urls read from the index); all config-driven, secrets via env. Grounded against the File Specifications example (golden model_test); example identifiers kept neutral. Tests: TDD throughout. DB-backed tests skip without CRAWLER_TEST_DB_DSN and use a per-package Postgres schema so parallel `go test ./...` stays isolated.
Close the design gaps found in review (all three were already in the
design/plan — this is code catching up):
- Metrics: inject a Metrics sink (NopMetrics default; OTel impl) — pushed and
failed{reason} counters, queue_depth gauge, push + index histograms.
- Batch push: split a resolved catalog by resource count; first batch FULL,
the rest MERGE, posted sequentially, stopping on the first failed batch
(single call below the threshold). sync_status rolled up over batch outcomes.
- next_crawl_at: the index job skips indexes not yet due (per-index cadence);
/crawl forces an immediate pass. GetIndex now reads next_crawl_at.
Plan (self-review) updated: these three marked done; the Registry HTTP client
and the scope filter (schemaTypes subset of approvedScopes) marked explicitly
deferred.
TDD: BatchCatalog unit test + engine metric assertion. Whole repo green; .so builds.
…LL removals)
Discovery's /push MERGE upserts by id and never deletes (only FULL does
deleteByCatalogId, confirmed in catalog-publish-job/PersistenceStep) — so the
previous always-FULL push was wrong: it delete+reinserted the whole catalog on
every change. The catalog job now picks the mode from what actually changed:
- ResolveWithChangeset folds the complete catalog AND records the changeset
since the cursor: upserted ids, whether any removal happened, and a
baseline-fallback flag (new / cursor behind the current baseline).
- only upserts -> MERGE, pushing catalog metadata + just the changed
resources/offers (filterCatalog).
- any removal, or new / re-baselined -> FULL (the complete catalog; replace).
- BatchCatalog now takes a base mode and keeps catalog metadata on every batch
(fixes a latent bug where MERGE batches dropped descriptor/provider/etc.).
Docs: design + plan updated to the mode-by-changeset push model.
TDD: ResolveWithChangeset, filterCatalog, and BatchCatalog(base-mode) tests.
Whole repo green; .so builds.
…safety, SSRF) External review found real concurrency/data-loss issues. Fixes: - #1 queue race: add a claim_id token; ClaimNext stamps it, Complete/ FailQueueItem settle only the holder's claim, and a coalescing Enqueue no longer resets an in-progress row (only bumps to_version). Complete releases (not deletes) a superseded row -> no lost update / double-push. - #2 data loss: the version cursor now advances ONLY on a successful push. failItem never advances it; persistent failures retry at capped backoff and are recorded via RecordFailure (no version write) -> an outage longer than MaxAttempts can't silently drop catalogs. - #3 orphaned claim: ClaimNext reclaims claims older than a lease (15m). - #4 CrawlNow use-after-close: runs on the engine ctx, tracked in wg, drained by Stop; the /crawl handler no longer detaches on context.Background(). - #5 interval panic: clamp IndexInterval/CatalogInterval in New (no NewTicker(0)). - #7 digest fail-open: FetchFile rejects an empty digest (integrity mandatory). - #8 change continuity: ResolveWithChangeset requires fromVersion == running; a gap fails the resolve instead of mis-composing. - #6 SSRF: CheckRedirect re-runs the host guard on every hop; deny CGNAT. - #10 telemetry: PluginEntries() reports the crawler slot. Also: BatchCatalog keeps catalog metadata on every batch (MERGE batches no longer drop descriptor/provider). Deferred (documented): consolidate onto pkg/security/artifactfetcher (#9), dial-time IP pinning for DNS-rebind, enforce validUntil (P2). Docs (design + plan) updated. TDD: claim-token/lease/supersede + continuity + digest-fail-closed tests. Whole repo green; .so builds.
The review-fix commit accidentally reformatted stdHandler.go, healthcheck*.go, http_metric.go and config_test.go via 'gofmt -w core/module/handler/'. Those are whitespace-only and unrelated to the crawler — reverted to keep the PR scoped to crawlHandler.go + config.go.
Add .json.gzip as a second catalog/index format alongside .json, built as an extensible codec registry so zstd/brotli are one entry + one decoder later (fetch/verify/schema/push untouched). - codec registry: compiled-in encoding->decoder map (not config- selectable — a decoder inflates untrusted bytes, so it is trust surface). Single decode() choke point applies a separate decompressed cap (decompression-bomb guard: reject, don't truncate). - format selection: explicit FileEntry.Encoding wins, else inferred from the .json.gzip / .json.gz URL suffix. - digest-before-inflate: verify the sha-256 over the compressed bytes at rest before spending CPU decoding; transport compression disabled so hashed bytes == artifact bytes (no digest-over-decompressed ambiguity). Only authenticated bytes are ever decoded. - failure classification: unsupported encoding, decompression bomb, corrupt/oversize/continuity-gap, schema-invalid, and 4xx rejections are now PERMANENT — parked (status='failed', next_attempt_at= infinity), cursor not advanced, logged at error level. A version bump re-activates the parked row, so a re-published catalog recovers automatically. Transient errors (network/5xx/timeout) still retry at capped backoff. Config: CRAWLER_MAX_DECOMPRESSED_BYTES (default 100 MiB). Tests: codec unit tests (passthrough, gzip round-trip, bomb-at-cap, corrupt, unknown-encoding, additive-codec, encodingFor); an end-to-end gzip fetch test (suffix + explicit encoding, digest over compressed bytes); and state TestParkAndRecover (park SQL + version-bump recovery). Full module go test ./... green; plugin .so builds.
Avoid re-downloading an unchanged index. The index job now sends the stored ETag / Last-Modified as If-None-Match / If-Modified-Since; a host that honours them answers 304 Not Modified and we skip the download entirely, advancing only the crawl cadence. Opportunistic with graceful fallback — a host that sends no validators simply returns 200 and we fall back to the existing version-based change gate, so correctness never depends on the headers. - migration 0003: crawler_index gains nullable etag / last_modified. - state: IndexState carries the validators; GetIndex/UpsertIndex read and store them; new TouchIndex bumps last_crawled_at + next_crawl_at on a 304 without touching version or validators. - http: FetchIndex takes an IndexCond and returns an IndexResult (NotModified + captured validators); getCond sends the conditional headers and detects 304. The catalog job fetches unconditionally (it always needs the body to resolve the entry). - engine: crawlIndex sends the stored validators, short-circuits on 304 (log + TouchIndex + return), and stores fresh validators on a 200. Plugin + driver unchanged — they pass httpc.FetchIndex as a method value, which follows the new signature. Tests: TouchIndex; the full 200->capture->304 HTTP round-trip; the no-validator 200 fallback; and an engine 304 pass (echoes the stored ETag, enqueues nothing, keeps the version, advances the cadence). Full crawler + state suites green against Postgres; module-wide go test green.
Two related crawler improvements (kept in one commit — both touch
engine.go/state.go and don't split into per-file commits that each
compile).
1) Detailed per-pass report. crawler_catalog.push_status stops being a
single value and becomes a jsonb ARRAY of pass records, appended on
every settle and capped to the most recent 20. Each element:
{ts, from, to, mode, resources, offers, removals, batchesAcked,
batchesTotal, outcome, httpStatus, reason}. outcome is one of
pushed | partial | failed | rejected | skipped. reason/http_status
still mirror the latest pass for cheap top-level queries; status
(active/retired) is unchanged. A partial push (batchesAcked <
batchesTotal) is now visible. Migration 0004 converts push_status
text -> jsonb, guarded so Migrate can replay it idempotently.
GetCatalogReports reads the array. Run totals ("how many catalogs /
resources") are an aggregate over the latest element.
- resolve.go: Changeset now counts RemovedResources/RemovedOffers.
- engine.go: builds a PassReport on success / retire / push-fail;
failReport() builds the minimal report for pre-push failures;
fail/failItem/failPermanent carry the report.
2) Size-based push batching. BatchCatalog packs resources by SERIALIZED
byte size instead of a fixed resource count, so a batch of large
resources can't exceed Discovery's payload cap and get 400'd. Config
CRAWLER_PUSH_BATCH_SIZE (count) -> CRAWLER_MAX_PUSH_BYTES (bytes,
default 10 MiB). The engine passes a doc budget = cap - envelope
headroom (4 KiB) - len(visibleTo). FULL lead (keeps offers) then
MERGE rest is preserved; a single resource over the budget still
forms its own batch (can't split) and its reject surfaces in the
pass report.
Tests: state append/cap/partial history against Postgres; engine pass
recorded with counts/mode; BatchCatalog byte-budget splits + oversize
single resource. gofmt/vet/build clean; full crawler + state suites and
module-wide go test green; plugin .so builds.
… entry The Discovery /push CatalogPublishAction schema REQUIRES catalogType on every publishDirective (beckn.yaml: publishDirectives.items.required = [catalogId, catalogType]; catalogType enum = MASTER | REGULAR). BuildPushBody omitted it, so every push was rejected with 400 SCH_SCHEMA_VALIDATION_FAILED. Caught by a local end-to-end run against the discover docker stack. Thread it through: PushMeta gains CatalogType, the engine sets it from the index CatalogEntry.CatalogType, and BuildPushBody emits it in the directive. Forwarded verbatim (no fabricated default) — a publisher that declares no/invalid catalogType is surfaced as a reject in the pass report, not silently defaulted. Verified: local crawl now pushes 200 for both a plain-JSON catalog and a json.gzip catalog; resources land in Elasticsearch. TestBuildPushBody asserts the directive carries catalogType. gofmt/build/vet clean; full crawler + state suites green.
This version pushes updates as MERGE only and builds the delta straight from the change files — FULL and removals are deferred to a later version (in both the crawler and the publish job). - CRAWLER_MERGE_ONLY (default true): always updateMode=MERGE. - ResolveDelta builds the payload from the change files' `catalog` metadata envelope (id/descriptor/provider — schema-required) + the union of resource/offer upserts (latest per id), with NO baseline fetch, for an incremental update. A first sync (cursor behind the baseline) or a change set without the metadata envelope falls back to a full resolve, still pushed as MERGE. - Removals are recorded in the changeset and logged (crawler.catalog.removals_skipped) but NOT applied; a change with nothing to upsert (e.g. removal-only) is skipped and the cursor still advances (completeSkipped, outcome=skipped). - The mode-by-changeset FULL path (resolvePush with MergeOnly=false) is kept dormant behind the flag for the next version. Grounded: publish job has no delete-by-id (MERGE upserts only, FULL wipes by catalog_id) and beckn.yaml Catalog requires id+descriptor+provider — so removals need FULL, and a MERGE delta must carry the metadata envelope. Tests: TestResolveDelta (delta built from change files, baseline not fetched, removals recorded-not-applied, ok=false without envelope); TestEngine_MergeOnly_DeltaPush (incremental → MERGE, baseline NOT fetched, delta-only resources, cursor advances). gofmt/build/vet clean; full crawler + state suites and module-wide go test green.
Domain-oriented (~6-7 pkg) layout with intent-revealing names, a two-PR plan (structural PR-A / observability PR-B), a function/type/package naming review (fix names that lie), file-cohesion contract, content schema-validation seam, and test-suite organization. Synthesis of a three-lens architecture review; packages decode/ and telemetry/ adopted.
…eady log spec Add §6b "Define the lifecycle explicitly": the four scattered status/outcome vocabularies (retry.go, model.go, engine.go, state/) consolidated into typed enums placed by layer — catalog/lifecycle.go (Outcome, CatalogStatus, Decision, DropReason) at the shared bottom, runner/lifecycle.go (DaemonState, PassState, PassKind, CatalogPhase) as orchestration-only — with classifyOutcome/Permanent collapsing the 3x-duplicated "4xx => rejected" rule and the ACTIVE/active casing bug. Rejects phase-named packages (cohesion over temporal grouping). Add §9b "Log specification (implementation-ready)": the exact log contract (~15 INFO/WARN/ERROR events, down from ~25) with keys, levels, required fields, and correlation (run_id/pass_id) — success is a trace, health is a metric, only exceptions and lifecycle boundaries are logs; ~10 *_failed DB logs collapse to crawler.store.unhealthy. Wire acceptance criteria for both.
… add phase-1 + gzip review docs - README: update to the current index/catalog scheduled-job wiring, config keys, on-demand /crawl, source resolution, and phase-1 verification scope. - add catalog-crawler-phase1-review.md (correctness findings). - add catalog-crawler-gzip-support-review.md (json.gzip decode design).
…subject
Fix the subject-less "crawler started/running/completed" framing: a lifecycle
belongs to a unit of work, not to the long-running process. Rework §6b and §9b
around a Catalog Sync (catalog_id + fromVersion->toVersion) as THE lifecycle —
started -> running(resolving/verifying/validating/scoping/publishing) -> done AS
a named terminal outcome (pushed|partial|skipped|dropped|retired|faulted). The
crawler process is a supervisor (ready->stopping->stopped, no "completed"); the
index crawl is the feeder.
- §6b: drop PassState{started,running,completed,failed}; enums regrouped by whose
lifecycle they describe (SyncOutcome/FaultClass/DropReason/CatalogStatus in
catalog/, SyncPhase/IndexOutcome/DaemonState in runner/); transitions redrawn.
- §9b: every log line now carries explicit lifecycle + state fields so the
lifecycle is legible on the line; catalog regrouped (sync exceptions, sync
batch, index crawl, daemon, store); ~13 events; rendered INFO vs DEBUG streams.
- §9: span tree + metrics realigned (crawler.sync/.index_crawl, crawler.sync.*).
…ecycle vocabulary Behavior-preserving restructure of the flat pkg/catalogcrawler into the domain-oriented layout from docs/catalog-crawler-restructure-cr.md (PR-A); telemetry (§9/§9b — instruments, log catalog, run_id/pass_id) is deferred to PR-B and untouched. Structure (flat -> 8 packages + composition root): - catalog/ pure domain (imports only stdlib + catalogfile): index, signature, visibility, decide, eligibility, compose (incl. ResolveDelta), fault, lifecycle - fetch/ client, guard, integrity, conditional - decode/ registry, gzip - publish/ request, batch, client (Push + BatchOutcome moved here from http.go; docCounts/ackedCount here) - source/ static, registry - store/ open, queue, cursor, indexstate, scan (one package, shared tx) - runner/ runner, indexpass, syncpass, backoff, ports, lifecycle (engine.go split by job) - config/ settings - catalogcrawler.go composition root (New(Settings, Options)); telemetry.go plumbing Renames (§5a, compiler-checked): failItem->scheduleRetry, failPermanent-> parkPermanently, fail->routeFailure, failReport->newFailureReport, FailQueueItem->RescheduleQueueItem; Source.IndexURLs->IndexRefs; PartOutcome->BatchOutcome; engine Config->EngineConfig; IndexCond-> IndexConditions; getCond->getConditional; TouchIndex->AdvanceIndexCadence; processItem split into handleQueueItem + syncCatalog. Lifecycle (§6b, recentered on the Catalog Sync as the subject): - catalog/lifecycle.go: SyncOutcome (pushed/partial/skipped/dropped/retired/ faulted — failed/rejected folded into faulted+FaultClass; dropped reserved for scope, a P2 path), CatalogStatus, Decision, DropReason; each String() is the stable wire value. - catalog/fault.go: FaultClass + PermanentError/IsPermanent + Permanent() (drives park-vs-retry). - runner/lifecycle.go: SyncPhase (resolving/verifying/validating/scoping/ publishing), IndexOutcome (unchanged/enqueued/failed), DaemonState (ready/ stopping/stopped/start_failed) + declared legal transitions; classifyOutcome (single home for the 4xx=>rejected/faulted rule). PassState/PassKind/ CatalogPhase removed (the subject-less framing). - Service is pre-production, so the persisted pass-report outcome adopts the new vocabulary outright; external behavior (what's pushed, retry-vs-park, cursor advancement) and the emitted metrics/logs are unchanged. Tests moved with their code (catalog/ white-box; adapters black-box + export_test where needed; store/ Postgres-gated per-package schema; runner/ in-memory port fakes). Plugin + cmd rewired to the composition root. Verified static (gofmt/build/vet clean; all sub-packages + module-wide go test green; plugin .so links) and live end-to-end against the real FreshMart catalog (MERGE push, 4 resources + 4 offers, data in Elasticsearch, pass report reads the new SyncOutcome).
…, audit hardening, file docs
Multi-agent audit + fix pass over the restructured crawler.
Observability / lifecycle (§6b/§9b), recentred on the Catalog Sync as the subject:
- Every log line now carries lifecycle + state (event key = crawler.<lifecycle>.<state>)
plus run_id (per tick) and pass_id (per catalog); ~26 ad-hoc events -> ~13 typed
INFO/WARN/ERROR events; success terminals + running sub-states are DEBUG/trace.
- Collapsed ~13 store *_failed logs into one crawler.store.unhealthy{operation}.
- Daemon lifecycle centralised in the composition root: crawler.daemon.ready{effective
config} / start_failed{stage} / stopping / stopped — identical contract for both drivers.
- SyncPhase + IndexOutcome now driven (were dead types); Debug added to the Logger port.
- Metric label for failures is now the typed FaultClass.String() (was reasonCategory).
Protocol fix (Discovery contract): context.action was catalog/publish (a provider->CS
action); the crawler calls Discovery's /catalog/push, so the correct Beckn action is
catalog/push (spec v2.0.0 beckn.yaml const catalog/push). Fixed in publish/request.go +
tests + docs; removed the fabricated CRAWLER_SCHEMA_ACTION README row (no such env exists).
Audit hardening:
- SSRF DNS-rebinding (TOCTOU): fetch now validates the IP at dial time via a custom
DialContext and dials the validated IP directly, closing the gap where a second,
independent resolution could reach an internal address. isPublicIP is the one source of truth.
- ClassifyFault: 408/425/429 are retryable, no longer parked as permanent.
- config: clamp FetchTimeout/IndexInterval/CatalogInterval/MaxAttempts <=0 to defaults
(a 0 fetch timeout on untrusted URLs was a real footgun).
Docs: clear top-of-file purpose header on every crawler source file (newcomer-friendly).
Behavior otherwise unchanged: same DB writes, push calls, retry/park, cursor advancement,
304 handling. Build/vet/gofmt/test green across all catalogcrawler packages + cmd.
Full rename per the CR decision: pkg/catalogcrawler -> pkg/crawler, cmd/catalog-crawler -> cmd/crawler, pkg/plugin/implementation/catalogcrawler -> .../crawler, plugin id + built .so catalogcrawler.so -> crawler.so (install/build-plugins.sh, config yaml), and all import paths, package clauses, error-message prefixes, and doc code-path references. Subpackages (catalog, fetch, decode, publish, source, store, runner, config) unchanged under the new root. Mechanical only — build/vet/gofmt/test green.
Rename the domain vocabulary and the two job files to say what they do:
- catalog.Decide -> catalog.DetectChange (change detection)
- catalog.Select -> catalog.ResolveScope (network-scope resolution)
- decide.go / eligibility.go / compose.go -> change.go / scope.go / resolve.go
- runner/indexpass.go -> crawl.go, runner/syncpass.go -> sync.go
('pass' read like pass/fail; the files hold the crawl job and the sync job)
Pure renames; behaviour unchanged (build/vet/test green).
syncCatalog becomes an 8-line stage list read top to bottom (resolveEntry -> fetchContent -> verifyContent -> scope -> batch -> publish -> complete) driven by runPipeline; a stage is one line to insert. crawlIndex becomes a linear spine (load -> dueForCrawl -> conditional fetch -> 304? -> processIndex -> recordIndex) with decideCatalog per catalog. Behaviour-preserving: same fetch/verify/scope/push, same phase breadcrumbs, same fault routing, same metrics/logs (every external call preserved 1:1). Verified: gofmt, build, vet, go test ./pkg/crawler/...
Review of pkg/crawler surfaced no correctness bugs (the crawl/sync refactor
was verified faithful). Applied cleanups:
Comments/naming — stale after the crawl.go/sync.go + DetectChange/ResolveScope rename:
- runner.go/index.go/lifecycle.go: drop stale indexpass/syncpass + eligibility
references; correct the DropReason doc
- sync.go: pipeline stage scope -> resolveVisibility (it resolves visibleTo;
membership is decided upstream in decideCatalog); helper resolvePush -> buildPushDoc
Observability:
- catalogPass: count OutcomePartial as 'retried', not 'faulted' (a partial push
is a retry-in-progress); additive 'retried' field on crawler.sync_pass.completed
De-duplication (drift risk):
- sync.go: extract settle() shared by complete/completeSkipped/retire
- publish: Rollup returns acked; delete AckedCount
- catalog/resolve.go: ResolveDelta reuses accumulateChangeset
- crawl.go: crawlResult.checked -> fetched (+ log field indexes_fetched)
Deliberately NOT applied (would regress): DetectChange->*int64 (nil-deref risk),
readCapped cross-pkg helper (couples fetch/decode), RecordFailure/upsertCatalog
SQL merge (DB-write regression risk). The dead-code report was a false positive:
every flagged symbol is test-covered / reserved P2 vocabulary.
Verified: gofmt, build, vet, go test ./pkg/crawler/... all green.
One process, two jobs → three log components (daemon/crawl/sync). Every line is component + stage + natural message + attributes + stats. Documents the stages, messages, levels, the fault->step mapping, reading/filtering, and the known Phase-2 gaps (removals-deferred 'skipped', retirement-not-propagated 'retired').
Implements docs/crawler-logs.md. Behaviour-preserving (only log output changes). - Collapse the 7 lifecycle values to 3 components: daemon / crawl / sync. - Every line: component + stage + natural message + attributes + stats. - crawl stages: polled / queued / finished / failed. sync stages: syncing / synced / skipped / retired / failed / finished. Failure message says WHERE it broke (fault->step) and WHETHER it recovers. - Collapse the sync phase breadcrumbs into one 'syncing' line; keep the readable syncCatalog pipeline. Drop the SyncPhase transition machine. - catalog/lifecycle.go -> catalog/status.go (persisted enums: store/DB contract). - Delete runner/lifecycle.go: DaemonState -> runner.go, classifyOutcome -> sync.go, IndexOutcome + unused crawlResult.outcome removed. - storeUnhealthy is component-aware (stage=failed, fault=store). - Fix: an attempts-exhausted transient fault still retries (item is rescheduled), so it logs WARN 'still retrying past the limit', NOT ERROR 'parked'. Only a permanent fault parks. Verified: gofmt, build (incl. cmd + plugin), vet, go test ./pkg/crawler/... green.
Re-review of the component/stage/message refactor (behavior verified faithful; these are cleanups + doc↔code sync): - Remove reserved-unused scope-drop vocabulary: DropReason (type/consts/String) and OutcomeDropped (+ its dead tally case) — re-add in Phase 2 with the code that produces them. - Delete dead DaemonStartFailed const (unused; value contradicted emitted 'failed'). - stepPhrase: add content_invalid -> 'build the push request', ssrf -> 'download the files' (these faults previously mis-reported the step). - Spell out the tally in crawl.finished / sync.finished messages. - Drop the sync.started event (asymmetric with crawl; keeps INFO quiet) — a tick is summarized by 'finished' only. - Replace the dead syncStage.what field with a []stageFn pipeline whose labels are inline comments (recipe stays readable, no write-only field). - Fix stale comments (lifecycle.go->status.go, sync_pass/index_pass wording; soften the telemetry header re: field order / run_id). - docs/crawler-logs.md: banner -> implemented; document crawl.failed, polled not_modified, daemon at=start; drop sync.started; add ssrf/content_invalid to the fault->step table. Verified: gofmt, build (crawler+cmd), vet, go test ./pkg/crawler/... green.
Rework the runner.Metrics port from the plumbing-only 5 signals to the set ops
and business actually need. New/changed signals:
- crawler_sync_outcome_total{outcome,fault} — one labeled counter (replaces the
separate CatalogPushed/CatalogFailed): real success rate + the fault label that
routes an error spike to publisher vs Discovery vs infra.
- crawler_seconds_since_last_success{job} — liveness gauge (OTel observable gauge
driven by MarkPassSuccess per tick) that disambiguates 'queue empty because
caught up' from 'queue empty because the crawler is wedged'.
- crawler_catalogs_parked / crawler_catalogs_tracked — standing gauges of stuck
inventory + coverage (new Store.CountParked/CountTracked).
- crawler_sync_lag_seconds — queue-residence freshness (synced_at - enqueued_at;
ClaimedItem now carries EnqueuedAt).
- crawler_index_poll_total{result} — per-source reachability.
- kept queue_depth + push/index-latency histograms.
Wiring: RecordSyncOutcome at the five sync terminals (fault only on failures),
MarkPassSuccess once per crawl/sync tick, parked/tracked refreshed each sync tick.
Port stays behind NopMetrics by default (framework-agnostic); the OTel adapter in
crawler/telemetry.go implements it (observable gauge is mutex-guarded, no race).
Verified: gofmt, build (crawler+cmd), vet, go test ./pkg/crawler/... (incl.
DB-backed store/engine) all green.
CrawlNow ran a full crawl but logged nothing on completion (only DEBUG-level per-index traces, off by default) and never returned a run_id, so a successful on-demand crawl was silent and uncorrelatable. CrawlNow now mints its run_id synchronously, returns it to the HTTP caller (surfaced as runId in the /crawl response), and emits the same INFO "crawl finished" summary a scheduled tick uses. Also add a trigger field (scheduled/on_demand) to every crawl-component log line, since a scheduled tick and an on-demand crawl otherwise look identical without cross-referencing run_id against the HTTP response that minted it.
…G_LEVEL Both drivers hardcoded slog.NewJSONHandler(..., nil), which defaults to INFO — so per-item DEBUG lines (logPolled, logQueued, etc.) were dropped in every deployment regardless of the onix core logger's own log.level (a separate sink). CRAWLER_LOG_LEVEL now controls this handler directly.
… work
catalogPass broke out of its claim loop on a Store.ClaimNext error and fell
through to the shared tail, which calls Metrics.MarkPassSuccess("sync")
unconditionally. A sustained Postgres outage therefore reset
crawler_seconds_since_last_success every ~30s (one tick), so the "crawler
wedged" alert — the single reason that gauge exists, per docs/crawler-metrics.md
— could never fire on the condition it was built to catch.
Return instead of break, matching indexPass, which already withholds
MarkPassSuccess("crawl") when Source.IndexRefs fails. The queue-drained break is
untouched: an idle tick is a successful tick, and must keep marking success —
that idle-vs-wedged distinction is the whole point of the gauge.
recMetrics.MarkPassSuccess was a bare no-op that no test asserted on; it now
counts per job. The new test drives a real Postgres store with a closed pool, so
the failure arrives through the actual ClaimNext path rather than a mock.
… cause
Two coupled defects in the fault taxonomy (B8 + B2 of the Phase-1 review).
B8 — the fetch layer raised its integrity and guard failures as plain
fmt.Errorf, so IsPermanent was false and routeFailure sent them to
scheduleRetry, which never parks. A digest mismatch — Phase 1's ONLY integrity
check, and the spec's "treat as tampering and flag it" signal — therefore
re-downloaded the same bad bytes every =<5 min forever, logged WARN and counted
as fault=transient, instead of parking with an ERROR an operator would see.
FaultDigestMismatch/FaultSSRF/FaultOversize were already declared permanent;
nothing produced them. Now raised via PermanentFaultf: digest mismatch, the
download cap, and both SSRF rejections.
The non-200 branch deliberately stays transient — a publisher's 503 must retry,
not park the catalog until they publish a new version. Covered by a test.
B2 — ClassifyFault collapsed every PermanentError to FaultDecode, so a
continuity gap, a tampered digest and an SSRF rejection all surfaced as
fault_class="decode" and pointed operators at the wrong stage. PermanentError
now carries an optional FaultClass that the raiser (which knows) sets, and
ClassifyFault reports it, falling back to FaultDecode when unset. The class
survives fmt.Errorf(%w) and *url.Error wrapping, which is how dial-time SSRF
rejections reach the classifier — both pinned by tests.
Knock-on fix: stepPhrase("oversize") returned "batch the catalog", written for a
batching fault no code path raises (Discovery rejecting a batch is a 413 =>
push_rejected). With the class now produced by the download cap, that phrase
would have printed the wrong stage, so it becomes "download the files". The
decompression-bomb cap stays unclassified (=> decode, "unpack the files"), which
is the truthful stage for it; both park, only the phrase differs.
docs/crawler-logs.md's fault table is corrected to match and now states which
classes park vs retry, and why the two size caps report different stages.
… the schedule The scheduled index pass enumerated only Source.IndexRefs() (config + registry), so an index crawled once via the on-demand /crawl endpoint was never revisited — a later change file needed another manual /crawl. Now indexPass unions those refs with every index already recorded in crawler_index (new Store.KnownIndexes), deduped by URL (configured refs win) and each still gated by its own next_crawl_at, so a once-crawled index keeps being re-polled on the normal cadence and picks up new change files on its own. - store.KnownIndexes: SELECT index_url/participant_id/source FROM crawler_index. - runner.withPersistedIndexes: merge + dedup; a store error is non-fatal (logged as store unhealthy, the pass proceeds with the configured refs). - On-demand still force-runs immediately (bypasses the next_crawl_at gate); it just also gets scheduled afterward. Deleting an index's crawler_index row (the reference UI's Reset) drops it from the schedule again. Test (DB-backed): TestIndexPass_ReCrawlsPersistedIndexes — a persisted on-demand index is re-crawled by the scheduled pass, and a URL that is both configured and persisted is crawled once (deduped). Verified: gofmt, build (crawler + cmd), vet, go test ./pkg/crawler/... green.
…es (no baseline fallback) An incremental MERGE builds the delta straight from the change files' catalog metadata envelope (id/descriptor/provider). When that envelope was absent, buildPushDoc fell back to ResolveWithChangeset — re-downloading the baseline and folding the whole catalog — which silently masked a malformed change file behind a full re-resolve. Now a missing envelope on an incremental update (cursor >= baseline) is a permanent content fault: buildPushDoc returns PermanentFaultf(FaultContentInvalid, ...), so the item parks (ERROR) with the baseline NOT fetched, nothing pushed, and the cursor unchanged — it re-activates when the publisher republishes a compliant change file (version bump). Change files must carry the required catalog metadata. First sync is unchanged: cursor < baseline (a new catalog, or a publisher re-baseline/compaction) still fetches the baseline via ResolveWithChangeset — that is the initial resolve, not a fallback. Test: TestEngine_MergeOnly_NoEnvelope_ParksNoBaselineFallback — an envelope-less change file parks the catalog with no baseline fetch, no push, cursor held. Verified: gofmt, build (crawler + cmd), vet, go test ./pkg/crawler/... green.
…ile-only; package/symbols/route unchanged)
Collaborator
|
Thanks for the writeup and the work here — the crawl engine's core mechanics (queue claim/lease, SSRF dial-time guard, decompression cap) are solid. I don't think this is mergeable in its current shape, though — there are two equally-weighted sets of blockers: structure/placement, and correctness/security. Both need to be addressed in the same pass, not deferred. Structure / placement1. 2. Hard-coded Postgres coupling. 3. 4. Correctness / security5. SSRF pre-check has no timeout. 6. Retry cap doesn't actually park. 7. Digest verification fails open. 8. No per-index lock between on-demand and scheduled crawls. 9. Signature verification silently dropped, not just an unused arg. 10. Please restructure per 1–4 and fix 5–10 in the same pass — happy to re-review once it's reshaped. |
ameersohel45
marked this pull request as draft
July 29, 2026 17:17
…d in review
Closes all ten findings from the PR 894 review, and five further defects found
during adversarial review that were not on that list.
Structure and placement
- Delete cmd/crawler. It was never wired into build-plugins.sh, any Dockerfile,
any compose file or CI; the plugin path was always the only wired path.
- Move pkg/crawler under pkg/plugin/implementation/crawler/internal/crawler, so
nothing outside the plugin can import it.
- Remove docs/. Three of its five files were internal working notes, including a
564 line change request marked "Status: Proposed". The logging and metrics
reference content is now part of the plugin README.
- Move the six persisted-state types (PassReport, CatalogState, IndexState,
KnownIndex, QueueItem, ClaimedItem) into package catalog, then declare
runner.Store over them. All 16 signatures name only catalog types and the
standard library, so a backend can satisfy the port without importing
Postgres. Backend selection goes through a registry and factory with pgx
confined to one file, and CRAWLER_ENABLED (default false) means an operator
not running the crawler needs no DSN and opens no database.
Signature verification
- Restore per-file signed tuple verification, which the branch had dropped to
digest only. Mandatory, fail closed, no off switch. Uses the existing
artifactverifier.VerifyFileTuple rather than new cryptography.
- Keys come from the network registry by {subscriberId, keyId}, the same path
signvalidator uses for transport signatures. There is no key configuration.
The crawl handler now requires a registry plugin and refuses to construct
without one, because a crawler that cannot verify parks everything silently.
- Registry errors are transient and retry; a definitive unknown key is permanent
and parks. Lookups cache on (subscriberId, keyId), never keyId alone. Add a
dedicated signature fault class so an operator key problem no longer reports
as digest_mismatch and blames the publisher.
Data integrity, found in review
- A partial push no longer parks. Batches are sent in order and stop at the
first un-acked batch, so a later 4xx after an earlier ack used to park the
item without advancing the cursor, leaving Discovery permanently half applied.
Two further park sites inside the batch loop bypassed the first fix and
reported BatchesAcked as zero, hiding it.
- The index version cursor is withheld when a catalog fails to enqueue. It used
to advance regardless, so a store blip mid pass dropped an update until the
publisher bumped the index, made worse by conditional GET answering 304.
- A corrupt but correctly signed baseline now parks instead of settling as a
clean skip and advancing the cursor. A genuinely empty catalog still skips.
- ResolveDelta enforces change-file continuity, so a missing intermediate
version parks rather than silently composing a divergent catalog.
- A crawl pass that could not do its work no longer marks itself successful, so
a wedged crawler stops reporting itself healthy.
Other
- Bound the SSRF DNS pre-check with the configured fetch timeout; the resolver
was context aware but no production caller supplied a deadline.
- Refuse 0.0.0.0/8, 240.0.0.0/4 and global multicast in the public-IP check.
- Park on exceeding MaxAttempts instead of retrying forever.
- Serialise crawlIndex per index URL so an on-demand and a scheduled crawl
cannot race the version cursor.
- QueueDepth counts only pending work, so one parked catalog no longer
permanently false-fires the backlog alert.
- CRAWLER_MERGE_ONLY parses with strconv.ParseBool; "0" and "no" used to mean
true. Reject CRAWLER_REGISTRY_URL, which was accepted while being ignored.
- Cap the /crawl request body. The endpoint still accepts any index URL, which
is the recorded scope for this release; trust rides on the signature check.
Verification
- go build and go vet clean, 61 test packages pass, no new gofmt findings.
- The Postgres-gated store and engine tests were run against a real database
rather than skipped.
- A new end-to-end test runs the whole chain against real Postgres with a
publisher origin, a registry and Discovery. The control case, a catalog signed
by a key the registry does not vouch for, parks with a signature fault and
nothing reaches Discovery.
Note: digest verification was reported as failing open. It already failed
closed, and integrity.go is unchanged by this work.
Collaborator
|
Thanks @ameersohel45 |
Wire a registry-backed index-discovery source and unify the on-demand
/crawl trigger onto it, so the scheduled pass and /crawl take one
registry-based input model instead of two.
Source: DediQueryClient calls GET {base}/query/{networkId} and keeps the
live provider records that publish a catalog_index_url; registrySource
dedups by index URL across networks. Config now accepts
CRAWLER_REGISTRY_URL, and selectSource picks the registry source only
when a network is also configured (matching LoadSettings' own
source-required check) - otherwise it falls back to the static
CRAWLER_INDEX_URLS list, closing a gap where a bare registry URL would
have started a crawler that discovered nothing.
/crawl: the body is now {registryUrl, networkId|networkIds}; the handler
runs /query discovery and crawls each discovered index on demand.
CrawlNow(indexURL) becomes CrawlRegistry(registryURL, networkIDs) across
the definition interface, the plugin adapter, the engine, and the
disabled stub, built through an injected registry-source factory so the
engine stays agnostic of how a query client is constructed.
Tests: TDD red->green for the /query source, the config/selection
predicate, and CrawlRegistry; the e2e now proves /query -> discover ->
verify -> push end-to-end on loopback (plus the wrong-key control still
parks). The signature gate is unchanged and still enforced.
Contributor
Author
|
| Field | Required | Notes |
|---|---|---|
registryUrl |
✅ | Absolute http/https. DeDi registry base; the crawler calls GET {registryUrl}/query/{networkId} to discover the provider catalog indexes, then fetches + verifies + pushes each. |
networkId |
✅* | Single network id, e.g. beckn.one/testnet. |
networkIds |
— | List form, e.g. ["beckn.one/testnet","beckn.one/mainnet"]. Merged with networkId, de-duplicated, order preserved. |
* at least one of networkId / networkIds must be present.
Response — 202 Accepted
{
"status": "ACCEPTED",
"registryUrl": "https://fabric.nfh.global/registry/dedi",
"networkIds": ["beckn.one/testnet"],
"runId": "…"
}Validation: a missing/blank registryUrl, a non-http(s) scheme, or no network → 400 Bad Request. The crawl runs asynchronously; progress surfaces in the crawler's telemetry, correlated by runId.
Scheduled pass (unchanged input model)
The background pass reads the same source from config — CRAWLER_REGISTRY_URL + CRAWLER_NETWORK_IDS — and discovers via /query every CRAWLER_INDEX_INTERVAL. Setting a registry base with no networks falls back to the static CRAWLER_INDEX_URLS list.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Phase-1 decentralized catalog crawler — a framework-agnostic module (
pkg/crawler) plus a thin onix plugin and a DS-internal/crawlhandler — built on top of the initialcatalog-crawlerscaffold.This branch builds on
catalog-crawler's last 4 commits and reworks/extends them into the full Phase-1 implementation:167eac3refactor(catalogcrawler): rewrite to match the file spec81e8981fix(dedi): real detached-JWS sign/verify per §7c38080afix(catalogcrawler): match manifest catalog-index entry on the registry36f8f2bcatalog-crawler: add crawler plugin and DS-facing catalog/pull handlerBehaviour change vs
catalog-crawler(the base)The base branch implemented a synchronous, pull-based crawler:
Crawler.CrawlSubscriber(CrawlRequest) → CrawlResult— one call fetched + verified a subscriber's manifest/index and composed the full catalog (CatalogResult.Catalog= the baseline with every change file applied).catalog/pullhandler invoked it synchronously and returned the catalogs inline in the HTTP response — aCatalogPullCallbackAction-shaped body{status: COMPLETED|FAILED, catalogs: […], error}("no ACK/callback split; a single synchronous response suffices").CrawlModeIncrementalskipped a part whose digest was unchanged); scope was one subscriber per request.This branch reworks that into a background, push-based crawler:
Start/Stop+ an on-demandCrawlNow(indexURL) → runId. Scheduled index/catalog jobs run continuously from plugin init.POST /crawl {indexUrl}returns202 ACCEPTED+ arunId— it does not return the catalog inline. The crawler pushes each resolved catalog to Discovery's/catalog/push(→ Kafka → publish job → Elasticsearch); outcomes are observed via logs/metrics and thecrawler_*tables, not the trigger's response.(The
/crawlhandler file is namedcatalogPullHandler.goafter a file-only rename; its struct and route are stillcrawl.)What's in it
Crawl model
crawler_index/crawler_queue/crawler_catalog) withFOR UPDATE SKIP LOCKED, a claim token + lease; the cursor advances only on a successful push.CRAWLER_MERGE_ONLY): the delta is built straight from the change file's catalog metadata envelope — no baseline fetch. First sync / re-baseline still fetches the baseline. A change file that omits the envelope is a permanent content fault (no silent baseline fallback)./catalog/pushwithpublishDirectives(catalogType,updateMode,visibleTo).Fetch / decode
json+json.gzipvia an extensible codec registry.Reliability / lifecycle
Observability
run_id/pass_id;CRAWLER_LOG_LEVEL.crawler_sync_outcome_total{outcome,fault}, aseconds_since_last_successliveness gauge,catalogs_parked/catalogs_tracked,sync_lag,index_poll,queue_depth, and push/index latency histograms.Scheduler
/crawl'd index joins the recurring schedule: the scheduled pass re-polls every persisted index (deduped, each gated by its ownnext_crawl_at), so a later change file is picked up with no second/crawl.Structure
pkg/crawler/{catalog,fetch,decode,publish,source,store,runner,config}+ a composition root; fullcatalogcrawler→crawlerrename across module / cmd / plugin./crawlhandler file renamedcrawlHandler.go→catalogPullHandler.go(file-only; symbols/route unchanged).Testing
go test ./pkg/crawler/...green (incl. DB-gated store/engine tests viaCRAWLER_TEST_DB_DSN), plusgofmt/go build/go vet.Deferred to Phase 2