[perf][tinker] Scan pending requests without loading their payloads

[avigyabb]

What

The Tinker engine's scheduling loop read every pending request's full payload on every poll, then threw away the ones it could not run. This makes the scheduler cost scale with the depth of the queue rather than with the work it dispatches — which is the multi-LoRA steady state, where some adapters are stepping while others have passes stacked behind a barrier.

This is the first of several PRs splitting up #1972. It contains only the scheduling-scan change — the covering index that originally rode along has moved to its own PR, since it turned out to be worth far less than this part (details under What this leaves on the table).

The problem

find_batchable_model_passes, find_batchable_sample, and find_single_requests each ran their own SELECT FutureDB ... over all pending rows and then filtered in Python. Three consequences:

1. Requests parked behind an optim_step/load_weights barrier had their payloads deserialized on every poll and immediately discarded. A forward_backward with 4×512 tokens is ~76 KiB of JSON, so a few hundred parked requests is tens of MB of wasted decode per 100 ms tick — enough to keep the engine CPU-bound and slow to dispatch anything.
2. The three finders issued eight queries between them, recomputing the same barrier set three times.
3. Scanning touched the payload-bearing table rows, so a poll that had nothing to run still paid for them.

The change

• scan_pending_requests reads only (request_id, model_id, request_type) for pending rows. Barrier detection and all three finders now work from that metadata as plain Python, and request_data is fetched only for the requests actually being dispatched. The main loop scans once and passes the result to all four finders, so a poll with nothing runnable is a single indexed query.
• Sample batch compatibility reads checkpoint_id through a JSON path extraction (json_extract) rather than loading whole prompts to reach one field. This is what makes skipping possible for samples at all: the scheduler needs that one field to group them, so without extracting it in the database you would have to pull every sample's payload into Python and there would be nothing left to skip.

Scheduling semantics are unchanged: barriers are still per-model, still keyed on request_id order, and the existing regression tests for that pin the behaviour.

Results

skyrl/benchmarks/bench_tinker_db.py (new; no GPU, drives the real engine functions, reports SQL statement and commit counts). 8 adapters, 64 concurrent submitters, 256 dispatchable requests plus 256 parked behind barriers, ~76 KiB payloads. Measured on this branch vs. an otherwise identical main worktree:

phase	before	after
scan_parked — backlog behind barriers, re-run every poll	247 ms	1.2 ms
scan_ready — scan + decode a dispatchable batch	928 ms	648 ms
find_batchable_sample, 640 samples, half held back by checkpoint	119 ms	72 ms
SQL statements per scan	8	2–3

scan_parked is the one that matters most: it is the cost the engine pays on every poll while it waits for a barrier to clear, and it previously grew with the queue. It is now flat.

scan_ready still decodes the batch it is about to dispatch, which is real work; the remaining win there comes from not also decoding the parked rows.

Where this is not a win

This is not strictly better, so here is everything I could find that it costs. Each row below is measured, not reasoned about — I was wrong twice while checking, so nothing here is an estimate unless it says so.

Neutral, verified rather than assumed:

	before	after
