Async vs Sync Geoprocessing Workflows

Some spatial operations finish in forty milliseconds and some take four minutes, and the same agent has to handle both without making the fast ones slow or the slow ones invisible. The decision of which is which is not a performance detail — it changes the shape of the conversation, because an operation that becomes a job needs a handle, a progress story and a way to deliver its result to a user who has moved on.

This topic belongs to geospatial prompt engineering and tool routing and operates under the constraints set by cost and latency budgets for spatial agents. Its failure modes overlap with fallback routing for geospatial queries, which decides what to do when neither path can answer.

Foundational Principles

The decision is made from an estimate, not from experience. Whether an operation is inline or queued has to be decided before it runs, from the size of its inputs, because deciding afterwards means the user has already waited.

A queued job needs a handle the conversation can carry. An operation that becomes a job produces an identifier, and every subsequent turn must be able to ask about it. Without that the job is fire-and-forget and its result has nowhere to go.

Backpressure is part of correctness. A queue with no admission limit accepts work faster than it completes it, and the visible symptom is not an error but a completion time that grows until it is unbounded. Reject at the door and say so.

Step-by-Step Implementation Pipeline

1. Estimate the operation’s cost from its inputs

The estimate does not need to be accurate; it needs to sort operations into the right bucket. Feature counts, vertex counts and raster extents are the inputs that predict cost, and they are all cheap to obtain.

import logging
from dataclasses import dataclass
from typing import Literal, Optional

log = logging.getLogger("geoprocessing_mode")

Mode = Literal["inline", "queued"]

INLINE_BUDGET_S = 2.0
ALWAYS_QUEUE_S = 30.0


@dataclass(frozen=True)
class Estimate:
    seconds: float
    basis: str


def estimate_cost(op: str, feature_count: int, vertex_count: int,
                  raster_cells: int = 0) -> Estimate:
    """A rough per-operation cost model, measured rather than assumed."""
    rates = {                                    # seconds per unit, from production timings
        "buffer":    (2.0e-4, 1.0e-6, 0.0),
        "overlay":   (8.0e-4, 4.0e-6, 0.0),
        "dissolve":  (1.5e-3, 6.0e-6, 0.0),
        "zonal":     (1.0e-4, 5.0e-7, 3.0e-8),
    }
    per_feature, per_vertex, per_cell = rates.get(op, (1.0e-3, 5.0e-6, 1.0e-7))
    seconds = (feature_count * per_feature + vertex_count * per_vertex
               + raster_cells * per_cell)
    return Estimate(round(seconds, 3),
                    f"{op}: {feature_count} features, {vertex_count} vertices")

2. Choose the mode from the estimate and the context

The estimate sorts the obvious cases; the middle band needs the request’s own context — whether a user is waiting, how much of their budget is left, and whether the answer is needed to continue.

def choose_mode(est: Estimate, remaining_s: float, interactive: bool) -> tuple[Mode, str]:
    """Decide inline or queued, and say why."""
    if est.seconds >= ALWAYS_QUEUE_S:
        return "queued", f"estimated {est.seconds:.1f}s exceeds the inline ceiling"
    if est.seconds <= INLINE_BUDGET_S and est.seconds < remaining_s:
        return "inline", ""
    if not interactive:
        return "inline", "batch context: waiting is acceptable"
    if est.seconds >= remaining_s:
        return "queued", (f"estimated {est.seconds:.1f}s exceeds the "
                          f"{remaining_s:.1f}s remaining in this turn")
    return "queued", "estimated cost is above the interactive budget"

Returning the reason alongside the mode is what lets the agent explain itself. “This will take about forty seconds, so I have started it as a job” is a far better turn than a silent pause followed by a result.

3. Admit work into the queue, or refuse at the door

A queue that accepts everything degrades invisibly. Admission control against depth and estimated completion time turns that into an explicit rejection with a number attached.

@dataclass(frozen=True)
class Admission:
    accepted: bool
    reason: str
    estimated_wait_s: float


def admit(queue_depth: int, in_flight_s: float, est: Estimate,
          max_depth: int = 200, max_wait_s: float = 600.0) -> Admission:
    """Accept work only when it can complete in a time worth waiting for."""
    if queue_depth >= max_depth:
        return Admission(False, f"queue is full ({queue_depth} jobs)", in_flight_s)
    wait = in_flight_s + est.seconds
    if wait > max_wait_s:
        return Admission(False,
                         f"estimated completion in {wait / 60:.0f} min exceeds the limit",
                         wait)
    return Admission(True, "", wait)

