[None][fix] Require physical quiescence before retiring C++ bounce KV

cpp/include/tensorrt_llm/executor/transferAgent.h

@@ -393,6 +393,19 @@ class TransferStatus
     {
         return {};
     }
+
+    /// True only after no future source read or destination write is possible.
+    /// Logical failure, cancellation and handle release are not this proof.
+    [[nodiscard]] virtual bool isQuiesced() const
+    {
+        return false;
+    }
+
+    /// Request cancellation without waiting; acceptance is not a quiescence proof.
+    [[nodiscard]] virtual bool requestCancel()
+    {
+        return false;
+    }
 };
 
 struct BaseAgentConfig

cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp

@@ -188,6 +188,8 @@ NB_MODULE(tensorrt_llm_transfer_agent_binding, m)
     // the GIL here is safe and necessary for correct behavior when the concrete
     // subclass type is not directly registered (e.g., agents created via factory).
     nb::class_<kvc::TransferStatus>(m, "TransferStatus")
+        .def("is_quiesced", &kvc::TransferStatus::isQuiesced, nb::call_guard<nb::gil_scoped_release>())
+        .def("request_cancel", &kvc::TransferStatus::requestCancel, nb::call_guard<nb::gil_scoped_release>())
         .def("is_completed", &kvc::TransferStatus::isCompleted, nb::call_guard<nb::gil_scoped_release>())
         .def("wait", &kvc::TransferStatus::wait, nb::arg("timeout_ms") = -1, nb::call_guard<nb::gil_scoped_release>())
         // Failure detail for the last terminal state (empty if unavailable). Named to match the

cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/bounce/BounceConfig.h

@@ -130,7 +130,7 @@ namespace tensorrt_llm::executor::kv_cache::bounce
     return parsed * mult;
 }
 
-/// POD config for the bounce v2 pipeline. There is no `enabled` field: the on/off switch is
+/// POD config for the bounce pipeline. There is no `enabled` field: the on/off switch is
 /// CacheTransceiverConfig's agent_bounce_buffer_enable + kv_cache_bounce_size_mb (which the Python
 /// frontend folds into the agent's arena size in MiB: 0 = off, >0 = arena size), and at runtime
 /// "bounce on" simply means the owning agent built its bounce state (mBounce != nullptr).
@@ -163,24 +163,22 @@ struct BounceConfig
     // means no minimum.
     std::size_t minDescriptorCount{1024};                   // min_descriptor_count
     std::size_t maxAverageDescriptorSizeBytes{16ULL << 10}; // max_average_descriptor_size
-    int requestTimeoutMs{30000}; // request_timeout_ms; must be > 0 — the whole failure model
-                                 // (abandoned-flow resolution, receiver lease, quarantine) hangs off
-                                 // this timer, so applyParam rejects 0 (negatives already fail the
-                                 // strict unsigned parse) and keeps the current value.
+    int requestTimeoutMs{30000}; // request_timeout_ms; must be > 0 for logical failure detection
+                                 // and receiver lease fencing. Expiry initiates physical drain; it
+                                 // cannot itself establish quiescence or permit memory reuse.
     // Receiver-side lease on granted regions — DERIVED in deriveDependentTimeouts() as
     // 2 x requestTimeoutMs, not an independent knob (tests may still set the field directly). A
     // dead sender emits neither DATA nor a cancel, which is unobservable through the protocol
     // alone — so a flow whose grants see no progress (no GRANT sent, no DATA received) for this
-    // long is reclaimed and its regions quarantined (below) before reuse. The lease must EXCEED
+    // long is fenced and its regions retained until explicit physical drain. The lease must EXCEED
     // the peers' requestTimeoutMs (a live sender abandons + cancels first, so only
     // dead/unreachable peers ever hit this) — the 2x derivation relies on both ends running the
     // SAME request_timeout_ms, which the capability handshake enforces (strict equality; a
     // mismatched peer falls back to standard NIXL).
     int receiverFlowTimeoutMs{60000};
-    // How long a receiver-reclaimed, possibly-still-being-written region stays out of the arena
-    // before reuse — DERIVED in deriveDependentTimeouts() as requestTimeoutMs (equal to the peer's
-    // by the handshake). A one-sided RDMA write cannot be aborted, so time is the only barrier
-    // against re-granting a region a gone peer's NIC may still be writing.
+    // Legacy scheduler quarantine interval, retained for direct scheduler callers/tests and sweep
+    // cadence. The v3 transport uses permanent per-flow quarantine until DRAIN proves that all
+    // remote writes stopped; this interval never permits release of those regions.
     int quarantineMs{30000};
     bool disableFabricMemory{false}; // disable_fabric_memory
     // enable_eager_gather: launch a chunk's gather at submit() time, before the receiver's GRANT
