Jobs and progress

Every long-running operation is a Job that owns a DAG of Tasks. Both live in the DB. Each Task is a single ARQ job at the executor layer; ARQ doesn’t know about the DAG, only the orchestrator does.

Job lifecycle

        stateDiagram-v2
    [*] --> pending
    pending --> running : worker leases task
    running --> succeeded
    running --> failed
    running --> cancelled : cancel request
    running --> cancelled_dirty : force cancel request
    failed --> pending : resume request
    cancelled --> pending : resume request
    succeeded --> [*]

A job stays in pending until at least one Task has been admitted by a worker. Cancellation is cooperative between phases; only ?force=true SIGKILLs the subprocess (and triggers a worker restart to flush the CUDA context).

Cache lookup

When the orchestrator persists a Task, it computes cache_key = sha256(canonical_json({kind, inputs_hash, params_hash, rv_id})) and queries task for an existing row with the same cache_key and status='succeeded'. If found, the new Task is immediately marked succeeded and inherits the cached outputs_ref_json — no enqueue.

This is what makes submit_features followed by submit_matches followed by re-submitting the same submit_features not re-extract: the second call hits cache.

Lease + heartbeat

async app.orchestrator.lease.try_acquire_lease(session, *, table, pk_col, lease_col, worker_col, pk_value, worker_id, ttl_seconds)[source]
Parameters:

  • session (AsyncSession)

  • pk_value (str)

  • worker_id (str)

  • ttl_seconds (int)

Return type:

bool

async app.orchestrator.lease.refresh_lease(session, *, table, pk_col, lease_col, worker_col, pk_value, worker_id, ttl_seconds)[source]
Parameters:

  • session (AsyncSession)

  • pk_value (str)

  • worker_id (str)

  • ttl_seconds (int)

Return type:

bool

The pattern works on both SQLite and Postgres without dialect branches. Workers refresh every lease_ttl_seconds // 3. If a worker crashes, the next worker that scans pending tasks reclaims the lease once lease_expires_at < now().

Fair-share scheduler

Fair-share scheduler.

Picks the next ready Task across all tenants, biased toward tenants with fewer recent admissions. The interleaving rule:

Pick the tenant with the smallest count of currently-running tasks; break ties by the tenant whose last-admitted task is oldest. Within a tenant, pick by Task.created_at (FIFO).

A simpler invariant we expose for testing:

max_consecutive_jobs_per_tenant: never admit more than N consecutive Tasks from the same tenant when other tenants have ready work.

This is intentionally a picker — it does not enqueue. The supervisor calls pick_next_task(…) between subprocess slots; in tests we use the picker directly to validate fairness.

class app.orchestrator.fair_share.FairShareState(max_consecutive_per_tenant: 'int' = 2, recent_tenants: 'deque[str]' = <factory>)[source]

Bases: object

Parameters:

  • max_consecutive_per_tenant (int)

  • recent_tenants (deque[str])

max_consecutive_per_tenant: int = 2
recent_tenants: deque[str]
admit(tenant_id)[source]
Parameters:

tenant_id (str)

Return type:

None

consecutive_for(tenant_id)[source]
Parameters:

tenant_id (str)

Return type:

int

async app.orchestrator.fair_share.pick_next_task(session, *, state)[source]

Find the highest-priority ready Task that doesn’t violate max_consecutive_per_tenant. Ready = pending + deps satisfied.

For Phase 5 v1 we treat ‘ready’ as pending with no remaining deps; dep satisfaction is a precondition the orchestrator already enforces when emitting tasks. Tenants are weighted by current running count.

Parameters:

  • session (AsyncSession)

  • state (FairShareState)

Return type:

Task | None

Picks the next ready Task biased toward tenants with the smallest running-task count, capped by max_consecutive_per_tenant to prevent starvation when one tenant submits 100 jobs at once.

ProgressEvent stream

Workers emit a versioned ProgressEvent for each phase boundary, metric, snapshot seal, or warning/error. Events are written to two sinks:

  1. events.jsonl on disk — durable, used for SSE replay via Last-Event-ID.

  2. JobEvent table — durable index for the API to count/serve.

The SSE endpoint streams from the DB (cursor on event_id) and forwards new events live until the client disconnects.

Choosing a progress endpoint

Use the three job reads for different purposes:

Endpoint

Use when

GET /v1/jobs/{id}

You need canonical lifecycle state, task outputs, or final errors

GET /v1/jobs/{id}/progress

You need a small polling response for a CLI, web UI, or dashboard

GET /v1/jobs/{id}/events

