feat(mp): add chunk-store server plugin

docs/design/v1/multiprocess/modules/chunk_store.md

@@ -0,0 +1,37 @@
+# Chunk-store plugin
+
+The plugin factory in `chunk_store.py` is loaded through `--server-module` from
+the pluggable MP server interface. It requires the existing
+`LMCacheDrivenTransferModule` and shares its worker registrations and streams.
+It owns no additional device resources. The ordinary store RPC is unchanged.
+
+`store_with_chunk_events(key, instance_id, block_ids, producer_event_handle)`
+returns `(terminal_handle, [(chunk_handle, start, end), ...], success)`.
+The typed `RequestClient` method and `ChunkStoreService` protobuf register the
+contract on the existing ZMQ/gRPC transport; the dynamically loaded module
+provides the handler. No legacy integer operation ID is added.
+
+The plugin validates the complete range and per-kernel-group block counts before
+submitting work. It splits raw block IDs at each group's chunk stride and calls
+the existing store once per chunk, preserving the full token prefix, salt and
+request metadata. Downsampling, null blocks, storage reservation, observability
+and commits remain the ordinary transfer module's responsibility.
+
+Each returned event follows all object-group copies for its chunk. The terminal
+handle is the last submitted chunk's event. Completion permits source-buffer
+reuse; publication and cache residency follow ordinary store semantics. A failed
+chunk stops further submission and retains any available completion handle.
+Earlier successful chunks may already be committed; the batch is not atomic.
+An empty, valid range succeeds without device work.
+
+Event handles use the existing backend's retention contract. In particular, the
+default backend retains a bounded ring of exports; arbitrary delayed imports
+after that ring turns over are not guaranteed. This plugin does not add event
+leases or change backend lifetime behavior. Clients must obey the same handle
+usage constraints as the ordinary store path.
+
+Tests cover actual factory loading, opt-in registration, group geometry,
+metadata preservation, rejection before submission, partial failure, both RPC
+transports and cross-process GPU source reuse followed by exact KV retrieval.
+The implementation issues one ordinary store per chunk, so throughput and CPU
+submission overhead need measurement before making performance claims.

docs/source/mp/chunk_store.rst

@@ -0,0 +1,42 @@
+Chunk-store plugin
+==================
+
+The chunk-store plugin exposes per-chunk source-buffer completion through the
+existing MP request transport. Enable it on an LMCache-driven server:
+
+.. code-block:: bash
+
+   python -m lmcache.v1.multiprocess.server \
+     --l1-size-gb 1 --l1-init-size-gb 1 --eviction-policy LRU \
+     --server-module '{"module_path":"lmcache.v1.multiprocess.modules.chunk_store"}'
+
+Add ``--transport grpc`` to use gRPC. The module lives in the LMCache package and
+uses the pluggable server-module loader; no separate plugin package is required.
+
+After registering KV caches through ``register_kv_cache``, an explicit RPC caller
+can submit a range of complete chunks:
+
+.. code-block:: python
+
+   terminal, chunks, success = client.store_with_chunk_events(
+       key, instance_id, block_ids, producer_event_handle
+   ).result(timeout=10)
+   pending = [
+       (backend.import_event(handle, device), start, end)
+       for handle, start, end in chunks
+   ]
+   for event, start, end in pending:
+       backend.synchronize_event(event, device)
+       # Source blocks for [start, end) can now be reused.
+
+For nonblocking completion checks, use ``backend.query_event(event)``. Ranges are
+absolute and end-exclusive. Keep the producer event alive until submitted work
+completes. Returned handles follow the ordinary store's backend lifetime
+constraints, including its bounded retention of exported events; this plugin
+does not provide a lease for arbitrarily delayed imports.
+
+``success`` means all chunk stores avoided a fatal error, not that every chunk
+was cached. A failure stops later chunks; earlier chunks may already be stored.
+The terminal handle covers submitted work and is empty if none was submitted.
+Source completion does not guarantee publication has finished. Engine adapters
+continue to use ordinary store unless the caller explicitly uses this RPC.

