From 74ba94a1df3208c84a3f33cd1a20f8a95711fae3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Sat, 11 Jul 2026 21:55:38 +0800 Subject: [PATCH 1/5] tent: add receiver credit ledger model --- .../include/tent/runtime/receiver_credit.h | 91 ++++++++++ .../tent/src/runtime/receiver_credit.cpp | 152 ++++++++++++++++ .../tent/tests/CMakeLists.txt | 7 + .../tent/tests/receiver_credit_test.cpp | 168 ++++++++++++++++++ 4 files changed, 418 insertions(+) create mode 100644 mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h create mode 100644 mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp create mode 100644 mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp diff --git a/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h b/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h new file mode 100644 index 0000000000..bc84721292 --- /dev/null +++ b/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h @@ -0,0 +1,91 @@ +// Copyright 2026 KVCache.AI +// SPDX-License-Identifier: Apache-2.0 + +#ifndef TENT_RUNTIME_RECEIVER_CREDIT_H +#define TENT_RUNTIME_RECEIVER_CREDIT_H + +#include +#include +#include +#include +#include +#include + +#include "tent/common/status.h" + +namespace mooncake::tent { + +struct ReceiverSessionId { + uint64_t high{0}, low{0}; + bool operator==(const ReceiverSessionId& o) const { + return high == o.high && low == o.low; + } +}; +struct CreditKey { + ReceiverSessionId receiver_session; + uint64_t sender_peer{0}; + uint32_t qos_class{0}; + bool operator==(const CreditKey& o) const { + return receiver_session == o.receiver_session && + sender_peer == o.sender_peer && qos_class == o.qos_class; + } +}; +struct CreditKeyHash { + size_t operator()(const CreditKey&) const noexcept; +}; +enum class CreditResource : uint16_t { + DataBytes = 1, + RequestSlots, + StagingSlots, + ConsumerSlots +}; +constexpr size_t kCreditResourceCount = 4; +struct CreditAmount { + CreditResource resource; + uint64_t grant_total{0}; +}; +struct CreditCharge { + std::vector> resources; +}; +struct ReceiverCreditUpdateV1 { + uint16_t schema_version{1}, flags{0}; + uint32_t qos_class{0}; + ReceiverSessionId receiver_session_id; + uint64_t epoch{0}, sequence{0}; + uint32_t freshness_ttl_ms{0}; + std::vector grants; +}; +enum class CreditUpdateDisposition : uint8_t { + Applied, + DuplicateOrOld, + SequenceGap +}; + +// Private, sender-side state model. It has no network or Admission integration. +class SenderCreditLedger { + public: + Status activate(const CreditKey&, uint64_t epoch); + Status applyUpdate(const CreditKey&, const ReceiverCreditUpdateV1&, + CreditUpdateDisposition&); + Status tryReserve(const CreditKey&, const CreditCharge&); + // Only for work not yet handed to a transport; completions need a new + // grant. + Status rollbackReservation(const CreditKey&, const CreditCharge&); + Status available(const CreditKey&, CreditResource, uint64_t&) const; + Status consumed(const CreditKey&, CreditResource, uint64_t&) const; + + private: + struct Entry { + uint64_t epoch{0}, last_sequence{0}; + bool has_update{false}; + std::array grants{}, consumed{}; + }; + static Status resourceIndex(CreditResource, size_t&); + static Status normalize(const CreditCharge&, + std::array&); + mutable std::mutex mutex_; + std::unordered_map entries_; +}; + +} // namespace mooncake::tent +#endif diff --git a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp new file mode 100644 index 0000000000..33a48b0897 --- /dev/null +++ b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp @@ -0,0 +1,152 @@ +// Copyright 2026 KVCache.AI +// SPDX-License-Identifier: Apache-2.0 + +#include "tent/runtime/receiver_credit.h" + +namespace mooncake::tent { + +size_t CreditKeyHash::operator()(const CreditKey& k) const noexcept { + size_t h = std::hash{}(k.receiver_session.high); + auto mix = [&h](uint64_t v) { + h ^= std::hash{}(v) + 0x9e3779b97f4a7c15ULL + (h << 6) + + (h >> 2); + }; + mix(k.receiver_session.low); + mix(k.sender_peer); + mix(k.qos_class); + return h; +} + +Status SenderCreditLedger::resourceIndex(CreditResource r, size_t& i) { + auto raw = static_cast(r); + if (raw < 1 || raw > kCreditResourceCount) + return Status::InvalidArgument("unknown credit resource" LOC_MARK); + i = raw - 1; + return Status::OK(); +} + +Status SenderCreditLedger::normalize( + const CreditCharge& c, std::array& out) { + out.fill(0); + if (c.resources.empty()) + return Status::InvalidArgument("empty credit charge" LOC_MARK); + for (auto [r, amount] : c.resources) { + size_t i; + CHECK_STATUS(resourceIndex(r, i)); + if (!amount || out[i]) + return Status::InvalidArgument( + "zero or duplicate credit charge" LOC_MARK); + out[i] = amount; + } + return Status::OK(); +} + +Status SenderCreditLedger::activate(const CreditKey& k, uint64_t epoch) { + if (!epoch) return Status::InvalidArgument("zero credit epoch" LOC_MARK); + std::lock_guard lock(mutex_); + Entry e; + e.epoch = epoch; + entries_.insert_or_assign(k, e); // fences all prior-epoch unused state + return Status::OK(); +} + +Status SenderCreditLedger::applyUpdate(const CreditKey& k, + const ReceiverCreditUpdateV1& u, + CreditUpdateDisposition& disposition) { + if (u.schema_version != 1 || !u.epoch || !u.sequence) + return Status::InvalidArgument("invalid credit update header" LOC_MARK); + if (!(u.receiver_session_id == k.receiver_session) || + u.qos_class != k.qos_class || u.grants.size() > kCreditResourceCount) + return Status::InvalidArgument("credit update identity/size" LOC_MARK); + std::array proposed{}; + std::array present{}; + for (auto a : u.grants) { + size_t i; + CHECK_STATUS(resourceIndex(a.resource, i)); + if (present[i]) + return Status::InvalidArgument("duplicate grant resource" LOC_MARK); + present[i] = true; + proposed[i] = a.grant_total; + } + std::lock_guard lock(mutex_); + auto it = entries_.find(k); + if (it == entries_.end() || it->second.epoch != u.epoch) + return Status::InvalidEntry("inactive or stale credit epoch" LOC_MARK); + auto& e = it->second; + if (e.has_update && u.sequence <= e.last_sequence) { + disposition = CreditUpdateDisposition::DuplicateOrOld; + return Status::OK(); + } + for (size_t i = 0; i < kCreditResourceCount; ++i) + if (present[i] && + (proposed[i] < e.grants[i] || proposed[i] < e.consumed[i])) + return Status::InvalidArgument( + "decreasing or under-consumed grant" LOC_MARK); + bool gap = e.has_update && u.sequence > e.last_sequence + 1; + for (size_t i = 0; i < kCreditResourceCount; ++i) + if (present[i]) e.grants[i] = proposed[i]; + e.last_sequence = u.sequence; + e.has_update = true; + disposition = gap ? CreditUpdateDisposition::SequenceGap + : CreditUpdateDisposition::Applied; + return Status::OK(); +} + +Status SenderCreditLedger::tryReserve(const CreditKey& k, + const CreditCharge& c) { + std::array n; + CHECK_STATUS(normalize(c, n)); + std::lock_guard lock(mutex_); + auto it = entries_.find(k); + if (it == entries_.end() || !it->second.has_update) + return Status::InvalidEntry("credit unavailable" LOC_MARK); + auto& e = it->second; + for (size_t i = 0; i < kCreditResourceCount; ++i) + if (e.consumed[i] > e.grants[i] || n[i] > e.grants[i] - e.consumed[i]) + return Status::TooManyRequests("insufficient credit" LOC_MARK); + for (size_t i = 0; i < kCreditResourceCount; ++i) e.consumed[i] += n[i]; + return Status::OK(); +} + +Status SenderCreditLedger::rollbackReservation(const CreditKey& k, + const CreditCharge& c) { + std::array n; + CHECK_STATUS(normalize(c, n)); + std::lock_guard lock(mutex_); + auto it = entries_.find(k); + if (it == entries_.end()) + return Status::InvalidEntry("credit session inactive" LOC_MARK); + for (size_t i = 0; i < kCreditResourceCount; ++i) + if (n[i] > it->second.consumed[i]) + return Status::InvalidArgument( + "credit rollback underflow" LOC_MARK); + for (size_t i = 0; i < kCreditResourceCount; ++i) + it->second.consumed[i] -= n[i]; + return Status::OK(); +} + +Status SenderCreditLedger::available(const CreditKey& k, CreditResource r, + uint64_t& v) const { + size_t i; + CHECK_STATUS(resourceIndex(r, i)); + std::lock_guard lock(mutex_); + auto it = entries_.find(k); + if (it == entries_.end() || !it->second.has_update) + return Status::InvalidEntry("credit unavailable" LOC_MARK); + v = it->second.grants[i] - it->second.consumed[i]; + return Status::OK(); +} + +Status SenderCreditLedger::consumed(const CreditKey& k, CreditResource r, + uint64_t& v) const { + size_t i; + CHECK_STATUS(resourceIndex(r, i)); + std::lock_guard lock(mutex_); + auto it = entries_.find(k); + if (it == entries_.end()) + return Status::InvalidEntry("credit session inactive" LOC_MARK); + v = it->second.consumed[i]; + return Status::OK(); +} + +} // namespace mooncake::tent diff --git a/mooncake-transfer-engine/tent/tests/CMakeLists.txt b/mooncake-transfer-engine/tent/tests/CMakeLists.txt index d04805f1ae..8838f8da25 100644 --- a/mooncake-transfer-engine/tent/tests/CMakeLists.txt +++ b/mooncake-transfer-engine/tent/tests/CMakeLists.txt @@ -27,6 +27,13 @@ target_include_directories(admission_queue_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../include) add_test(NAME admission_queue_test COMMAND admission_queue_test) +add_executable(receiver_credit_test receiver_credit_test.cpp + ../src/runtime/receiver_credit.cpp) +target_link_libraries(receiver_credit_test PRIVATE tent_common gtest gtest_main) +target_include_directories(receiver_credit_test + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../include) +add_test(NAME receiver_credit_test COMMAND receiver_credit_test) + # Reproducible hot-path microbenchmark; intentionally not registered with # ctest. Run manually when changing deadline promotion partitioning. add_executable(deadline_promotion_bench deadline_promotion_bench.cpp diff --git a/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp b/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp new file mode 100644 index 0000000000..a328d61601 --- /dev/null +++ b/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp @@ -0,0 +1,168 @@ +// Copyright 2026 KVCache.AI +// SPDX-License-Identifier: Apache-2.0 + +#include "tent/runtime/receiver_credit.h" + +#include +#include + +#include + +namespace mooncake::tent { +namespace { +CreditKey key() { return {{1, 2}, 3, 4}; } +ReceiverCreditUpdateV1 update(uint64_t epoch, uint64_t seq, + std::vector grants) { + ReceiverCreditUpdateV1 u; + u.receiver_session_id = key().receiver_session; + u.qos_class = key().qos_class; + u.epoch = epoch; + u.sequence = seq; + u.grants = std::move(grants); + return u; +} +CreditCharge charge(uint64_t bytes, uint64_t slots) { + return {{{CreditResource::DataBytes, bytes}, + {CreditResource::RequestSlots, slots}}}; +} +void grant(SenderCreditLedger& l, uint64_t seq, uint64_t bytes = 100, + uint64_t slots = 2) { + CreditUpdateDisposition d; + ASSERT_TRUE(l.applyUpdate(key(), + update(7, seq, + {{CreditResource::DataBytes, bytes}, + {CreditResource::RequestSlots, slots}}), + d) + .ok()); +} + +TEST(ReceiverCredit, MultiResourceReserveIsAtomic) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1); + ASSERT_TRUE(l.tryReserve(key(), charge(60, 1)).ok()); + EXPECT_TRUE(l.tryReserve(key(), charge(41, 1)).IsTooManyRequests()); + uint64_t v; + ASSERT_TRUE(l.available(key(), CreditResource::RequestSlots, v).ok()); + EXPECT_EQ(v, 1); // failed byte reservation did not consume a slot +} + +TEST(ReceiverCredit, DuplicateAndReorderedUpdatesCannotMintCredit) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 2); + CreditUpdateDisposition d; + ASSERT_TRUE(l.applyUpdate( + key(), update(7, 2, {{CreditResource::DataBytes, 999}}), d) + .ok()); + EXPECT_EQ(d, CreditUpdateDisposition::DuplicateOrOld); + ASSERT_TRUE(l.applyUpdate( + key(), update(7, 1, {{CreditResource::DataBytes, 999}}), d) + .ok()); + uint64_t v; + ASSERT_TRUE(l.available(key(), CreditResource::DataBytes, v).ok()); + EXPECT_EQ(v, 100); +} + +TEST(ReceiverCredit, SequenceGapIsVisibleAndSafe) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1); + CreditUpdateDisposition d; + ASSERT_TRUE(l.applyUpdate( + key(), update(7, 4, {{CreditResource::DataBytes, 120}}), d) + .ok()); + EXPECT_EQ(d, CreditUpdateDisposition::SequenceGap); +} + +TEST(ReceiverCredit, StaleEpochFailsAndActivationFencesOldState) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1); + ASSERT_TRUE(l.tryReserve(key(), charge(60, 1)).ok()); + CreditUpdateDisposition d; + EXPECT_TRUE(l.applyUpdate( + key(), update(6, 2, {{CreditResource::DataBytes, 999}}), d) + .IsInvalidEntry()); + ASSERT_TRUE(l.activate(key(), 8).ok()); + uint64_t v; + EXPECT_TRUE( + l.available(key(), CreditResource::DataBytes, v).IsInvalidEntry()); +} + +TEST(ReceiverCredit, InvalidUpdateDoesNotPartiallyMutate) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1); + CreditUpdateDisposition d; + EXPECT_TRUE(l.applyUpdate(key(), + update(7, 2, + {{CreditResource::DataBytes, 200}, + {CreditResource::RequestSlots, 1}}), + d) + .IsInvalidArgument()); + uint64_t v; + ASSERT_TRUE(l.available(key(), CreditResource::DataBytes, v).ok()); + EXPECT_EQ(v, 100); +} + +TEST(ReceiverCredit, DuplicateUnknownAndZeroResourcesFailClosed) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + CreditUpdateDisposition d; + EXPECT_TRUE(l.applyUpdate(key(), + update(7, 1, + {{CreditResource::DataBytes, 1}, + {CreditResource::DataBytes, 2}}), + d) + .IsInvalidArgument()); + EXPECT_TRUE(l.tryReserve(key(), {{{static_cast(99), 1}}}) + .IsInvalidArgument()); + EXPECT_TRUE(l.tryReserve(key(), {{{CreditResource::DataBytes, 0}}}) + .IsInvalidArgument()); +} + +TEST(ReceiverCredit, RollbackChecksUnderflowAtomically) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1); + ASSERT_TRUE(l.tryReserve(key(), charge(60, 1)).ok()); + EXPECT_TRUE( + l.rollbackReservation(key(), charge(61, 1)).IsInvalidArgument()); + uint64_t v; + ASSERT_TRUE(l.consumed(key(), CreditResource::RequestSlots, v).ok()); + EXPECT_EQ(v, 1); + ASSERT_TRUE(l.rollbackReservation(key(), charge(60, 1)).ok()); +} + +TEST(ReceiverCredit, GrantCannotDecreaseOrFallBelowConsumption) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1); + ASSERT_TRUE(l.tryReserve(key(), charge(80, 1)).ok()); + CreditUpdateDisposition d; + EXPECT_TRUE( + l.applyUpdate(key(), update(7, 2, {{CreditResource::DataBytes, 79}}), d) + .IsInvalidArgument()); +} + +TEST(ReceiverCredit, ConcurrentReservationsNeverExceedGrant) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1, 100, 100); + std::atomic admitted{0}; + std::vector threads; + for (int i = 0; i < 16; ++i) { + threads.emplace_back([&] { + for (int j = 0; j < 20; ++j) + if (l.tryReserve(key(), charge(1, 1)).ok()) ++admitted; + }); + } + for (auto& thread : threads) thread.join(); + EXPECT_EQ(admitted, 100); + uint64_t consumed; + ASSERT_TRUE(l.consumed(key(), CreditResource::DataBytes, consumed).ok()); + EXPECT_EQ(consumed, 100); +} +} // namespace +} // namespace mooncake::tent From 28f00f5aee5d711fa38b5a23518a0c951336c836 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Sat, 11 Jul 2026 21:58:34 +0800 Subject: [PATCH 2/5] tent: initialize credit resource indices --- .../tent/src/runtime/receiver_credit.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp index 33a48b0897..a990c208fb 100644 --- a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp @@ -31,7 +31,7 @@ Status SenderCreditLedger::normalize( if (c.resources.empty()) return Status::InvalidArgument("empty credit charge" LOC_MARK); for (auto [r, amount] : c.resources) { - size_t i; + size_t i = 0; CHECK_STATUS(resourceIndex(r, i)); if (!amount || out[i]) return Status::InvalidArgument( @@ -61,7 +61,7 @@ Status SenderCreditLedger::applyUpdate(const CreditKey& k, std::array proposed{}; std::array present{}; for (auto a : u.grants) { - size_t i; + size_t i = 0; CHECK_STATUS(resourceIndex(a.resource, i)); if (present[i]) return Status::InvalidArgument("duplicate grant resource" LOC_MARK); @@ -127,7 +127,7 @@ Status SenderCreditLedger::rollbackReservation(const CreditKey& k, Status SenderCreditLedger::available(const CreditKey& k, CreditResource r, uint64_t& v) const { - size_t i; + size_t i = 0; CHECK_STATUS(resourceIndex(r, i)); std::lock_guard lock(mutex_); auto it = entries_.find(k); @@ -139,7 +139,7 @@ Status SenderCreditLedger::available(const CreditKey& k, CreditResource r, Status SenderCreditLedger::consumed(const CreditKey& k, CreditResource r, uint64_t& v) const { - size_t i; + size_t i = 0; CHECK_STATUS(resourceIndex(r, i)); std::lock_guard lock(mutex_); auto it = entries_.find(k); From 818bac5232a8ab911068fa9656dfc7c486bd824e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Sat, 11 Jul 2026 22:04:04 +0800 Subject: [PATCH 3/5] tent: fence replayed credit activations --- .../include/tent/runtime/receiver_credit.h | 3 +++ .../tent/src/runtime/receiver_credit.cpp | 8 +++++++ .../tent/tests/receiver_credit_test.cpp | 24 +++++++++++++++++++ 3 files changed, 35 insertions(+) diff --git a/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h b/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h index bc84721292..8948526106 100644 --- a/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h +++ b/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h @@ -64,6 +64,8 @@ enum class CreditUpdateDisposition : uint8_t { // Private, sender-side state model. It has no network or Admission integration. class SenderCreditLedger { public: + explicit SenderCreditLedger(size_t max_entries = 1024) + : max_entries_(max_entries) {} Status activate(const CreditKey&, uint64_t epoch); Status applyUpdate(const CreditKey&, const ReceiverCreditUpdateV1&, CreditUpdateDisposition&); @@ -84,6 +86,7 @@ class SenderCreditLedger { static Status normalize(const CreditCharge&, std::array&); mutable std::mutex mutex_; + const size_t max_entries_; std::unordered_map entries_; }; diff --git a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp index a990c208fb..57f833662e 100644 --- a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp @@ -44,6 +44,14 @@ Status SenderCreditLedger::normalize( Status SenderCreditLedger::activate(const CreditKey& k, uint64_t epoch) { if (!epoch) return Status::InvalidArgument("zero credit epoch" LOC_MARK); std::lock_guard lock(mutex_); + auto existing = entries_.find(k); + if (existing != entries_.end()) { + if (epoch < existing->second.epoch) + return Status::InvalidEntry("stale credit activation" LOC_MARK); + if (epoch == existing->second.epoch) return Status::OK(); + } else if (entries_.size() >= max_entries_) { + return Status::TooManyRequests("credit ledger entry limit" LOC_MARK); + } Entry e; e.epoch = epoch; entries_.insert_or_assign(k, e); // fences all prior-epoch unused state diff --git a/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp b/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp index a328d61601..950c490410 100644 --- a/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp +++ b/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp @@ -90,6 +90,30 @@ TEST(ReceiverCredit, StaleEpochFailsAndActivationFencesOldState) { l.available(key(), CreditResource::DataBytes, v).IsInvalidEntry()); } +TEST(ReceiverCredit, ActivationReplayCannotMintCredit) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1); + ASSERT_TRUE(l.tryReserve(key(), charge(80, 1)).ok()); + ASSERT_TRUE(l.activate(key(), 7).ok()); // idempotent, not a reset + uint64_t v; + ASSERT_TRUE(l.consumed(key(), CreditResource::DataBytes, v).ok()); + EXPECT_EQ(v, 80); + EXPECT_TRUE(l.activate(key(), 6).IsInvalidEntry()); + ASSERT_TRUE(l.consumed(key(), CreditResource::DataBytes, v).ok()); + EXPECT_EQ(v, 80); +} + +TEST(ReceiverCredit, LedgerEntryCountIsBounded) { + SenderCreditLedger l(1); + ASSERT_TRUE(l.activate(key(), 7).ok()); + auto other = key(); + ++other.sender_peer; + EXPECT_TRUE(l.activate(other, 7).IsTooManyRequests()); + // A new epoch for an existing key does not consume another entry. + EXPECT_TRUE(l.activate(key(), 8).ok()); +} + TEST(ReceiverCredit, InvalidUpdateDoesNotPartiallyMutate) { SenderCreditLedger l; ASSERT_TRUE(l.activate(key(), 7).ok()); From d20fcb23f8c3a56383baeea69195094a1e964832 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Sat, 11 Jul 2026 22:42:27 +0800 Subject: [PATCH 4/5] tent: add epoch-safe credit session cleanup --- .../include/tent/runtime/receiver_credit.h | 4 +++ .../tent/src/runtime/receiver_credit.cpp | 17 +++++++++++- .../tent/tests/receiver_credit_test.cpp | 26 +++++++++++++++++++ 3 files changed, 46 insertions(+), 1 deletion(-) diff --git a/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h b/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h index 8948526106..09b243ab13 100644 --- a/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h +++ b/mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h @@ -67,6 +67,10 @@ class SenderCreditLedger { explicit SenderCreditLedger(size_t max_entries = 1024) : max_entries_(max_entries) {} Status activate(const CreditKey&, uint64_t epoch); + // Removes an exactly matched epoch after the caller has fenced and drained + // (or failed) its transport-owned work. An old cleanup cannot erase a + // reactivated, newer epoch. + Status deactivate(const CreditKey&, uint64_t epoch); Status applyUpdate(const CreditKey&, const ReceiverCreditUpdateV1&, CreditUpdateDisposition&); Status tryReserve(const CreditKey&, const CreditCharge&); diff --git a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp index 57f833662e..7de3eb6704 100644 --- a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp @@ -49,12 +49,27 @@ Status SenderCreditLedger::activate(const CreditKey& k, uint64_t epoch) { if (epoch < existing->second.epoch) return Status::InvalidEntry("stale credit activation" LOC_MARK); if (epoch == existing->second.epoch) return Status::OK(); + Entry replacement; + replacement.epoch = epoch; + existing->second = replacement; + return Status::OK(); } else if (entries_.size() >= max_entries_) { return Status::TooManyRequests("credit ledger entry limit" LOC_MARK); } Entry e; e.epoch = epoch; - entries_.insert_or_assign(k, e); // fences all prior-epoch unused state + entries_.emplace(k, e); + return Status::OK(); +} + +Status SenderCreditLedger::deactivate(const CreditKey& k, uint64_t epoch) { + if (!epoch) return Status::InvalidArgument("zero credit epoch" LOC_MARK); + std::lock_guard lock(mutex_); + auto existing = entries_.find(k); + if (existing == entries_.end()) return Status::OK(); // idempotent cleanup + if (existing->second.epoch != epoch) + return Status::InvalidEntry("credit cleanup epoch mismatch" LOC_MARK); + entries_.erase(existing); return Status::OK(); } diff --git a/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp b/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp index 950c490410..c50726d262 100644 --- a/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp +++ b/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp @@ -114,6 +114,32 @@ TEST(ReceiverCredit, LedgerEntryCountIsBounded) { EXPECT_TRUE(l.activate(key(), 8).ok()); } +TEST(ReceiverCredit, DeactivationReleasesCapacityAfterExactEpochFence) { + SenderCreditLedger l(1); + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1); + ASSERT_TRUE(l.tryReserve(key(), charge(80, 1)).ok()); + auto other = key(); + ++other.sender_peer; + EXPECT_TRUE(l.activate(other, 1).IsTooManyRequests()); + + EXPECT_TRUE(l.deactivate(key(), 6).IsInvalidEntry()); + EXPECT_TRUE(l.activate(other, 1).IsTooManyRequests()); + ASSERT_TRUE(l.deactivate(key(), 7).ok()); + ASSERT_TRUE(l.deactivate(key(), 7).ok()); // cleanup is idempotent + EXPECT_TRUE(l.activate(other, 1).ok()); +} + +TEST(ReceiverCredit, OldCleanupCannotEraseReactivatedEpoch) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + ASSERT_TRUE(l.activate(key(), 8).ok()); + EXPECT_TRUE(l.deactivate(key(), 7).IsInvalidEntry()); + CreditUpdateDisposition disposition; + auto fresh = update(8, 1, {{CreditResource::DataBytes, 10}}); + ASSERT_TRUE(l.applyUpdate(key(), fresh, disposition).ok()); +} + TEST(ReceiverCredit, InvalidUpdateDoesNotPartiallyMutate) { SenderCreditLedger l; ASSERT_TRUE(l.activate(key(), 7).ok()); From 5382c31a0f7bb8758dec918f335ddc1461bbcbf2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Tue, 14 Jul 2026 10:20:11 +0800 Subject: [PATCH 5/5] [TENT] Document partial receiver credit grants --- .../tent/src/runtime/receiver_credit.cpp | 4 ++++ .../tent/tests/receiver_credit_test.cpp | 18 ++++++++++++++++++ 2 files changed, 22 insertions(+) diff --git a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp index 7de3eb6704..5ae6c14a2e 100644 --- a/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp @@ -100,6 +100,10 @@ Status SenderCreditLedger::applyUpdate(const CreditKey& k, disposition = CreditUpdateDisposition::DuplicateOrOld; return Status::OK(); } + // `grants` is a partial cumulative update: each resource present in this + // message replaces that resource's cumulative grant total, while omitted + // resources retain their previous totals. This lets the receiver refresh + // only the resources whose available capacity changed. for (size_t i = 0; i < kCreditResourceCount; ++i) if (present[i] && (proposed[i] < e.grants[i] || proposed[i] < e.consumed[i])) diff --git a/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp b/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp index c50726d262..33faa04a2c 100644 --- a/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp +++ b/mooncake-transfer-engine/tent/tests/receiver_credit_test.cpp @@ -75,6 +75,24 @@ TEST(ReceiverCredit, SequenceGapIsVisibleAndSafe) { EXPECT_EQ(d, CreditUpdateDisposition::SequenceGap); } +TEST(ReceiverCredit, PartialGrantUpdateRetainsOmittedResources) { + SenderCreditLedger l; + ASSERT_TRUE(l.activate(key(), 7).ok()); + grant(l, 1, 100, 5); + + CreditUpdateDisposition d; + ASSERT_TRUE(l.applyUpdate( + key(), update(7, 2, {{CreditResource::DataBytes, 160}}), d) + .ok()); + EXPECT_EQ(d, CreditUpdateDisposition::Applied); + + uint64_t v; + ASSERT_TRUE(l.available(key(), CreditResource::DataBytes, v).ok()); + EXPECT_EQ(v, 160); + ASSERT_TRUE(l.available(key(), CreditResource::RequestSlots, v).ok()); + EXPECT_EQ(v, 5); +} + TEST(ReceiverCredit, StaleEpochFailsAndActivationFencesOldState) { SenderCreditLedger l; ASSERT_TRUE(l.activate(key(), 7).ok());