Reporting the estimated wait even on rejection is what makes the refusal actionable: a user told “about twenty minutes” can decide to narrow the region, and one told “the queue is full” cannot. The rate-limiting mechanics are developed in backpressure and rate limiting for spatial API calls.

4. Give every job a handle and a status

The handle is what the conversation carries. It needs to be stable, short enough to appear in a sentence, and resolvable to a status without the original context.

@dataclass(frozen=True)
class Job:
    job_id: str
    op: str
    status: Literal["queued", "running", "done", "failed", "cancelled"]
    submitted_at: float
    estimated_s: float
    result_ref: Optional[str] = None
    error: Optional[str] = None


def status_sentence(job: Job, now: float) -> str:
    """What the agent tells the user about this job, in one line."""
    waited = now - job.submitted_at
    if job.status == "queued":
        return (f"Job {job.job_id} is queued; estimated {job.estimated_s:.0f}s of work, "
                f"waiting {waited:.0f}s so far.")
    if job.status == "running":
        return f"Job {job.job_id} is running, {waited:.0f}s elapsed."
    if job.status == "done":
        return f"Job {job.job_id} finished; the result is available."
    if job.status == "cancelled":
        return f"Job {job.job_id} was cancelled."
    return f"Job {job.job_id} failed: {job.error or 'no reason recorded'}."

5. Make the inline path cancellable too

An inline operation that exceeds its estimate is the worst case: the user is waiting and nothing is watching. Running it under the turn’s deadline, with cancellation, converts that into a bounded failure that can be re-offered as a job.

def run_inline(op_callable, est: Estimate, deadline_s: float):
    """Run inline under the deadline; on overrun, offer the queued path instead."""
    if est.seconds > deadline_s:
        raise WouldExceedDeadline(
            f"estimated {est.seconds:.1f}s against {deadline_s:.1f}s available")
    try:
        return op_callable(timeout=deadline_s)
    except TimeoutError as exc:
        log.info("inline operation exceeded its estimate: %s", exc)
        raise WouldExceedDeadline("the operation ran longer than estimated") from exc

6. Deliver results to a conversation that has moved on

A job finishing after the user has asked three other questions needs somewhere to go. The workable pattern is a result reference the agent can mention on the next turn, plus a pinned entry in the conversation state so the handle survives history pruning.

def pending_jobs_hint(jobs: list[Job], now: float) -> str:
    """A short line for the model's context so it can mention finished work."""
    done = [j for j in jobs if j.status == "done"]
    running = [j for j in jobs if j.status in {"queued", "running"}]
    parts = []
    if done:
        parts.append(f"{len(done)} job(s) finished: " + ", ".join(j.job_id for j in done))
    if running:
        parts.append(f"{len(running)} still running")
    return "; ".join(parts)

7. Retry jobs, not inline calls

An inline operation that fails has a user waiting and should degrade rather than retry. A queued job has nobody waiting and can retry with backoff, provided the failure is transient and the retry is bounded.

def should_retry(job: Job, attempts: int, error_class: str,
                 max_attempts: int = 3) -> tuple[bool, float]:
    """Retry queued work on transient failures only, with bounded backoff."""
    if job.status != "failed" or attempts >= max_attempts:
        return False, 0.0
    if error_class not in {"transient", "resource"}:
        return False, 0.0
    return True, min(60.0, 2.0 ** attempts)

8. Report queue health as a first-class metric

Depth, wait time and rejection rate together describe whether the async path is working. Depth alone does not: a short queue that rejects half its arrivals is in worse shape than a long one that accepts everything and completes it.

def queue_health(depth: int, p95_wait_s: float, rejected: int, accepted: int) -> dict:
    total = rejected + accepted
    return {
        "depth": depth,
        "p95_wait_s": round(p95_wait_s, 1),
        "rejection_rate": round(rejected / total, 4) if total else 0.0,
        "healthy": depth < 150 and p95_wait_s < 300 and (not total or rejected / total < 0.05),
    }

9. Keep the two paths behaviourally identical

The most confusing bug in a dual-path system is an operation that produces one answer inline and a different one through the queue. It happens easily: the worker runs with different defaults, reads a different snapshot, or applies a simplification the inline path does not. The fix is structural — one implementation, invoked two ways.

def execute(op: str, params: dict, snapshot_id: str) -> dict:
    """The single implementation. Both paths call this; neither has its own defaults."""
    return OPERATIONS[op](**params, snapshot=snapshot_id)