You need every phase, metric, warning, log line, or snapshot event

For polling clients, GET /v1/jobs/{id}/progress returns the current job status, task counts, active task, latest event, and a best-effort progress fraction. Use it for CLIs and dashboards that need a snapshot rather than a live stream.

The progress fraction is computed from task state plus the latest phase_progress event for each task. Terminal tasks count as 1.0. Running tasks use current / total when the backend reports both values. Pending tasks and tasks with no measurable total count as 0.0. This makes the field stable for UI display without making it a scheduler guarantee.

Example snapshot:

{
  "job_id": "01J...",
  "recipe": "global",
  "status": "running",
  "progress": 0.42,
  "total_tasks": 4,
  "completed_tasks": 2,
  "task_counts": {"succeeded": 2, "running": 1, "pending": 1},
  "current_task_kind": "match",
  "current_phase": "matching",
  "latest_event_id": 182,
  "tasks": [
    {"kind": "extract", "status": "succeeded", "progress": 1.0},
    {"kind": "match", "status": "running", "progress": 0.68}
  ]
}

Backend-reported progress

Backends may opt in to fine-grained percentages by accepting an optional progress keyword on long-running methods. Workers pass this reporter only when the method signature supports it, then persist the reported phase_progress events. Report counts (current / total) where possible; the API derives the 0.0 to 1.0 fraction from the latest durable event.

Good totals are domain-specific:

Stage

Useful total

Feature extraction

Number of input images

Exhaustive matching

Number of image pairs, n * (n - 1) / 2

Sequential matching

Number of selected neighboring pairs

Geometric verification

Number of candidate match pairs

Mapping

Registered images or backend-specific milestones when available

Backend progress must remain best-effort. A reporter failure should be logged or ignored by the backend, not fail the reconstruction job.

ProgressEvent v1 — discriminated union for the SSE stream.

Workers emit events to events.jsonl (one JSON line per event); the API serves them via SSE with Last-Event-ID resume.

Schema is versioned: a schema_version literal is pinned per kind. New kinds extend the union; old kinds never change shape.

ProgressEvent kinds

ProgressEvent is a tagged union discriminated on kind:

  • phase_started / phase_progress / phase_completed — per-phase lifecycle (phase is one of Phase).

  • metric — scalar telemetry sample (key + value).

  • snapshot_available — a sealed snapshot is now readable (snapshot_seq + summary).

  • log_line / warning / error — opaque message channels.

Forward-compatibility: SDKs MUST treat unknown kind values as unknown and surface them through a generic catch-all rather than crashing the stream. New kinds are additive (new kind literal plus new arm of the union); existing kinds never change shape. schema_version is pinned to 1 across every variant — bump it in lock-step if the wire shape ever has to break, never per-kind.

class app.schemas.progress_event.ErrorEvent(*, schema_version=1, ts, job_id, task_id=None, seq, kind='error', error_class, message, detail=None)[source]

Bases: _Base

Parameters:

  • schema_version (Literal[1])

  • ts (datetime)

  • job_id (str)

  • task_id (str | None)

  • seq (int)

  • kind (Literal['error'])

  • error_class (str)

  • message (str)

  • detail (dict[str, Any] | None)

kind: Literal['error']
error_class: str
message: str
detail: dict[str, Any] | None
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.progress_event.LogLine(*, schema_version=1, ts, job_id, task_id=None, seq, kind='log_line', level, message)[source]

Bases: _Base

Parameters:

  • schema_version (Literal[1])

  • ts (datetime)

  • job_id (str)

  • task_id (str | None)

  • seq (int)

  • kind (Literal['log_line'])

  • level (Literal['DEBUG', 'INFO', 'WARNING', 'ERROR'])

  • message (str)

kind: Literal['log_line']
level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR']
message: str
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.progress_event.Metric(*, schema_version=1, ts, job_id, task_id=None, seq, kind='metric', key, value)[source]

Bases: _Base

Parameters:

  • schema_version (Literal[1])

  • ts (datetime)

  • job_id (str)

  • task_id (str | None)

  • seq (int)

  • kind (Literal['metric'])

  • key (str)

  • value (float)

kind: Literal['metric']
key: str
value: float
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.progress_event.PhaseCompleted(*, schema_version=1, ts, job_id, task_id=None, seq, kind='phase_completed', phase)[source]

Bases: _Base

