_execute_bounded can refuse a query that actually returned, when the worker finishes as the join expires

[sandeep-agami]

What happens

_execute_bounded in packages/agami-core/src/execute_sql.py runs the executor on a worker thread and joins with a deadline. When the worker finishes at the same instant the join expires, the outcome depends on which thread acquires _abandoned_lock first.

def call() -> None:
    try:
        outcome["result"] = ctx.run(executor.execute, sql, creds, profile=profile)
    except BaseException as exc:
        outcome["error"] = exc
    finally:
        with _abandoned_lock:                 # <- the worker has to WAIT for the lock
            if outcome.get("abandoned"):
                _abandoned_workers -= 1
            outcome["finished"] = True

worker.join(timeout_s + _OUTER_BOUND_SKEW_S)
with _abandoned_lock:
    abandoned = not outcome.get("finished")   # <- may run before the worker sets it
    if abandoned:
        _abandoned_workers += 1
        outcome["abandoned"] = True
if abandoned:
    raise _OuterBoundExpired(_OUTLIVED_OUTER_BOUND)

Worker acquires first: finished is set, the caller sees it, the result is used. Correct.

Caller acquires first: finished is still unset, so the call is marked abandoned and a resource_limit refusal is raised, even though executor.execute had already returned successfully. The worker then takes the lock and decrements the counter, so the accounting is right, but the result is discarded and the caller is told the executor did not return.

The comment at that site currently claims the opposite — that a worker finishing in this window "is counted as having returned, and its result is used rather than a refusal invented over the top of it". That is true of the counter and false of the outcome.

Impact

Low, and it fails closed. The caller gets a spurious refusal for a query that had in fact completed; nothing runs unbounded and no data is lost. A retry succeeds.

It is also hard to reach. On the built-in executor the per-statement watchdog fires _OUTER_BOUND_SKEW_S seconds earlier, so the outer bound is only reached when a driver cancel has already failed. That leaves the injected-executor path, where a query would have to return within the few microseconds (worst case a GIL switch, so single-digit milliseconds) between the join expiring and the next lock acquisition.

Why worker.is_alive() is not the fix

The obvious repair is to decide abandonment from worker.is_alive() after the join. It does not close the window: a worker that has finished the work but is still inside its finally, waiting on the lock, is alive, so the same ordering still reads as abandoned.

Suggested fix

Signal the completion of the work separately from the accounting, so the flag the caller reads cannot be delayed by lock contention:

done = threading.Event()

def call() -> None:
    try:
        outcome["result"] = ctx.run(executor.execute, sql, creds, profile=profile)
    except BaseException as exc:
        outcome["error"] = exc
    finally:
        done.set()                            # the work is over, before contending for any lock
        with _abandoned_lock:
            if outcome.get("abandoned"):
                _abandoned_workers -= 1
            outcome["finished"] = True

worker.start()
done.wait(timeout_s + _OUTER_BOUND_SKEW_S)
with _abandoned_lock:
    abandoned = not done.is_set()
    ...

A residual window remains where the work completes exactly as the wait expires, but it then genuinely means the work was unfinished at the bound, which is the irreducible boundary condition of any timeout.

Definition of done

• Abandonment is decided from a signal the worker sets before any lock contention.
• A regression test forces the losing ordering deterministically (hold _abandoned_lock so the worker blocks in its finally, then assert the caller still returns the result rather than a resource_limit refusal) and fails without the fix.
• The comment at the call site describes what the code does.

Found by review on #174.