@@ -195,8 +193,8 @@ struct BounceConfig
     bool useZeroCopyArguments{true};
 
     /// Re-derive the lease/quarantine values from the one user-visible timeout (see the field
-    /// comments): the lease must exceed the peers' request timeout, and time is the only write
-    /// barrier for a reclaimed region. The config layer guarantees requestTimeoutMs > 0 (applyParam
+    /// comments): the lease exceeds the peers' request timeout but is only a fencing deadline,
+    /// never a physical completion barrier. The config layer guarantees requestTimeoutMs > 0 (applyParam
     /// rejects 0); the > 0 guard below only matters for directly-constructed configs. Must be
     /// re-run whenever requestTimeoutMs changes: fromEnv() always calls it, fromParams() only when
     /// the dict actually provides request_timeout_ms — so a caller-tweaked receiverFlowTimeoutMs /

cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/bounce/BounceMessage.cpp

@@ -118,6 +118,21 @@ std::string encodeCancel(std::uint64_t requestId, std::string const& endpoint)
     return encodeWant(requestId, {}, endpoint);
 }
 
+std::string encodeDrain(std::uint64_t requestId, std::string const& endpoint)
+{
+    auto blob = encodeWant(requestId, {}, endpoint);
+    BounceMsgHeader header{};
+    std::memcpy(&header, blob.data(), sizeof(header));
+    header.msgType = static_cast<std::uint16_t>(BounceMsgType::kDRAIN);
+    std::memcpy(blob.data(), &header, sizeof(header));
+    return blob;
+}
+
+std::string encodeDrainAck(std::uint64_t requestId)
+{
+    return encodeHeaderOnly(makeHeader(BounceMsgType::kDRAIN_ACK, requestId, 0, 0, 0, 0, 0));
+}
+
 std::string encodeGrant(std::uint64_t requestId, std::vector<BounceCreditEntry> const& credits)
 {
     auto const bytes = static_cast<std::uint32_t>(credits.size() * sizeof(BounceCreditEntry));

cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/bounce/BounceMessage.h

@@ -24,9 +24,9 @@
 namespace tensorrt_llm::executor::kv_cache::bounce
 {
 
-/// Bounce v2 control-plane wire format. Carried over the ControlChannel (zmq by default).
+/// Bounce v3 control-plane wire format. Carried over the ControlChannel (zmq by default).
 /// Pure encode/decode — no IO — so it is fully unit-testable. NOTE: this is NOT the NIXL
-/// notifMsg (v2 does not use notifMsg at all); it is our own message set.
+/// notifMsg (bounce does not use notifMsg); it is our own message set.
 ///
 /// ENDIANNESS: fields are memcpy'd raw (host byte order), so this assumes both peers are
 /// little-endian — true for all NVIDIA GPU hosts (x86_64 / aarch64 LE). It is NOT a portable
@@ -35,11 +35,13 @@ namespace tensorrt_llm::executor::kv_cache::bounce
 /// treats any unknown `msgType` as a no-op (default branch), so a bogus type can't misroute.
 enum class BounceMsgType : std::uint16_t
 {
-    kWANT = 1,  // sender -> receiver: per-chunk byte sizes[] for requestId (empty list cancels)
-    kGRANT = 2, // receiver -> sender: credits[] for requestId
-    kDATA = 3,  // sender -> receiver: region written; scatter entries[] (after getXferStatus SUCCESS)
-    kACK = 4,   // receiver -> sender: chunk scattered; region freed
-    kNACK = 5,  // receiver -> sender: chunk/request cannot be completed; sender fails the request
+    kWANT = 1,      // sender -> receiver: per-chunk byte sizes[] for requestId (empty list cancels)
+    kGRANT = 2,     // receiver -> sender: credits[] for requestId
+    kDATA = 3,      // sender -> receiver: region written; scatter entries[] (after getXferStatus SUCCESS)
+    kACK = 4,       // receiver -> sender: chunk scattered; region freed
+    kDRAIN = 6,     // sender -> receiver: no more RDMA; fence and drain every scatter
+    kDRAIN_ACK = 7, // receiver -> sender: flow fenced and all scatter physically complete
+    kNACK = 5,      // receiver -> sender: chunk/request cannot be completed; sender fails the request
 };
 
 #pragma pack(push, 1)
@@ -51,7 +53,7 @@ struct BounceMsgHeader
     std::uint32_t magic;     // kMagic
     std::uint16_t version;   // kBounceVersion
     std::uint16_t msgType;   // BounceMsgType
-    std::uint64_t requestId; // WANT/GRANT/DATA/ACK
+    std::uint64_t requestId; // all control messages
     std::uint64_t
         regionHandle; // DATA/ACK: the arena region offset of this chunk; 0 elsewhere (64-bit -> arena may exceed 4 GiB)
     std::uint32_t chunkIdx;     // DATA/ACK
@@ -97,10 +99,10 @@ static_assert(sizeof(BounceCreditEntry) == 24, "BounceCreditEntry must be 24 byt
 static_assert(sizeof(BounceScatterRun) == 36, "BounceScatterRun must be 36 bytes");
 
 inline constexpr std::uint32_t kBounceMagic = 0x424E4332U; // 'B''N''C''2'
-// v2: DATA scatter entries became strided RUNS (BounceScatterRun). The out-of-band capability
+// v3: explicit DRAIN/DRAIN_ACK proves physical quiescence after logical failure. The out-of-band capability
 // handshake checks this version before bounce is selected; decodeHeader also rejects a mismatched
 // control message defensively.
-inline constexpr std::uint16_t kBounceVersion = 2U;
+inline constexpr std::uint16_t kBounceVersion = 3U;
 
 // ---- encode (each returns a self-contained blob: header + payload) ----
 /// WANT carries the per-chunk byte sizes the sender will write (the receiver allocates a region of
@@ -112,12 +114,14 @@ inline constexpr std::uint16_t kBounceVersion = 2U;
 /// decode via decodeWant.
 [[nodiscard]] std::string encodeWant(
     std::uint64_t requestId, std::vector<std::uint32_t> const& chunkBytes, std::string const& endpoint);
-/// Cancel/retract a request: a WANT with an EMPTY chunk list (the receiver frees everything it
-/// allocated/held for `requestId`). This is the ONLY meaning of an empty WANT — submit() never sends
-/// a zero-chunk WANT (a 0-chunk transfer resolves SUCCESS without any WANT) — so an empty chunk list
-/// is unambiguously a cancel. Thin named wrapper over encodeWant to make the intent explicit at call
-/// sites; the wire form is identical (still a WANT), so it reuses the receiver's onWant/reclaim path.
+/// Cancel/retract a request with an EMPTY WANT. This fences future WANT/DATA for the flow but
+/// supplies no physical completion evidence: unknown writes remain quarantined and queued scatters
+/// remain owned until completion. A zero-chunk transfer succeeds without sending WANT. Use DRAIN
+/// only after every write into the peer arena is proved stopped; its acknowledgment proves that
+/// the receiver has fenced the flow and physically completed all admitted scatters.
 [[nodiscard]] std::string encodeCancel(std::uint64_t requestId, std::string const& endpoint);
+[[nodiscard]] std::string encodeDrain(std::uint64_t requestId, std::string const& endpoint);
+[[nodiscard]] std::string encodeDrainAck(std::uint64_t requestId);
 [[nodiscard]] std::string encodeGrant(std::uint64_t requestId, std::vector<BounceCreditEntry> const& credits);
 [[nodiscard]] std::string encodeData(std::uint64_t requestId, std::uint32_t chunkIdx, std::uint32_t numChunks,
     std::uint64_t regionHandle, std::vector<BounceScatterRun> const& entries);
@@ -184,10 +188,9 @@ struct BounceHandshake
     // This agent's effective per-chunk cap after the arena-capacity clamp. Peers compare this field
     // exactly, which also keeps their scatter-plan capacities consistent.
     std::uint64_t maxChunkSizeBytes{0};
-    // This agent's request_timeout_ms. Peers compare it exactly: the receiver's region lease (2x its
-    // value) must exceed the sender's timeout. The quarantine (1x) starts at lease expiry OR at a
-    // receiver-side peer teardown (forget()/invalidateRemoteAgent); on the lease path it needs no
-    // extra margin of its own.
+    // This agent's request_timeout_ms. Peers compare it exactly so the receiver lease (2x this
+    // value) gives live senders time to initiate drain. Lease expiry fences a flow; it never proves
+    // that a remote write stopped and cannot release quarantined storage.
     std::int32_t requestTimeoutMs{0};
     // Control-channel address: a zmq endpoint (kZMQ).
     std::string endpoint;