Parameters:

  • schema_version (Literal[1])

  • ts (datetime)

  • job_id (str)

  • task_id (str | None)

  • seq (int)

  • kind (Literal['phase_completed'])

  • phase (Literal['feature_extraction', 'matching', 'geometric_verification', 'incremental_init', 'incremental_register', 'incremental_ba', 'global_rotation_avg', 'global_positioning', 'global_ba', 'hierarchical_cluster', 'hierarchical_merge', 'panorama', 'spherical', 'bundle_adjust', 'triangulate', 'relocalize', 'pose_graph_optimize', 'segment', 'export', 'vlad_index', 'backend_action', 'artifact_conversion'])

kind: Literal['phase_completed']
phase: Phase
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.progress_event.PhaseProgress(*, schema_version=1, ts, job_id, task_id=None, seq, kind='phase_progress', phase, current, total=None, rate=None)[source]

Bases: _Base

Parameters:

  • schema_version (Literal[1])

  • ts (datetime)

  • job_id (str)

  • task_id (str | None)

  • seq (int)

  • kind (Literal['phase_progress'])

  • phase (Literal['feature_extraction', 'matching', 'geometric_verification', 'incremental_init', 'incremental_register', 'incremental_ba', 'global_rotation_avg', 'global_positioning', 'global_ba', 'hierarchical_cluster', 'hierarchical_merge', 'panorama', 'spherical', 'bundle_adjust', 'triangulate', 'relocalize', 'pose_graph_optimize', 'segment', 'export', 'vlad_index', 'backend_action', 'artifact_conversion'])

  • current (int)

  • total (int | None)

  • rate (float | None)

kind: Literal['phase_progress']
phase: Phase
current: int
total: int | None
rate: float | None
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.progress_event.PhaseStarted(*, schema_version=1, ts, job_id, task_id=None, seq, kind='phase_started', phase)[source]

Bases: _Base

Parameters:

  • schema_version (Literal[1])

  • ts (datetime)

  • job_id (str)

  • task_id (str | None)

  • seq (int)

  • kind (Literal['phase_started'])

  • phase (Literal['feature_extraction', 'matching', 'geometric_verification', 'incremental_init', 'incremental_register', 'incremental_ba', 'global_rotation_avg', 'global_positioning', 'global_ba', 'hierarchical_cluster', 'hierarchical_merge', 'panorama', 'spherical', 'bundle_adjust', 'triangulate', 'relocalize', 'pose_graph_optimize', 'segment', 'export', 'vlad_index', 'backend_action', 'artifact_conversion'])

kind: Literal['phase_started']
phase: Phase
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.progress_event.SnapshotAvailable(*, schema_version=1, ts, job_id, task_id=None, seq, kind='snapshot_available', snapshot_seq, summary)[source]

Bases: _Base

Parameters:

  • schema_version (Literal[1])

  • ts (datetime)

  • job_id (str)

  • task_id (str | None)

  • seq (int)

  • kind (Literal['snapshot_available'])

  • snapshot_seq (int)

  • summary (dict[str, Any])

kind: Literal['snapshot_available']
snapshot_seq: int
summary: dict[str, Any]
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.progress_event.Warning_(*, schema_version=1, ts, job_id, task_id=None, seq, kind='warning', message)[source]

Bases: _Base

Parameters:

  • schema_version (Literal[1])

  • ts (datetime)

  • job_id (str)

  • task_id (str | None)

  • seq (int)

  • kind (Literal['warning'])

  • message (str)

kind: Literal['warning']
message: str
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Resume

async app.orchestrator.resume.resume_job(session, *, tenant_id, job_id, inline=None)[source]
Parameters:

  • session (AsyncSession)

  • tenant_id (str)

  • job_id (str)

  • inline (bool | None)

Return type:

Job

POST /v1/jobs/{id}:resume resets only failed / cancelled / cancelled_dirty tasks back to pending, then re-enqueues them (or runs inline in test mode). Mapping tasks specifically pick up from the latest MappingInput.pcmapin checkpoint in jobs/{id}/checkpoints/, so a multi-hour mapping run that crashed at 80% picks up near 80% rather than restarting from scratch.

Snapshot endpoints

Endpoint

Returns

GET /v1/reconstructions/{rid}/snapshots

List of sealed seq numbers

GET /v1/reconstructions/{rid}/snapshots/{seq}/cameras.json

JSON

GET /v1/reconstructions/{rid}/snapshots/{seq}/images.json

JSON

GET /v1/reconstructions/{rid}/snapshots/{seq}/points.bin

application/x-sfm-points-v1

GET /v1/reconstructions/{rid}/snapshots/{seq}/points_preview.bin

decimated

The API path-traversal-checks name, then serves the file directly with FileResponse. No DB read, no backend import, no race risk.