def run_any(op: str, params: dict, mode: Mode, snapshot_id: str, enqueue, deadline_s: float):
    """Dispatch by mode without duplicating the operation itself."""
    if mode == "inline":
        return execute(op, params, snapshot_id)
    return enqueue(op=op, params=params, snapshot=snapshot_id)

Passing the snapshot identifier through both paths is the detail that matters most. A queued job that runs twenty minutes later against a corpus that has since been rebuilt produces an answer to a slightly different question, and the user has no way to tell — the handle they were given makes it look like the same request.

10. Decide what a cancelled job leaves behind

Cancellation is easy to offer and easy to implement badly. A job cancelled mid-write can leave partial output that a later read treats as complete, which is a worse outcome than not offering cancellation at all.

def cancel(job: Job, store) -> Job:
    """Mark cancelled and remove partial output; never leave a half-written result."""
    if job.status in {"done", "failed", "cancelled"}:
        return job
    try:
        if job.result_ref:
            store.delete(job.result_ref)          # partial output is not a result
    except Exception as exc:
        log.warning("could not remove partial output for %s: %s", job.job_id, exc)
    return Job(job.job_id, job.op, "cancelled", job.submitted_at,
               job.estimated_s, None, None)

Writing results under a temporary reference and promoting them atomically on completion is the arrangement that makes this safe. Without it, cancellation and failure both leave debris that looks like output, and the read path has to distinguish them by inspecting content — which it will do incorrectly at least once.

Operating This Stage Over Time

The cost model is the part that ages. Operation timings are measured against a corpus and a machine, and both change; a buffer that took two seconds against a million features will take longer against ten million, and every mode decision made from the old figures will send work down the inline path that no longer belongs there. Refresh the rates from observed durations on a schedule, and alert when the median estimate error exceeds a third.

The second drift is in what users ask for. A system whose questions were originally local acquires regional ones as people discover it works, and the share of operations landing in the queued band grows without anything in the code changing. Track the inline/queued split as a ratio rather than as counts, because the ratio moving is the signal that the workload has changed shape.

The third is a quiet one: jobs that nobody collects. Results delivered to a conversation that ended are storage with no consumer, and they accumulate. Expire them on a schedule tied to how long a conversation plausibly lasts, and count the expirations — a high rate means the delivery path is not working rather than that users are careless.

Finally, keep the two paths behaviourally identical where it matters. An operation that produces one answer inline and a subtly different one through the queue — because the job runs with different defaults, or against a different snapshot — is a bug that only appears at the boundary, and the boundary moves whenever the cost model is refreshed.

Failure Modes & Root Causes

The inline operation that was not. An operation estimated at one second takes ninety because the input was larger than the estimate assumed. Root cause: a cost model that ignores the dominant input. Mitigation: include vertex and cell counts, not just feature counts, and cancel on the deadline.

The unbounded queue. Completion times grow until the feature is unusable, with no error at any point. Root cause: no admission control. Mitigation: reject on depth and estimated wait, with the wait reported.

The orphaned job. Work completes and the result has nowhere to go because the conversation ended or the handle was pruned. Root cause: handles not pinned in conversation state. Mitigation: pin them, and expire results deliberately.

The retry storm. A failing dependency causes every queued job to retry simultaneously, which keeps it failing. Root cause: unbounded retries with no jitter. Mitigation: bounded attempts, exponential backoff, and classification so only transient failures retry.

Production Validation Protocols

1. Mode-decision test. Assert that an input above the queue threshold is never run inline, using a fixture sized to trigger it.
2. Admission-under-load test. Assert the queue rejects with a stated wait once depth or projected wait exceeds the limits.
3. Handle-survival test. Assert a job handle survives history pruning and is resolvable after the conversation has moved on.
4. Deadline-cancellation test. Assert an inline operation that overruns is cancelled and re-offered as a job rather than allowed to run.
5. Estimate-accuracy indicator. Compare estimated against actual durations and alert when the median error exceeds a third.
6. Queue-health publication. Publish depth, tail wait and rejection rate together; any one alone is misleading.

Frequently Asked Questions

Related

• Up to the section overview: Geospatial Prompt Engineering and Tool Routing
• Technique: Handling Async Spatial Processing in Python Workflows
• Technique: Backpressure and Rate Limiting for Spatial API Calls
• Technique: Choosing Between Queued and Inline Geoprocessing
• Peer topic: Cost and Latency Budgets for Spatial Agents
• Peer topic: Multi-Step Spatial Agent Orchestration