app.services.polling

Poll orchestration: list, skip unchanged, fetch, ingest, reconcile, log.

PollingService polls registrations one after another in configuration order (ownership depends on that order). Each registration runs in three phases:

1. Run log – an intake_run row is started in its own unit of work, so it survives a rollback of the ingest transaction.

2. Collect (no transaction held, no lock) – open the adapter (open_adapter), list projects, list each project’s documents with the stored change tokens (bounded concurrency), drop documents owned by an earlier registration (ownership_conflict warning), fetch only changed documents (bounded concurrency) and prepare them with the shared ingest path (app.services.ingest.prepare_document()).

3. Apply – under lock_registration() (one transaction): ingest the fetched documents, rewrite change tokens of unchanged documents whose token moved, and delete published rows that a complete listing no longer contains.

Finally, in a separate unit of work, the run is finished and the registration’s last_attempt_at / last_success_at / last_status / last_error are set.

Reconciliation rules: nothing is deleted unless the project listing was complete; within a project, nothing is deleted unless its document listing was complete. A vanished project (complete project listing) loses its published rows. Candidates are never deleted by polling.

Ownership: the first enabled registration in configuration order whose stored published rows contain a (project, path) owns it; later registrations skip that file (and drop their own stale copy).

Statuses: failed when no project listing could be produced, the adapter could not be opened or the write failed (prior content is kept); partial when any listing was incomplete or any fetch failed; otherwise ok. A failure in one registration never affects others.

Public API: DEFAULT_CONCURRENCY, PollingService, and re-exported from app.services.poll_plan: PollProblem, PollCounts, PollOutcome.

app.services.polling.DEFAULT_CONCURRENCY = 8

Default bound for concurrent listings and fetches per registration.

class app.services.polling.PollCounts(projects=0, documents=0, unchanged=0, fetched=0, created=0, updated=0, same_content=0, tokens_refreshed=0, deleted=0, skipped_owned=0, failed=0)

Bases: object

Counters of one poll run.

Parameters:

• projects (int) – Projects listed.

• documents (int) – Documents listed and owned by the registration.

• unchanged (int) – Documents skipped without fetching.

• fetched (int) – Documents fetched.

• created (int) – New published rows.

• updated (int) – Rows whose digest changed.

• same_content (int) – Fetched rows whose digest did not change.

• tokens_refreshed (int) – Unchanged rows whose change token moved.

• deleted (int) – Published rows removed by reconciliation.

• skipped_owned (int) – Files skipped (owned by an earlier registration).

• failed (int) – Listings or fetches that failed.

projects: int

documents: int

unchanged: int

fetched: int

created: int

updated: int

same_content: int

tokens_refreshed: int

deleted: int

skipped_owned: int

failed: int

to_json()

Convert to the JSON stored in intake_run.counts.

Return type:

dict[str, JsonValue]

Returns:

The counters.

class app.services.polling.PollOutcome(*, source_id, status, run_id, counts, problems)

Bases: object

Result of polling one registration.

Parameters:

• source_id (str) – Registration id.

• status (PollStatus) – Outcome (disabled when not pollable).

• run_id (UUID | None) – Intake run id (None when nothing ran).

• counts (PollCounts) – Counters.

• problems (tuple[PollProblem, ...]) – Findings.

source_id: str

status: PollStatus

run_id: UUID | None

counts: PollCounts

problems: tuple[PollProblem, ...]

class app.services.polling.PollProblem(*, code, message, project=None, path=None)

Bases: object

One finding of a poll run.

Parameters:

• code (str) – Stable code: source_error, projects_incomplete, listing_failed, listing_incomplete, fetch_failed, ownership_conflict, not_pollable, internal_error.

• message (str) – Actionable, secret-free description.

• project (str | None) – Affected project, if any.

• path (str | None) – Affected file path, if any.

code: str

message: str

project: str | None

path: str | None

to_json()

Convert to the JSON stored in intake_run.errors.

Return type:

dict[str, JsonValue]

Returns:

The mapping (None fields omitted).

class app.services.polling.PollingService(provider, sources, *, concurrency=8, clock=<function _utcnow>)

Bases: object

Polls source registrations into the catalog.

Parameters:

• provider (CatalogStoreProvider) – Storage provider.

• sources (Sequence[SourceRegistration]) – All configured registrations, in configuration order.

• concurrency (int) – Bound for concurrent listings and fetches.

• clock (Callable[[], datetime]) – Time source for last_attempt_at.

async poll_all(*, trigger=None)

Poll every pollable registration, in configuration order.

Parameters:

trigger (JsonValue) – Who triggered the poll (stored, no secrets).

Return type:

list[PollOutcome]

Returns:

One outcome per pollable registration.

async poll_source(source_id, *, trigger=None)

Poll one registration.

Parameters:

• source_id (str) – Registration id.

• trigger (JsonValue) – Who triggered the poll (stored, no secrets).

Return type:

PollOutcome

Returns:

The outcome; disabled (nothing ran) when the registration is invalid or has no poll intake.

Raises:

NotFoundError – If no such registration is configured.