nothing parked — the case that should regress	661 ms	630 ms
insert 512 rows, one transaction	59.1 ms	59.4 ms
insert 128 rows, one transaction each (the API's submit shape)	91.2 ms	89.6 ms
find_batchable_sample, 640 samples, all dispatchable	143 ms	144 ms
status write-back (isolated)	~105 ms	~105 ms

The no-parked case is the one I expected to lose: it now pays for a metadata scan plus a primary-key payload fetch, where before a single query got rows and payloads together. It comes out ~5% ahead anyway, because the old path constructed full ORM FutureDB objects for every pending row and the new one returns lightweight tuples.

Actually worse:

1. Weaker atomicity. Scheduling metadata and payloads now come from two queries rather than one snapshot, so a row can in principle change between them. Guarded with if request_id in payloads, and safe today only because the single-threaded engine is the sole completer of these request types. That is a new invariant a future change could break, and it is the part of this PR I would most like a second opinion on.
2. No gain when everything is dispatchable — a hair worse, in fact. The whole saving is in not loading payloads for requests that cannot run, so a queue where everything can run has nothing to skip, and pays for one extra query. Measured on samples: 143 → 144 ms all-dispatchable, versus 119 → 72 ms when half are held back. Same for passes: 661 → 630 ms with nothing parked, versus 247 → 1.2 ms with a parked backlog. The change is a win in proportion to how much of the queue is blocked.
3. More surface area — three helper methods and a dataclass in place of three inline queries.

One number in the benchmark output is misleading: its complete phase reads ~1.7x slower here, and nothing in this PR touches the write path. It is a harness artifact — all phases share one database in a fixed order, and this write picks up process and I/O state from the scan phases that precede it. An isolated harness measuring only the status update reports ~105 ms either way, and --scan-iterations 1 gives identical numbers on both sides (~0.122 s). The phase docstring now says so. I took it for a real regression at first and only found otherwise by measuring, so it is called out here rather than left for a reviewer to trip over.

What this leaves on the table

A covering index on (status, request_id, request_type, model_id) would let the scan run without visiting the table rows at all. I had it in here originally and split it out, because measuring the two halves separately showed the split is very lopsided:

• full-row query → 247 ms
• 3-column query, existing status index (this PR) → 1.2 ms
• 3-column query, covering index → 1.2 ms (below noise end-to-end; ~2x on an isolated 512-row scan)

Not asking for the payload is the whole win. The index is a refinement worth roughly 2x on a microbenchmark, and it costs ~2.8x index storage, so it deserves its own accept-or-decline decision rather than being smuggled in here. It follows as a separate PR.

Not in this PR

Split out of #1972 and coming separately, in rough priority order:

• Covering index for the pending scan, as above.
• Retry the result write so a forwarded sample cannot hang forever (EXTERNAL requests are completed by exactly one fire-and-forget task; any failure there orphans them permanently). Relevant to the stuck Tinker Fully Async E2E job.
• Shared batched poller for retrieve_future, replacing per-caller polling that scales connection-pool checkouts with in-flight requests.
• orjson codec for JSON payload columns.
• DB path handling — node-local default, missing-directory creation, network-filesystem warning.

Also unaddressed anywhere: completed futures are never pruned and each retains its full payload, so a long run grows the SQLite file without bound. That needs a policy decision on how long results stay fetchable.

Testing

uv run --isolated --extra dev --extra jax --extra tinker pytest tests/tinker/ --ignore=tests/tinker/skyrl_train

55 passed. New coverage in tests/tinker/test_engine.py:

• metadata scan ordering, and that it excludes non-pending rows
• passes held back at a barrier while another model's pass still dispatches
• one-checkpoint-per-model sample batching, exercising the JSON path extraction
• a backlog larger than the bound-parameter chunk size, covering the chunking in _load_request_payloads

The pre-existing barrier regression tests pass untouched, which is the main evidence that scheduling semantics did not move.

🤖 Generated with Claude Code

[avigyabb]

could potentially combine these functions to reduce code

[gemini-code-assist]

Code Review

This pull request introduces a database benchmark script for the Tinker server and optimizes the engine's scheduling queries by separating the initial lightweight metadata scan from the loading of large request payloads. This prevents deserializing the entire backlog on every poll. The feedback focuses on enhancing robustness and atomicity by filtering for pending status when loading payloads and checkpoint IDs, as well as safely handling potential null values when retrieving checkpoint IDs.

[gemini-code-assist]

To address the atomicity concern mentioned in the PR description, we can filter by FutureDB.status == RequestStatus.PENDING when loading payloads. This ensures that if a request's status is updated, completed, or cancelled by another process/thread between the initial metadata scan and this payload load, we won't load or dispatch it.

Suggested change

	rows = session.exec(
	select(FutureDB.request_id, FutureDB.request_data).where(FutureDB.request_id.in_(chunk))
	).all()
	rows = session.exec(
	select(FutureDB.request_id, FutureDB.request_data)
	.where(FutureDB.request_id.in_(chunk))
	.where(FutureDB.status == RequestStatus.PENDING)
	).all()

[gemini-code-assist]

For consistency and robustness, filter by FutureDB.status == RequestStatus.PENDING when loading sample checkpoint IDs to avoid loading metadata for requests that are no longer pending.

Suggested change

	rows = session.exec(
	select(FutureDB.request_id, FutureDB.request_data["checkpoint_id"].as_string()).where(
	FutureDB.request_id.in_(chunk)
	)
	).all()
	rows = session.exec(
	select(FutureDB.request_id, FutureDB.request_data["checkpoint_id"].as_string())
	.where(FutureDB.request_id.in_(chunk))
	.where(FutureDB.status == RequestStatus.PENDING)
	).all()

[gemini-code-assist]

Using checkpoint_ids.get(op.request_id, "") can return None if the key exists in the dictionary but its value is None (which happens if the checkpoint_id field is missing or null in the JSON payload extracted by the database). Using checkpoint_ids.get(op.request_id) or "" safely coerces any None values to "".

Suggested change

	checkpoint_id = checkpoint_ids.get(op.request_id, "")
	checkpoint_id = checkpoint_ids.get(op.request_id) or ""