Turns a claimed import run into a verified release.

execute/3 receives a run id GeoGenius.import/1 already obtained from the atomic candidate-registration statement, reads what it needs out of PostgreSQL, and walks the run through its phases: download every artifact, check each required one arrived, stage them through the manifest's provider, normalize the staged rows into the catalog, rebuild relations and write every edge the provider asserts, analyze, and verify. It atomically claims the run's sole executor before doing that work. It never registers the attempt or opens the release; both belong to the caller that prepared the manifest.

The phases themselves live next door where they have room: GeoGenius.Pipeline.Artifacts owns everything that touches the cache, the downloader, and checksums; GeoGenius.Pipeline.Normalize owns everything that knows GeoGenius.Provider.Area; GeoGenius.Pipeline.Relate owns rebuilding measured relations and writing asserted edges. This module owns the driver, the failure path, staging, and the two remaining one-line phases, indexing and verifying.

What an import does not do

It does not publish. An import produces a verified candidate; making it visible is a separate, explicit act, which is why mix geo_genius.publish exists as its own task. publish: true is the convenience for a caller that wants both in one call.

It does not open a transaction. Every catalog function is atomic as a single statement, and durable phase progress must survive each commit. Wrapping the run would roll all of that evidence back and hold one connection for the length of the slowest phase. Recovery creates an explicit new attempt rather than resuming execution in place.

Staging is emptied before it is written and dropped after

Every attempt stages from its own immutable artifact snapshot. The staging phase starts from an empty table for that run -- GeoGenius.Staging.reset/3 -- so a leaked or manually pre-created table can never contribute rows the owning executor did not stage.

The staging table is dropped in an after, on success and on failure alike. No retry reuses a run_id; a corrected retry is a new attempt with a new table name. A kept table is therefore a leak that the schema down migration's teardown check would later refuse to drop through. What makes a retry cheap is the artifact cache, which skips the network -- the expensive part. Re-parsing a local file is not worth leaking a table per failure. A drop that fails is logged at warning and the table left behind, because the alternative -- letting cleanup raise -- would replace the run's own outcome with it.

The failure path survives its own failures

A phase that returns {:error, reason}, raises, exits, or throws all reach fail_import/4 the same way, and the stored error column is the durable record of what happened. Everything after that point is written so it cannot destroy that record: cleanup cannot replace the return value, and the re-read that turns a recorded failure into a %GeoGenius.ImportRun{} cannot either. A caller that gets {:error, {:unrecorded, certainty, reason}} is being told the run's own row could not be read or written. :not_started means execution never claimed the run; :outcome_unknown means work started but its durable outcome could not be confirmed. That, and only that, is when the reason arrives as a string instead of a run.

A notifier cannot fail an import

Every notify/3 call is wrapped, so a notifier that raises, exits, or returns garbage is logged at warning level and dropped, and the phase continues. A host's event delivery is a side channel; letting it abort a two-hour import would be a worse contract than losing an event.

:phase_advanced for a phase carries the metrics of the phase before it: a phase's own metrics are not known until it finishes, and they are written onto the run by the next advance_import. The metrics of the last phase arrive with :import_completed.

What a provider is handed

stage/5 receives opts[:work_dir], one directory resolved per run and removed when the run ends, and opts[:command], the configured command adapter wrapped by GeoGenius.Pipeline.CommandAllowlist so a provider that reaches for something other than ogr2ogr is refused.

Options

• :publish -- publish the release after it verifies. Defaults to false.
• :batch_size -- staged rows read and normalized or asserted per round trip, and the relate phase's heartbeat cadence. Defaults to 500.
• :work_dir -- the directory the run's own working directory is created beneath. Defaults to the system temporary directory.
• :stale_after_seconds -- the staleness window this run was claimed under (the registration claim's :stale_after_seconds, carried here through GeoGenius.import/1 and a runner's args). Defaults to 900, matching the public import default.
• :timeout -- milliseconds allowed for each ingestion batch or long statement: staging and normalization batches, asserted relation batches, the measured relation rebuild, analyze, verification, and publication, which re-runs that same verification inside itself. Defaults to :stale_after_seconds converted to milliseconds, so an explicit :timeout here is only needed to diverge from the window the run was actually claimed under.

Every other option is passed through to the cache, downloader, command, and notifier adapters untouched.

What :timeout bounds

DBConnection's default is fifteen seconds, which a national relation rebuild or the verification a publication re-runs exceeds by orders of magnitude; without an option here a host's only recourse is raising :timeout on its own Repo, for every query it runs. :timeout now derives from :stale_after_seconds, the same window candidate registration claimed the run under, rather than a value that merely resembled it: GeoGenius.import/1 knows the window it claimed with and carries it through a runner's args to execute/3, so a caller that claims with a shorter window gets a shorter statement timeout to match, instead of a timeout longer than its own lease.

The statement is narrow: a statement that runs past :timeout fails and is recorded as a failure rather than running indefinitely. It is not a guarantee that the statement finishes inside the lease -- at the defaults, advance_import renews the lease immediately before the statement and the statement times out at roughly the moment the lease becomes stale. A stale lease is diagnostic only and never authorizes another executor.

Ingestion statements also set checkout_retries: 0. Once a mutating statement disconnects, the caller cannot know whether PostgreSQL committed any of its work, so DBConnection must surface that one failure instead of checking out another connection and repeating the statement. A runner retry or a newly claimed import attempt owns recovery at the pipeline boundary.