[TENT] Add receiver-credit ledger model and protocol invariants

mooncake-transfer-engine/tent/include/tent/runtime/receiver_credit.h

@@ -0,0 +1,98 @@
+// Copyright 2026 KVCache.AI
+// SPDX-License-Identifier: Apache-2.0
+
+#ifndef TENT_RUNTIME_RECEIVER_CREDIT_H
+#define TENT_RUNTIME_RECEIVER_CREDIT_H
+
+#include <array>
+#include <cstdint>
+#include <mutex>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#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<std::pair<CreditResource, uint64_t>> 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<CreditAmount> grants;
+};
+enum class CreditUpdateDisposition : uint8_t {
+    Applied,
+    DuplicateOrOld,
+    SequenceGap
+};
+
+// 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);
+    // 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&);
+    // 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<uint64_t, kCreditResourceCount> grants{}, consumed{};
+    };
+    static Status resourceIndex(CreditResource, size_t&);
+    static Status normalize(const CreditCharge&,
+                            std::array<uint64_t, kCreditResourceCount>&);
+    mutable std::mutex mutex_;
+    const size_t max_entries_;
+    std::unordered_map<CreditKey, Entry, CreditKeyHash> entries_;
+};
+
+}  // namespace mooncake::tent
+#endif

mooncake-transfer-engine/tent/src/runtime/receiver_credit.cpp

@@ -0,0 +1,179 @@
+// 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<uint64_t>{}(k.receiver_session.high);
+    auto mix = [&h](uint64_t v) {
+        h ^= std::hash<uint64_t>{}(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<uint16_t>(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<uint64_t, kCreditResourceCount>& 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 = 0;
+        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_);
+    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();
+        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_.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();
+}
+
+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<uint64_t, kCreditResourceCount> proposed{};
+    std::array<bool, kCreditResourceCount> present{};
+    for (auto a : u.grants) {
+        size_t i = 0;
+        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();
+    }
+    // `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]))
+            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<uint64_t, kCreditResourceCount> 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<uint64_t, kCreditResourceCount> 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 = 0;
+    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 = 0;
+    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

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)
+
 add_executable(bw_arbitration_test bw_arbitration_test.cpp)
 target_link_libraries(bw_arbitration_test PRIVATE tent_common gtest gtest_main)
 target_include_directories(bw_arbitration_test