[KV Offloading] Detect and clean up stalled transfer jobs

[Alex-ai-future]

Problem

KV Offloading transfer jobs (store/load) can stall indefinitely when workers fail to report completion due to GPU hangs, CUDA driver bugs, or network failures. Without detection, stalled jobs cause cascading resource leaks:

• _blocks_being_loaded permanently occupied → other requests can't load those blocks
• Block ref_cnt stuck at -1 → GPU memory blocks can't be freed
• _req_status grows unbounded → memory leak
• Only recovery is reset_cache() → full stop, discarding all offloaded blocks

Solution

Add stalled transfer detection to OffloadingConnectorScheduler. Jobs exceeding a 300s timeout are automatically cleaned up.

Key design decisions

1. success=False for stalled stores: Stalled store data may be incomplete. Using success=False ensures the manager discards the data rather than marking it as stored. Other requests will re-offload from scratch instead of loading corrupt data.

2. _stalled_job_ids as dict[int, int]: After stalled cleanup, late worker reports may still arrive. The dict tracks remaining worker count per job so late reports decrement cleanly to 0, avoiding KeyError in multi-worker scenarios.

3. Throttled stall check (10s interval): Avoids traversing all jobs on every update_connector_output call.

4. _complete_job shared cleanup path: Extracted from both normal and stalled completion paths to eliminate ~45 lines of code duplication. The shared method does not call on_request_finished — that is request_finished()'s responsibility, not the job cleanup path's.

Testing

11 tests covering:

Test	Path
test_stalled_store_job_detected	Store timeout → complete_store(success=False)
test_stalled_load_unblocks_being_loaded	Load timeout → _blocks_being_loaded released
test_stalled_job_skips_late_worker_report	Late report doesn't trigger KeyError
test_stalled_job_with_missing_req_status	req_status=None edge case
test_stall_check_is_throttled	10s throttle verification
test_reset_cache_clears_stalled_job_state	reset_cache clears stalled state
test_stalled_job_no_double_on_request_finished	on_request_finished not called from job cleanup
test_normal_load_completes_clears_blocks_being_loaded	Normal load path cleanup
test_normal_last_job_deletes_req_status	Normal path _req_status deletion
test_complete_job_clears_block_id_to_pending_jobs	Fence cleanup
test_stalled_load_does_not_pass_success	complete_load without success kwarg

.venv/bin/python -m pytest tests/v1/kv_connector/unit/offloading_connector/test_scheduler.py -v

Follow-ups

• Configurable timeout: The 300s timeout is currently hardcoded. A follow-up will make it configurable via OffloadingSpec.
• Stalled transfer metrics: Currently only logger.error is emitted. A follow-up will add a stalled_transfers_total counter for observability and alerting.

Roadmap

Part of KV Offloading roadmap (#33689).

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @Alex-ai-future.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @Alex-ai-future.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @Alex-ai-future.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork