[kv_offload] Emit tier-owned BlockStored events from FS/OBJ secondary tiers (#47923)

Signed-off-by: Change72 <changg@nvidia.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Or Ozeri <oro@il.ibm.com>

docs/features/kv_offloading_usage.md

@@ -81,6 +81,8 @@ vllm serve <model> \
 
 Each entry in `secondary_tiers` is a dict with a required `type` field plus tier-specific fields.
 
+The filesystem and object-store tiers can publish hash-only `BlockStored` KV events for blocks they successfully store, tagged with a stable per-tier `medium` (`FS` for the filesystem tier, `OBJ` for the object-store tier). Set `enable_kv_events: true` in the tier's entry to opt in; events are published only when KV cache events are also enabled globally via `--kv-events-config`.
+
 ### Filesystem (FS)
 
 The filesystem tier (`type: "fs"`) writes blocks to a directory on local storage.
@@ -91,6 +93,7 @@ The filesystem tier (`type: "fs"`) writes blocks to a directory on local storage
 | `root_dir` | yes | — | Base directory; vLLM creates subdirectories beneath it (see [On-Disk Layout](#on-disk-layout)). |
 | `n_read_threads` | no | `16` | Read-priority I/O threads (load path). |
 | `n_write_threads` | no | `16` | Write-priority I/O threads (store path). |
+| `enable_kv_events` | no | `false` | Publish `BlockStored` KV events (medium `FS`) for successfully stored blocks. Requires KV cache events to be enabled globally. |
 
 Each thread group prefers its own queue but pulls from the other when its primary queue is empty, so a write-heavy or read-heavy burst won't leave the off-priority queue waiting. Size the totals to your storage's effective concurrency.
 
@@ -120,6 +123,31 @@ To enable KV cache sharing between multiple vLLM instances using the same `root_
 PYTHONHASHSEED=0 vllm serve ...
 ```
 
+### Object Store (OBJ)
+
+The object-store tier (`type: "obj"`) offloads blocks to an S3-compatible object store through the NIXL OBJ backend.
+
+| Key | Required | Default | Notes |
+| --- | --- | --- | --- |
+| `type` | yes | — | Must be `obj`. |
+| `store_config` | yes | — | Object store connection parameters (see below). |
+| `prefix` | no | `""` | Key prefix prepended to all object keys. |
+| `io_threads` | no | `4` | Number of NIXL OBJ backend I/O threads. |
+| `enable_kv_events` | no | `false` | Publish `BlockStored` KV events (medium `OBJ`) for successfully stored blocks. Requires KV cache events to be enabled globally. |
+
+`store_config` fields:
+
+| Key | Required | Default | Notes |
+| --- | --- | --- | --- |
+| `bucket` | yes | — | Bucket name. |
+| `endpoint_override` | yes | — | Object store endpoint host; the URL scheme is set separately via `scheme`. |
+| `scheme` | no | `http` | `http` or `https`. |
+| `access_key`, `secret_key`, `session_token` | no | `""` | Explicit credentials. When left empty, the NIXL OBJ plugin falls back to the AWS SDK default credential provider chain (IAM roles, environment variables, credential files), which enables workload-identity auth on Kubernetes. |
+| `region` | no | `""` | Bucket region, if the endpoint requires one. |
+| `ca_bundle` | no | `""` | CA bundle path for TLS verification. |
+
+Object keys follow the same run-configuration digest scheme as the filesystem tier (see [On-Disk Layout](#on-disk-layout)) and are stored under the optional `prefix`. The [Cross-Process Sharing](#cross-process-sharing) requirement (`PYTHONHASHSEED`) applies to shared buckets as well, so instances sharing a bucket produce identical keys for identical content. At startup the tier probes object store connectivity and fails fast with a configuration error if the bucket is unreachable.
+
 ### P2P (Including P/D)
 
 The P2P tier (`type: "p2p"`) shares completed KV blocks between vLLM instances over RDMA via NIXL. Each instance binds a control socket on `host:port` and exchanges blocks directly with peers — no shared filesystem required.

tests/v1/kv_offload/tiering/test_fs_tier.py

@@ -18,8 +18,10 @@ import numpy as np
 import pytest
 import torch
 
+from vllm.distributed.kv_events import MEDIUM_FS
 from vllm.v1.kv_offload.base import (
     LookupResult,
+    OffloadingEvent,
     OffloadKey,
     ReqContext,
     ScheduleEndContext,
@@ -58,6 +60,16 @@ _MOCK_OFFLOADING_SPEC.kv_cache_config = _MOCK_KV_CACHE_CONFIG
 _MOCK_OFFLOADING_SPEC.block_size_factor = 1
 
 
+def _make_offloading_spec(enable_kv_cache_events: bool) -> MagicMock:
+    """Mock spec with an explicit global KV events flag."""
+    spec = MagicMock()
+    spec.vllm_config = _MOCK_VLLM_CONFIG
+    spec.kv_cache_config = _MOCK_KV_CACHE_CONFIG
+    spec.block_size_factor = 1
+    spec.kv_events_config.enable_kv_cache_events = enable_kv_cache_events
+    return spec
+
+
 def key(n: int) -> OffloadKey:
     return make_offload_key(n.to_bytes(8, "big"), 0)
 
@@ -164,6 +176,23 @@ def fs_tier(tmp_path):
     tier.shutdown()
 
 
+@pytest.fixture
+def fs_tier_with_events(tmp_path):
+    tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS)
+    mock_view = memoryview(tensor.numpy())
+    tier = FileSystemTierManager(
+        offloading_spec=_make_offloading_spec(enable_kv_cache_events=True),
+        primary_kv_view=mock_view,
+        tier_type="fs",
+        root_dir=str(tmp_path),
+        n_read_threads=4,
+        n_write_threads=4,
+        enable_kv_events=True,
+    )
+    yield tier
+    tier.shutdown()
+
+
 # ---------------------------------------------------------------------------
 # Tests
 # ---------------------------------------------------------------------------
@@ -396,3 +425,170 @@ def test_batch_lookup_dispatch(fs_tier, monkeypatch, use_c_ext):
 
     results = lookup_and_wait(tier, [key(1), key(2)])
     assert results == [LookupResult.HIT, LookupResult.MISS]
+
+
+# ---------------------------------------------------------------------------
+# KV events
+# ---------------------------------------------------------------------------
+
+
+def test_successful_store_emits_stored_event(fs_tier_with_events):
+    """A completed store job emits one stored event with the job's keys."""
+    tier = fs_tier_with_events
+    keys = [key(1), key(2)]
+    tier.submit_store(make_job(1, keys, [0, 1]))
+    assert all(r.success for r in drain(tier))
+
+    events = list(tier.take_events())
+    assert len(events) == 1
+    assert events[0].keys == keys
+    # Literal medium pins the wire contract, not just the constant choice.
+    assert events[0].medium == "FS"
+    assert not events[0].removed
+    # take_events drains the buffer.
+    assert list(tier.take_events()) == []
+
+
+def test_load_job_emits_no_event(fs_tier_with_events):
+    tier = fs_tier_with_events
+    tier.submit_store(make_job(1, [key(1)], [0]))
+    results = drain(tier)
+    assert len(results) == 1
+    assert results[0].success
+    list(tier.take_events())
+
+    tier.submit_load(make_job(2, [key(1)], [1], is_promotion=True))
+    results = drain(tier)
+    assert len(results) == 1
+    assert results[0].success
+    assert list(tier.take_events()) == []
+
+
+def test_mixed_job_results_emit_event_only_for_successful_job(
+    fs_tier_with_events, monkeypatch
+):
+    """With a failed and a successful store job in flight, exactly one event
+    is emitted and its keys belong to the successful job."""
+    import vllm.v1.kv_offload.tiering.fs.manager as mgr_mod
+
+    tier = fs_tier_with_events
+    failing_path = tier.file_mapper.get_file_name(key(1))
+    original_store_block = mgr_mod.store_block
+
+    def flaky_store_block(dest_path, *args, **kwargs):
+        if dest_path == failing_path:
+            raise OSError("injected store failure")
+        return original_store_block(dest_path, *args, **kwargs)
+
+    monkeypatch.setattr(mgr_mod, "store_block", flaky_store_block)
+
+    tier.submit_store(make_job(1, [key(1)], [0]))
+    tier.submit_store(make_job(2, [key(2)], [1]))
+    results = drain(tier)
+    assert len(results) == 2
+    by_id = {r.job_id: r for r in results}
+    assert not by_id[1].success
+    assert by_id[2].success
+
+    events = list(tier.take_events())
+    assert len(events) == 1
+    assert events[0].keys == [key(2)]
+
+
+def test_partially_failed_store_emits_no_event(fs_tier_with_events, monkeypatch):
+    """A store job with any failed block emits no event for the whole job."""
+    import vllm.v1.kv_offload.tiering.fs.manager as mgr_mod
+
+    tier = fs_tier_with_events
+    failing_path = tier.file_mapper.get_file_name(key(2))
+    original_store_block = mgr_mod.store_block
+
+    def flaky_store_block(dest_path, *args, **kwargs):
+        if dest_path == failing_path:
+            raise OSError("injected store failure")
+        return original_store_block(dest_path, *args, **kwargs)
+
+    monkeypatch.setattr(mgr_mod, "store_block", flaky_store_block)
+
+    tier.submit_store(make_job(1, [key(1), key(2)], [0, 1]))
+    results = drain(tier)
+    assert len(results) == 1
+    assert not results[0].success
+    assert list(tier.take_events()) == []
+    assert tier._store_job_keys == {}
+
+
+def test_events_disabled_by_default(fs_tier):
+    tier, _ = fs_tier
+    tier.submit_store(make_job(1, [key(1)], [0]))
+    results = drain(tier)
+    assert len(results) == 1
+    assert results[0].success
+    assert tier.events is None
+    assert tier._store_job_keys == {}
+    assert list(tier.take_events()) == []
+
+
+def test_events_require_global_kv_events_flag(tmp_path):
+    """Tier-level opt-in alone is not enough; the global flag gates events."""
+    tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS)
+    tier = FileSystemTierManager(
+        offloading_spec=_make_offloading_spec(enable_kv_cache_events=False),
+        primary_kv_view=memoryview(tensor.numpy()),
+        tier_type="fs",
+        root_dir=str(tmp_path),
+        enable_kv_events=True,
+    )
+    try:
+        assert tier.events is None
+        tier.submit_store(make_job(1, [key(1)], [0]))
+        results = drain(tier)
+        assert len(results) == 1
+        assert results[0].success
+        assert list(tier.take_events()) == []
+        assert tier._store_job_keys == {}
+    finally:
+        tier.shutdown()
+
+
+def test_cascade_store_emits_fs_event_through_tiering_manager(tmp_path):
+    """A GPU->CPU->fs cascade surfaces the tier-owned FS stored event via the
+    TieringOffloadingManager's aggregated take_events()."""
+    from vllm.v1.kv_offload.tiering.manager import (
+        CPUPrimaryTierOffloadingManager,
+        TieringOffloadingManager,
+    )
+
+    tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS)
+    view = memoryview(tensor.numpy())
+    mock_region = MagicMock()
+    mock_region.create_kv_memoryview.return_value = view
+    primary = CPUPrimaryTierOffloadingManager(num_blocks=4, mmap_region=mock_region)
+    tier = FileSystemTierManager(
+        offloading_spec=_make_offloading_spec(enable_kv_cache_events=True),
+        primary_kv_view=primary.get_kv_memoryview(),
+        tier_type="fs",
+        root_dir=str(tmp_path),
+        enable_kv_events=True,
+    )
+    manager = TieringOffloadingManager(primary_tier=primary, secondary_tiers=[tier])
+    try:
+        keys = [key(1), key(2)]
+        manager.on_new_request(_CTX)
+        assert manager.prepare_store(keys, _CTX) is not None
+        manager.complete_store(keys, _CTX)  # cascades to the fs tier
+
+        events: list[OffloadingEvent] = []
+        ctx = ScheduleEndContext(new_req_ids=[], preempted_req_ids=())
+        deadline = time.monotonic() + 5.0
+        while time.monotonic() < deadline and not events:
+            manager.on_schedule_end(ctx)
+            events.extend(manager.take_events())
+            time.sleep(0.01)
+
+        fs_events = [e for e in events if e.medium == MEDIUM_FS]
+        assert len(fs_events) == 1
+        assert set(fs_events[0].keys) == set(keys)
+        assert not fs_events[0].removed
+    finally:
+        tier.shutdown()

tests/v1/kv_offload/tiering/test_obj_tier.py

@@ -179,8 +179,19 @@ class MockNixlAgent:
 # ---------------------------------------------------------------------------
 
 
+def _make_events_spec(enable_kv_cache_events: bool) -> SimpleNamespace:
+    """Offloading spec stub with an explicit global KV events flag."""
+    return SimpleNamespace(
+        vllm_config=_make_vllm_config(),
+        kv_cache_config=SimpleNamespace(kv_cache_groups=[]),
+        kv_events_config=SimpleNamespace(enable_kv_cache_events=enable_kv_cache_events),
+    )
+
+
 def _make_tier(
     num_blocks: int = 4,
+    offloading_spec: SimpleNamespace = _OFFLOADING_SPEC,
+    **tier_kwargs,
 ) -> tuple[ObjectStoreSecondaryTierManager, MockNixlAgent]:
     """Create a tier backed by a fresh MockNixlAgent."""
     mock_agent = MockNixlAgent()
@@ -194,11 +205,12 @@ def _make_tier(
         ),
     ):
         tier = ObjectStoreSecondaryTierManager(
-            offloading_spec=_OFFLOADING_SPEC,
+            offloading_spec=offloading_spec,
             primary_kv_view=view,
             tier_type="obj",
             store_config=_STORE_CONFIG,
             prefix=_RUN_PREFIX,
+            **tier_kwargs,
         )
     return tier, mock_agent
 
@@ -421,6 +433,100 @@ class TestMockObjTierShutdown:
         tier.shutdown()  # must not raise
 
 
+class TestObjTierKVEvents:
+    def setup_method(self):
+        self.tier, self.agent = _make_tier(
+            offloading_spec=_make_events_spec(enable_kv_cache_events=True),
+            enable_kv_events=True,
+        )
+
+    def test_successful_store_emits_stored_event(self):
+        """A completed store transfer emits one stored event with the job's keys."""
+        keys = [key(1), key(2)]
+        self.tier.submit_store(make_job(1, keys, [0, 1]))
+        assert all(r.success for r in drain(self.tier))
+
+        events = list(self.tier.take_events())
+        assert len(events) == 1
+        assert events[0].keys == keys
+        # Literal medium pins the wire contract, not just the constant choice.
+        assert events[0].medium == "OBJ"
+        assert not events[0].removed
+        # take_events drains the buffer.
+        assert list(self.tier.take_events()) == []
+
+    def test_mixed_job_results_emit_event_only_for_successful_job(self):
+        """With a failed and a successful store job resolving in the same
+        poll, exactly one event is emitted and its keys belong to the
+        successful job."""
+        original = self.agent.check_xfer_state
+        self.agent.check_xfer_state = lambda h: "ERR" if h._id == 0 else original(h)
+        self.tier.submit_store(make_job(1, [key(1)], [0]))  # handle 0: fails
+        self.tier.submit_store(make_job(2, [key(2)], [1]))  # handle 1: succeeds
+        results = drain(self.tier)
+        by_id = {r.job_id: r for r in results}
+        assert not by_id[1].success
+        assert by_id[2].success
+
+        events = list(self.tier.take_events())
+        assert len(events) == 1
+        assert events[0].keys == [key(2)]
+        assert self.tier._store_job_keys == {}
+
+    def test_load_job_emits_no_event(self):
+        self.tier.submit_store(make_job(1, [key(1)], [0]))
+        results = drain(self.tier)
+        assert len(results) == 1
+        assert results[0].success
+        list(self.tier.take_events())
+
+        self.tier.submit_load(make_job(2, [key(1)], [0]))
+        results = drain(self.tier)
+        assert len(results) == 1
+        assert results[0].success
+        assert list(self.tier.take_events()) == []
+
+    def test_failed_transfer_emits_no_event(self):
+        self.agent.check_xfer_state = lambda h: "ERR"
+        self.tier.submit_store(make_job(1, [key(1)], [0]))
+        results = drain(self.tier)
+        assert not results[0].success
+        assert list(self.tier.take_events()) == []
+        assert self.tier._store_job_keys == {}
+
+    def test_submission_failure_emits_no_event(self):
+        self.agent.make_prepped_xfer = lambda *a, **k: None
+        self.tier.submit_store(make_job(1, [key(1)], [0]))
+        results = list(self.tier.get_finished_jobs())
+        assert not results[0].success
+        assert list(self.tier.take_events()) == []
+        assert self.tier._store_job_keys == {}
+
+    def test_events_disabled_by_default(self):
+        tier, _ = _make_tier()
+        tier.submit_store(make_job(1, [key(1)], [0]))
+        results = drain(tier)
+        assert len(results) == 1
+        assert results[0].success
+        assert tier.events is None
+        assert tier._store_job_keys == {}
+        assert list(tier.take_events()) == []
+
+    def test_events_require_global_kv_events_flag(self):
+        """Tier-level opt-in alone is not enough; the global flag gates events."""
+        tier, _ = _make_tier(
+            offloading_spec=_make_events_spec(enable_kv_cache_events=False),
+            enable_kv_events=True,
+        )
+        tier.submit_store(make_job(1, [key(1)], [0]))
+        results = drain(tier)
+        assert len(results) == 1
+        assert results[0].success
+        assert tier.events is None
+        assert tier._store_job_keys == {}
+        assert list(tier.take_events()) == []
+
+
 class TestObjStoreConfig:
     def test_explicit_credentials_included(self):
         cfg = ObjStoreConfig(

vllm/distributed/kv_events.py

@@ -44,6 +44,8 @@ class KVCacheEvent(
 
 MEDIUM_GPU = "GPU"
 MEDIUM_CPU = "CPU"
+MEDIUM_FS = "FS"
+MEDIUM_OBJ = "OBJ"
 
 
 class BlockStored(KVCacheEvent):

vllm/v1/kv_offload/tiering/fs/manager.py

@@ -19,7 +19,7 @@ import functools
 import json
 import os
 from collections.abc import Iterable
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 try:
     from vllm.fs_io_C import batch_lookup as batch_lookup_C
@@ -30,11 +30,18 @@ except ImportError:
 
 from typing_extensions import override
 
+from vllm.distributed.kv_events import MEDIUM_FS
 from vllm.logger import init_logger
-from vllm.v1.kv_offload.base import LookupResult, OffloadKey, ReqContext
+from vllm.v1.kv_offload.base import (
+    LookupResult,
+    OffloadingEvent,
+    OffloadKey,
+    ReqContext,
+)
 from vllm.v1.kv_offload.file_mapper import FileMapper
 from vllm.v1.kv_offload.tiering.async_lookup import AsyncLookupManager
 from vllm.v1.kv_offload.tiering.base import (
+    JobId,
     JobMetadata,
     JobResult,
     RequestOffloadingContext,
@@ -92,6 +99,8 @@ class FileSystemTierManager(SecondaryTierManager):
         content.
     """
 
+    medium: ClassVar[str] = MEDIUM_FS
+
     def __init__(
         self,
         offloading_spec: "OffloadingSpec",
@@ -100,6 +109,7 @@ class FileSystemTierManager(SecondaryTierManager):
         root_dir: str,
         n_read_threads: int = 16,
         n_write_threads: int = 16,
+        enable_kv_events: bool = False,
     ):
         """
         Args:
@@ -110,9 +120,26 @@ class FileSystemTierManager(SecondaryTierManager):
             root_dir: Root directory for block files.
             n_read_threads: Number of read-priority I/O threads.
             n_write_threads: Number of write-priority I/O threads.
+            enable_kv_events: Emit BlockStored KV events for blocks
+                successfully stored to this tier. Effective only when KV
+                cache events are enabled globally (kv_events_config).
         """
         super().__init__(offloading_spec, primary_kv_view, tier_type)
 
+        self.events: list[OffloadingEvent] | None = None
+        if enable_kv_events:
+            if offloading_spec.kv_events_config.enable_kv_cache_events:
+                self.events = []
+            else:
+                logger.warning(
+                    "enable_kv_events is set on secondary tier '%s' but KV "
+                    "cache events are disabled globally; the tier will not "
+                    "emit events.",
+                    tier_type,
+                )
+        # Keys of in-flight store jobs, tracked only when events are enabled.
+        self._store_job_keys: dict[JobId, list[OffloadKey]] = {}
+
         # Extract block size from primary view
         assert primary_kv_view.strides is not None, (
             "primary_kv_view.strides cannot be None"
@@ -157,6 +184,8 @@ class FileSystemTierManager(SecondaryTierManager):
 
     @override
     def submit_store(self, job_metadata: JobMetadata) -> None:
+        if self.events is not None:
+            self._store_job_keys[job_metadata.job_id] = list(job_metadata.keys)
         tasks = (
             functools.partial(
                 store_block,
@@ -188,10 +217,22 @@ class FileSystemTierManager(SecondaryTierManager):
         """
         Collect completed jobs from the finished-jobs queue.
         """
-        return (
-            JobResult(job_id=job_id, success=success)
-            for job_id, success in self._pool.get_finished()
-        )
+        results = []
+        for job_id, success in self._pool.get_finished():
+            if self.events is not None:
+                keys = self._store_job_keys.pop(job_id, None)
+                if success and keys:
+                    self.events.append(
+                        OffloadingEvent(keys=keys, medium=self.medium, removed=False)
+                    )
+            results.append(JobResult(job_id=job_id, success=success))
+        return results
+
+    @override
+    def take_events(self) -> Iterable[OffloadingEvent]:
+        if self.events is not None:
+            yield from self.events
+            self.events.clear()
 
     @override
     def drain_jobs(self) -> None:

vllm/v1/kv_offload/tiering/obj/manager.py

@@ -5,15 +5,22 @@
 import ctypes
 import time
 from collections.abc import Iterable
-from typing import TYPE_CHECKING, NamedTuple
+from typing import TYPE_CHECKING, ClassVar, NamedTuple
 
+from vllm.distributed.kv_events import MEDIUM_OBJ
 from vllm.distributed.nixl_utils import NixlWrapper as nixl_agent
 from vllm.distributed.nixl_utils import nixl_agent_config
 from vllm.logger import init_logger
-from vllm.v1.kv_offload.base import LookupResult, OffloadKey, ReqContext
+from vllm.v1.kv_offload.base import (
+    LookupResult,
+    OffloadingEvent,
+    OffloadKey,
+    ReqContext,
+)
 from vllm.v1.kv_offload.file_mapper import FileMapper
 from vllm.v1.kv_offload.tiering.async_lookup import AsyncLookupManager
 from vllm.v1.kv_offload.tiering.base import (
+    JobId,
     JobMetadata,
     JobResult,
     RequestOffloadingContext,
@@ -90,6 +97,8 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
     primary tier. Object keys are formed as ``{prefix}/{hash_shard}/{hash}.bin``.
     """
 
+    medium: ClassVar[str] = MEDIUM_OBJ
+
     def __init__(
         self,
         offloading_spec: "OffloadingSpec",
@@ -98,8 +107,36 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
         store_config: dict,
         prefix: str = "",
         io_threads: int = 4,
+        enable_kv_events: bool = False,
     ):
+        """
+        Args:
+            offloading_spec: Offloading configuration.
+            primary_kv_view: Memoryview of the primary tier's CPU KV cache.
+            tier_type: Tier type identifier, set by SecondaryTierFactory.
+            store_config: Object store connection parameters (see ObjStoreConfig).
+            prefix: Key prefix prepended to all object keys.
+            io_threads: Number of NIXL I/O threads.
+            enable_kv_events: Emit BlockStored KV events for blocks
+                successfully stored to this tier. Effective only when KV
+                cache events are enabled globally (kv_events_config).
+        """
         super().__init__(offloading_spec, primary_kv_view, tier_type)
+
+        self.events: list[OffloadingEvent] | None = None
+        if enable_kv_events:
+            if offloading_spec.kv_events_config.enable_kv_cache_events:
+                self.events = []
+            else:
+                logger.warning(
+                    "enable_kv_events is set on secondary tier '%s' but KV "
+                    "cache events are disabled globally; the tier will not "
+                    "emit events.",
+                    tier_type,
+                )
+        # Keys of in-flight store jobs, tracked only when events are enabled.
+        self._store_job_keys: dict[JobId, list[OffloadKey]] = {}
+
         agent_config = nixl_agent_config(backends=[])
         self._agent = nixl_agent("ObjAgent", agent_config)
         obj_config = ObjStoreConfig(**store_config)
@@ -231,6 +268,8 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
         return LookupResult.HIT if result else LookupResult.MISS
 
     def submit_store(self, job_metadata: JobMetadata) -> None:
+        if self.events is not None:
+            self._store_job_keys[job_metadata.job_id] = list(job_metadata.keys)
         obj_keys = (self._file_mapper.get_file_name(k) for k in job_metadata.keys)
         self._submit_transfer(
             job_metadata.job_id, job_metadata.block_ids, obj_keys, NIXL_WRITE
@@ -279,8 +318,20 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
         self._poll_active_transfers()
         results = self._pending_results
         self._pending_results = []
+        if self.events is not None:
+            for result in results:
+                keys = self._store_job_keys.pop(result.job_id, None)
+                if result.success and keys:
+                    self.events.append(
+                        OffloadingEvent(keys=keys, medium=self.medium, removed=False)
+                    )
         return results
 
+    def take_events(self) -> Iterable[OffloadingEvent]:
+        if self.events is not None:
+            yield from self.events
+            self.events.clear()
+
     def drain_jobs(self) -> None:
         """Block until every submitted transfer has completed or failed.