[Bugfix][KV Offloading] Drop conflicting finished-req store jobs to prevent data corruption

When _finished_req_store_jobs contains a store job and the block is
reallocated to a new request, the store job would read corrupted data.

Fix: In build_connector_meta, after merging _finished_req_store_jobs into
store_jobs, check if any of those jobs' blocks are in
_current_batch_allocated_block_ids. If so, drop the job entirely (same
behavior as before the request_finished fix — the last block is simply
not offloaded when its block is reused).

This prevents the race where wait() is called on an unsubmitted job
(no-op), allowing block corruption before the store reads it.

Signed-off-by: Alex <alex.tech.lab@outlook.com>

Author: Alex-ai-future

tests/v1/kv_connector/unit/offloading_connector/test_scheduler.py

@@ -2902,14 +2902,17 @@ def inject_request_b():
 
     # Run Request A with pre_schedule_fn to inject Request B at the right time
     # Use complete_transfers=True to allow jobs to complete and loop to exit
+    # decoded_tokens has three EOS tokens: one for Request A, two for Request B
+    # (async mode may need extra steps)
+    # In async mode, block reuse triggers flush; in sync mode, job is dropped before flush
     runner.run(
-        decoded_tokens=[EOS_TOKEN_ID],
+        decoded_tokens=[EOS_TOKEN_ID, EOS_TOKEN_ID, EOS_TOKEN_ID],
         complete_transfers=True,
         drain_finished_req_stores=True,
         pre_schedule_fn=inject_request_b,
         expected_stored=(0,),  # Request A stores block 0
-        expected_flushed=(0,),  # Request B reuses block 0, triggering flush
-        max_steps=20,
+        expected_flushed=(0,) if async_scheduling else (),  # flush depends on timing
+        max_steps=30,
     )
 
     # Verify Request B was injected

vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py

@@ -930,9 +930,8 @@ def _build_store_jobs_for_req(
             # A block_id of 0 means either a sliding window / SSM skip
             # or a stale entry that was zeroed out — skip it either way.
             offload_block_ids = group_state.block_ids[
-                start_block_idx * block_size_factor
-                + block_size_factor
-                - 1 : num_blocks * block_size_factor : block_size_factor
+                start_block_idx * block_size_factor + block_size_factor - 1 : num_blocks
+                * block_size_factor : block_size_factor
             ]
             assert len(offload_keys) == len(offload_block_ids)
 
@@ -987,9 +986,7 @@ def _build_store_jobs_for_req(
         for group_config, group_state in zip(
             self.config.kv_group_configs, req_status.group_states
         ):
-            is_sliding_window = (
-                group_config.sliding_window_size_in_blocks is not None
-            )
+            is_sliding_window = group_config.sliding_window_size_in_blocks is not None
             num_blocks = req_status.storable_blocks(
                 group_config, num_offloadable_tokens
             )
@@ -1137,9 +1134,53 @@ def build_connector_meta(
             )
 
         store_jobs = self._build_store_jobs(scheduler_output)
+
+        # Track which jobs came from _finished_req_store_jobs before merging
+        finished_req_job_ids = set(self._finished_req_store_jobs.keys())
+
         store_jobs.update(self._finished_req_store_jobs)
         self._finished_req_store_jobs.clear()
 
+        # Drop finished-req store jobs whose blocks have been reallocated.
+        # If a block is in _current_batch_allocated_block_ids, a new request
+        # is about to overwrite it. The store job would read stale/corrupted
+        # data, so we skip it entirely (same behavior as before the
+        # request_finished fix — the last block is simply not offloaded).
+        # Only drop jobs that came from _finished_req_store_jobs, not normal
+        # store jobs from _build_store_jobs.
+        if self._current_batch_allocated_block_ids:
+            jobs_to_drop = set()
+            for job_id in finished_req_job_ids:
+                if job_id not in store_jobs:
+                    continue  # Already removed or not present
+                job = self._jobs.get(job_id)
+                if job is not None and job.is_store:
+                    job_blocks = set(job.non_sliding_window_block_ids or [])
+                    job_blocks.update(job.sliding_window_block_ids or [])
+                    if job_blocks & self._current_batch_allocated_block_ids:
+                        jobs_to_drop.add(job_id)
+            for job_id in jobs_to_drop:
+                del store_jobs[job_id]
+                self._current_batch_jobs_to_flush.discard(job_id)
+                job = self._jobs.pop(job_id, None)
+                if job is not None:
+                    for bid in job.sliding_window_block_ids or []:
+                        pjs = self._block_id_to_pending_jobs.get(bid)
+                        if pjs is not None:
+                            pjs.discard(job_id)
+                            if not pjs:
+                                del self._block_id_to_pending_jobs[bid]
+                    for bid in job.non_sliding_window_block_ids or []:
+                        pjs = self._block_id_to_pending_jobs.get(bid)
+                        if pjs is not None:
+                            pjs.discard(job_id)
+                            if not pjs:
+                                del self._block_id_to_pending_jobs[bid]
+                    # Also remove from req_status.transfer_jobs
+                    req_status = self._req_status.get(job.req_id)
+                    if req_status is not None:
+                        req_status.transfer_jobs.discard(job_id)
+
         meta = OffloadingConnectorMetadata(
             load_jobs=self._current_batch_load_jobs,
             store_jobs=store_jobs,