docs/source/mp/index.rst

@@ -5,6 +5,7 @@ Overview
    :hidden:
 
    request_transport
+   chunk_store
 
 LMCache multiprocess (MP) mode runs LMCache as a **standalone service** that
 vLLM instances reach through a configurable ZMQ or gRPC request transport.

lmcache/v1/multiprocess/modules/chunk_store.py

@@ -0,0 +1,143 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Dynamically loaded chunk-store RPC using the existing transfer module."""
+
+# Standard
+from dataclasses import replace
+
+# First Party
+from lmcache.v1.multiprocess.custom_types import IPCCacheServerKey
+from lmcache.v1.multiprocess.engine_context import MPCacheServerContext
+from lmcache.v1.multiprocess.modules.lmcache_driven_transfer import (
+    LMCacheDrivenTransferModule,
+)
+from lmcache.v1.multiprocess.request_handler import HandlerType, request_handler
+from lmcache.v1.multiprocess.server_module import ServerModuleBuildContext
+
+
+def build_server_modules(ctx: ServerModuleBuildContext) -> "ChunkStoreModule":
+    """Build the chunk-store plugin against the server's registered KV contexts.
+
+    Args:
+        ctx: Plugin factory context supplied by ``--server-module`` loading.
+
+    Returns:
+        A module exposing the chunk-store RPC on the existing request transport.
+
+    Raises:
+        ValueError: The server has no LMCache-driven transfer module.
+    """
+    for module in ctx.modules:
+        if isinstance(module, LMCacheDrivenTransferModule):
+            return ChunkStoreModule(ctx.server_context, module)
+    raise ValueError("The chunk-store plugin requires LMCache-driven transfer")
+
+
+class ChunkStoreModule:
+    """Store chunks independently and report their source-buffer completion.
+
+    The plugin shares registrations, transfer streams, storage and event backend
+    with the ordinary store path. Each chunk uses that path's commit and error
+    semantics; a later failure does not roll back earlier chunks.
+
+    Args:
+        ctx: Shared server context providing chunk geometry.
+        transfer_module: Existing owner of registered KV caches and transfers.
+    """
+
+    def __init__(
+        self, ctx: MPCacheServerContext, transfer_module: LMCacheDrivenTransferModule
+    ) -> None:
+        self._ctx = ctx
+        self._transfer = transfer_module
+
+    @property
+    def context(self) -> MPCacheServerContext:
+        """Return the shared server context."""
+        return self._ctx
+
+    def report_status(self) -> dict[str, dict[str, bool]]:
+        """Report that the chunk-store plugin is loaded."""
+        return {"chunk_store": {"is_healthy": True}}
+
+    def close(self) -> None:
+        """Leave shared registrations and streams to their transfer-module owner."""
+        return None
+
+    @request_handler(HandlerType.BLOCKING, requires_client_affinity=True)
+    def store_with_chunk_events(
+        self,
+        key: IPCCacheServerKey,
+        instance_id: int,
+        block_ids: list[list[int]],
+        event_ipc_handle: bytes,
+    ) -> tuple[bytes, list[tuple[bytes, int, int]], bool]:
+        """Submit a chunk-aligned range and return one completion per chunk.
+
+        Args:
+            key: Worker key with the full token prefix and an absolute range
+                consisting of complete LMCache chunks.
+            instance_id: Worker ID registered through ``register_kv_cache``.
+            block_ids: Raw block IDs for the range, in kernel-group order.
+            event_ipc_handle: Producer event ordering reads of source KV blocks.
+
+        Returns:
+            Last submitted completion handle, ordered ``(handle, start, end)``
+            entries, and whether every chunk store avoided a fatal error.
+            Ranges are absolute and end-exclusive. Events permit source reuse;
+            they do not guarantee cache residency or completed publication.
+            Invalid ranges, block counts and unregistered workers return empty
+            handles and False before submission. An empty valid range succeeds.
+            A failed chunk stops submission, preserving any earlier completions
+            and its own completion handle if device work was submitted.
+
+        Raises:
+            RuntimeError: The registered backend cannot provide IPC events.
+
+        Notes:
+            Handles follow the ordinary store's event-backend lifetime contract.
+            No host/device synchronization is added during submission.
+        """
+        chunk_size = self._ctx.chunk_size
+        if (
+            key.worker_id is None
+            or key.start < 0
+            or key.end < key.start
+            or key.end > len(key.token_ids)
+            or key.start % chunk_size
+            or key.end % chunk_size
+        ):
+            return b"", [], False
+        entry = self._transfer.get_and_touch_context_entry(instance_id)
+        if entry is None:
+            return b"", [], False
+        cache = entry.cache_context
+        strides = [
+            cache.calculate_num_blocks(chunk_size, group)
+            for group in range(cache.kv_layer_groups_manager.num_kernel_groups)
+        ]
+        num_chunks = (key.end - key.start) // chunk_size
+        if len(block_ids) != len(strides) or any(
+            len(ids) != num_chunks * stride
+            for ids, stride in zip(block_ids, strides, strict=True)
+        ):
+            return b"", [], False
+
+        terminal = b""
+        completions: list[tuple[bytes, int, int]] = []
+        for chunk, start in enumerate(range(key.start, key.end, chunk_size)):
+            end = start + chunk_size
+            handle, success = self._transfer.store(
+                replace(key, start=start, end=end),
+                instance_id,
+                [
+                    ids[chunk * stride : (chunk + 1) * stride]
+                    for ids, stride in zip(block_ids, strides, strict=True)
+                ],
+                event_ipc_handle,
+            )
+            if handle:
+                terminal = handle
+                completions.append((handle, start, end))
+            if not success:
+                return terminal, completions, False
+        return terminal, completions, True

lmcache/v1/multiprocess/transport/base.py

@@ -92,6 +92,31 @@ def store(
         event_ipc_handle: bytes,
     ) -> MessagingFuture[tuple[bytes, bool]]: ...
 
+    @rpc_method
+    def store_with_chunk_events(
+        self,
+        key: IPCCacheServerKey,
+        instance_id: int,
+        block_ids: list[list[int]],
+        event_ipc_handle: bytes,
+    ) -> MessagingFuture[tuple[bytes, list[tuple[bytes, int, int]], bool]]:
+        """Submit complete chunks to the configured chunk-store server plugin.
+
+        Args:
+            key: Worker key describing an absolute, chunk-aligned token range.
+            instance_id: Registered worker ID.
+            block_ids: Raw source blocks for the range, in kernel-group order.
+            event_ipc_handle: Producer event ordering reads of the source KV.
+
+        Returns:
+            A future resolving to the terminal handle, per-chunk
+            ``(handle, start, end)`` completions, and success. Completion permits
+            source reuse, not guaranteed cache residency. Handles follow the
+            same backend lifetime contract as ``store``. Requires the server
+            module ``lmcache.v1.multiprocess.modules.chunk_store`` to be loaded.
+        """
+        ...
+
     @rpc_method
     def retrieve(
         self,

lmcache/v1/multiprocess/transport/grpc_impl/protos/chunk_store_service.proto

@@ -0,0 +1,22 @@
+// SPDX-License-Identifier: Apache-2.0
+syntax = "proto3";
+
+package lmcache.mp;
+
+import "lmcache_driven_service.proto";
+
+message ChunkCompletion {
+  bytes event_ipc_handle = 1;
+  int64 start = 2;
+  int64 end = 3;
+}
+
+message ChunkStoreResponse {
+  bytes terminal_event_ipc_handle = 1;
+  repeated ChunkCompletion chunk_events = 2;
+  bool success = 3;
+}
+
+service ChunkStoreService {
+  rpc StoreWithChunkEvents(StoreRequest) returns (ChunkStoreResponse);
+}