YT-28832: Reject oversized SLRU cache items

This PR goes from this one https://nda.ya.ru/t/7joaDLmz7jP3J8
* Changelog entry
Type: fix
Component: misc-server

Prevent oversized SLRU cache entries from evicting cached data during insertion and resurrection.
commit_hash:3746876f597a05a6629be8bfb92763d5cc457ca5

5 files changed, 789 insertions, 62 deletions

diff --git a/yt/yt/core/misc/async_slru_cache-inl.h b/yt/yt/core/misc/async_slru_cache-inl.h
index ba7a7d1ae21..a137cd5e096 100644
--- a/yt/yt/core/misc/async_slru_cache-inl.h
+++ b/yt/yt/core/misc/async_slru_cache-inl.h
@@ -147,6 +147,17 @@ TIntrusiveListWithAutoDelete<TItem, TDelete> TAsyncSlruCacheListManager<TItem, T
 }
 
 template <class TItem, class TDerived>
+bool TAsyncSlruCacheListManager<TItem, TDerived>::IsOversized(i64 weight, i64 cookieWeight) const
+{
+    auto otherCookieWeight = CookieWeightCounter_ - cookieWeight;
+    YT_VERIFY(otherCookieWeight >= 0);
+
+    // Existing younger items are evicted before the newly inserted item. Older items, including
+    // those moved back to younger during trimming, and other insertion cookies are not.
+    return weight > Capacity_.load() - OlderWeightCounter_ - otherCookieWeight;
+}
+
+template <class TItem, class TDerived>
 bool TAsyncSlruCacheListManager<TItem, TDerived>::TouchItem(TItem* item)
 {
     if (item->Empty()) {
@@ -264,6 +275,9 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::TCounters::TCounters(
     auto profilerWithAsyncTag = profiler.WithTag("hit_type", "async");
     AsyncHitWeightCounter = profilerWithAsyncTag.Counter("/hit_weight");
     AsyncHitCounter = profilerWithAsyncTag.Counter("/hit_count");
+
+    RejectedOversizedCounter = profiler.Counter("/rejected_oversized");
+    RejectedOversizedWeightCounter = profiler.Counter("/rejected_oversized_weight");
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -274,7 +288,7 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::TAsyncSlruCacheBase(
     const NProfiling::TProfiler& profiler)
     : Config_(std::move(config))
     , Capacity_(Config_->Capacity)
-    , Counters_(profiler)
+    , MainCounters_(profiler)
     , SmallGhostCounters_(profiler.WithPrefix("/small_ghost_cache"))
     , LargeGhostCounters_(profiler.WithPrefix("/large_ghost_cache"))
 {
@@ -307,6 +321,7 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::TAsyncSlruCacheBase(
     });
 
     GhostCachesEnabled_.store(Config_->EnableGhostCaches);
+    RejectOversizedItems_.store(Config_->RejectOversizedItems);
 
     YT_VERIFY(IsPowerOf2(Config_->ShardCount));
     Shards_.reset(new TShard[Config_->ShardCount]);
@@ -328,10 +343,12 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::TAsyncSlruCacheBase(
 
             shard.SmallGhost.Reconfigure(
                 static_cast<i64>(shardCapacity * Config_->SmallGhostCacheRatio),
-                Config_->YoungerSizeFraction);
+                Config_->YoungerSizeFraction,
+                Config_->RejectOversizedItems);
             shard.LargeGhost.Reconfigure(
                 static_cast<i64>(shardCapacity * Config_->LargeGhostCacheRatio),
-                Config_->YoungerSizeFraction);
+                Config_->YoungerSizeFraction,
+                Config_->RejectOversizedItems);
         }
 
         shard.Parent = this;
@@ -350,17 +367,21 @@ void TAsyncSlruCacheBase<TKey, TValue, THash>::Reconfigure(const TSlruCacheDynam
     if (!config->EnableGhostCaches) {
         GhostCachesEnabled_.store(false);
     }
+    RejectOversizedItems_.store(config->RejectOversizedItems.value_or(Config_->RejectOversizedItems));
 
     for (int shardIndex = 0; shardIndex < Config_->ShardCount; ++shardIndex) {
         auto& shard = Shards_[shardIndex];
 
         if (GhostCachesEnabled_.load()) {
+            auto rejectOversizedItems = RejectOversizedItems_.load();
             shard.SmallGhost.Reconfigure(
                 static_cast<i64>(shardCapacity * Config_->SmallGhostCacheRatio),
-                youngerSizeFraction);
+                youngerSizeFraction,
+                rejectOversizedItems);
             shard.LargeGhost.Reconfigure(
                 static_cast<i64>(shardCapacity * Config_->LargeGhostCacheRatio),
-                youngerSizeFraction);
+                youngerSizeFraction,
+                rejectOversizedItems);
         }
 
         auto writerGuard = WriterGuard(shard.SpinLock);
@@ -393,21 +414,21 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::Find(const THeterogenousKey& key)
 
     auto itemIt = shard->ItemMap.find(key);
     if (itemIt == shard->ItemMap.end()) {
-        Counters_.MissedCounter.Increment();
+        MainCounters_.MissedCounter.Increment();
         return nullptr;
     }
 
     auto* item = itemIt->second;
     auto value = item->Value;
     if (!value) {
-        Counters_.MissedCounter.Increment();
+        MainCounters_.MissedCounter.Increment();
         return nullptr;
     }
 
     bool needToDrain = shard->TouchItem(item);
 
-    Counters_.SyncHitWeightCounter.Increment(item->CachedWeight);
-    Counters_.SyncHitCounter.Increment();
+    MainCounters_.SyncHitWeightCounter.Increment(item->CachedWeight);
+    MainCounters_.SyncHitCounter.Increment();
 
     readerGuard.Release();
 
@@ -470,9 +491,9 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::Lookup(const THeterogenousKey& key)
         shard->LargeGhost.Lookup(key);
     }
 
-    auto valueFuture = DoLookup(shard, key);
+    auto valueFuture = DoLookup(shard, key, /*resurrectGhostCachesOnRejectedValue*/ true);
     if (!valueFuture) {
-        Counters_.MissedCounter.Increment();
+        MainCounters_.MissedCounter.Increment();
     }
     return valueFuture;
 }
@@ -506,7 +527,10 @@ void TAsyncSlruCacheBase<TKey, TValue, THash>::Touch(const TValuePtr& value)
 template <class TKey, class TValue, class THash>
 template <class THeterogenousKey>
 typename TAsyncSlruCacheBase<TKey, TValue, THash>::TValueFuture
-TAsyncSlruCacheBase<TKey, TValue, THash>::DoLookup(TShard* shard, const THeterogenousKey& key)
+TAsyncSlruCacheBase<TKey, TValue, THash>::DoLookup(
+    TShard* shard,
+    const THeterogenousKey& key,
+    bool resurrectGhostCachesOnRejectedValue)
 {
     auto readerGuard = ReaderGuard(shard->SpinLock);
 
@@ -519,10 +543,10 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::DoLookup(TShard* shard, const THeterog
         auto valueFuture = item->GetValueFuture();
 
         if (item->Value) {
-            Counters_.SyncHitWeightCounter.Increment(item->CachedWeight);
-            Counters_.SyncHitCounter.Increment();
+            MainCounters_.SyncHitWeightCounter.Increment(item->CachedWeight);
+            MainCounters_.SyncHitCounter.Increment();
         } else {
-            Counters_.AsyncHitCounter.Increment();
+            MainCounters_.AsyncHitCounter.Increment();
             item->AsyncHitCount.fetch_add(1);
         }
 
@@ -557,10 +581,10 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::DoLookup(TShard* shard, const THeterog
         auto valueFuture = item->GetValueFuture();
 
         if (item->Value) {
-            Counters_.SyncHitWeightCounter.Increment(item->CachedWeight);
-            Counters_.SyncHitCounter.Increment();
+            MainCounters_.SyncHitWeightCounter.Increment(item->CachedWeight);
+            MainCounters_.SyncHitCounter.Increment();
         } else {
-            Counters_.AsyncHitCounter.Increment();
+            MainCounters_.AsyncHitCounter.Increment();
             item->AsyncHitCount.fetch_add(1);
         }
 
@@ -571,6 +595,23 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::DoLookup(TShard* shard, const THeterog
 
     shard->DrainTouchBuffer();
 
+    i64 weight = GetWeight(value);
+    if (RejectOversizedItems_.load() && shard->IsOversized(weight, /*cookieWeight*/ 0)) {
+        MainCounters_.SyncHitWeightCounter.Increment(weight);
+        MainCounters_.SyncHitCounter.Increment();
+        MainCounters_.RejectedOversizedCounter.Increment();
+        MainCounters_.RejectedOversizedWeightCounter.Increment(weight);
+
+        writerGuard.Release();
+
+        if (resurrectGhostCachesOnRejectedValue && GhostCachesEnabled_.load()) {
+            shard->SmallGhost.Resurrect(value, weight);
+            shard->LargeGhost.Resurrect(value, weight);
+        }
+
+        return MakeFuture(value);
+    }
+
     {
         auto* item = new TItem(value);
         value->Item_ = item;
@@ -580,10 +621,9 @@ TAsyncSlruCacheBase<TKey, TValue, THash>::DoLookup(TShard* shard, const THeterog
         EmplaceOrCrash(itemMap, key, item);
         ++Size_;
 
-        i64 weight = GetWeight(item->Value);
         shard->PushToYounger(item, weight);
-        Counters_.SyncHitWeightCounter.Increment(weight);
-        Counters_.SyncHitCounter.Increment();
+        MainCounters_.SyncHitWeightCounter.Increment(weight);
+        MainCounters_.SyncHitCounter.Increment();
 
         // NB: Releases the lock.
         TrimWithNotify(shard, writerGuard, value);
@@ -602,7 +642,7 @@ auto TAsyncSlruCacheBase<TKey, TValue, THash>::BeginInsert(const TKey& key, i64
 {
     auto* shard = GetShardByKey(key);
 
-    if (auto valueFuture = DoLookup(shard, key)) {
+    if (auto valueFuture = DoLookup(shard, key, /*resurrectGhostCachesOnRejectedValue*/ false)) {
         if (GhostCachesEnabled_.load()) {
             if (valueFuture.IsSet() && valueFuture.GetOrCrash().IsOK()) {
                 bool smallInserted = shard->SmallGhost.BeginInsert(key, cookieWeight);
@@ -645,10 +685,10 @@ auto TAsyncSlruCacheBase<TKey, TValue, THash>::BeginInsert(const TKey& key, i64
             auto valueFuture = item->GetValueFuture();
 
             if (item->Value) {
-                Counters_.SyncHitWeightCounter.Increment(item->CachedWeight);
-                Counters_.SyncHitCounter.Increment();
+                MainCounters_.SyncHitWeightCounter.Increment(item->CachedWeight);
+                MainCounters_.SyncHitCounter.Increment();
             } else {
-                Counters_.AsyncHitCounter.Increment();
+                MainCounters_.AsyncHitCounter.Increment();
                 item->AsyncHitCount.fetch_add(1);
             }
 
@@ -686,7 +726,7 @@ auto TAsyncSlruCacheBase<TKey, TValue, THash>::BeginInsert(const TKey& key, i64
             EmplaceOrCrash(itemMap, key, item);
             ++Size_;
 
-            Counters_.MissedCounter.Increment();
+            MainCounters_.MissedCounter.Increment();
 
             shard->UpdateCookie(item, /*countDelta*/ 1, cookieWeight);
             if (cookieWeight > 0) {
@@ -711,21 +751,30 @@ auto TAsyncSlruCacheBase<TKey, TValue, THash>::BeginInsert(const TKey& key, i64
         }
 
         if (auto value = valueIt->second.Lock()) {
-            auto* item = new TItem(value);
-            value->Item_ = item;
+            i64 weight = GetWeight(value);
+            if (RejectOversizedItems_.load() && shard->IsOversized(weight, /*cookieWeight*/ 0)) {
+                MainCounters_.SyncHitWeightCounter.Increment(weight);
+                MainCounters_.SyncHitCounter.Increment();
+                MainCounters_.RejectedOversizedCounter.Increment();
+                MainCounters_.RejectedOversizedWeightCounter.Increment(weight);
 
-            EmplaceOrCrash(itemMap, key, item);
-            ++Size_;
+                guard.Release();
+            } else {
+                auto* item = new TItem(value);
+                value->Item_ = item;
 
-            i64 weight = GetWeight(item->Value);
-            shard->PushToYounger(item, weight);
-            Counters_.SyncHitWeightCounter.Increment(weight);
-            Counters_.SyncHitCounter.Increment();
+                EmplaceOrCrash(itemMap, key, item);
+                ++Size_;
 
-            // NB: Releases the lock.
-            TrimWithNotify(shard, guard, value);
+                shard->PushToYounger(item, weight);
+                MainCounters_.SyncHitWeightCounter.Increment(weight);
+                MainCounters_.SyncHitCounter.Increment();
 
-            guard.Release();
+                // NB: Releases the lock.
+                TrimWithNotify(shard, guard, value);
+
+                guard.Release();
+            }
 
             if (GhostCachesEnabled_.load()) {
                 shard->SmallGhost.Resurrect(value, weight);
@@ -798,9 +847,43 @@ void TAsyncSlruCacheBase<TKey, TValue, THash>::EndInsert(const TInsertCookie& in
 
     shard->DrainTouchBuffer();
 
+    auto* item = GetOrCrash(shard->ItemMap, key);
+    YT_VERIFY(!item->Value);
+
+    i64 weight = GetWeight(value);
+    if (RejectOversizedItems_.load()) {
+        auto cookieWeight = item->CachedWeight;
+        if (shard->IsOversized(weight, cookieWeight)) {
+            auto asyncHitCount = item->AsyncHitCount.load();
+            shard->UpdateCookie(item, /*countDelta*/ -1, -cookieWeight);
+            EraseOrCrash(shard->ItemMap, key);
+            --Size_;
+
+            auto promise = std::move(item->ValuePromise);
+            delete item;
+
+            MainCounters_.MissedWeightCounter.Increment(weight);
+            MainCounters_.AsyncHitWeightCounter.Increment(weight * asyncHitCount);
+            MainCounters_.RejectedOversizedCounter.Increment();
+            MainCounters_.RejectedOversizedWeightCounter.Increment(weight);
+
+            // NB: Releases the lock.
+            TrimWithNotify(shard, guard, nullptr, -cookieWeight);
+
+            if (insertCookie.InsertedIntoSmallGhost_) {
+                shard->SmallGhost.EndInsert(value, weight);
+            }
+            if (insertCookie.InsertedIntoLargeGhost_) {
+                shard->LargeGhost.EndInsert(value, weight);
+            }
+
+            promise.Set(value);
+            return;
+        }
+    }
+
     value->SetCache(MakeWeak(this));
 
-    auto* item = GetOrCrash(shard->ItemMap, key);
     item->Value = value;
     value->Item_ = item;
     auto promise = item->ValuePromise;
@@ -810,11 +893,10 @@ void TAsyncSlruCacheBase<TKey, TValue, THash>::EndInsert(const TInsertCookie& in
     auto cookieWeight = item->CachedWeight;
     shard->UpdateCookie(item, /*countDelta*/ -1, -cookieWeight);
 
-    i64 weight = GetWeight(item->Value);
     shard->PushToYounger(item, weight);
     // MissedCounter and AsyncHitCounter have already been incremented in BeginInsert.
-    Counters_.MissedWeightCounter.Increment(weight);
-    Counters_.AsyncHitWeightCounter.Increment(weight * item->AsyncHitCount.load());
+    MainCounters_.MissedWeightCounter.Increment(weight);
+    MainCounters_.AsyncHitWeightCounter.Increment(weight * item->AsyncHitCount.load());
 
     // NB: Releases the lock.
     TrimWithNotify(shard, guard, value, -cookieWeight);
@@ -1023,7 +1105,7 @@ void TAsyncSlruCacheBase<TKey, TValue, THash>::UpdateWeight(const TKey& key)
     // If item weight increases, it means that some parts of the item were missing in cache,
     // so add delta to missed weight.
     if (weightDelta > 0) {
-        Counters_.MissedWeightCounter.Increment(weightDelta);
+        MainCounters_.MissedWeightCounter.Increment(weightDelta);
     }
 
     TrimWithNotify(shard, guard, nullptr, weightDelta);
@@ -1072,6 +1154,12 @@ bool TAsyncSlruCacheBase<TKey, TValue, THash>::IsResurrectionSupported() const
 }
 
 template <class TKey, class TValue, class THash>
+auto TAsyncSlruCacheBase<TKey, TValue, THash>::GetMainCounters() const -> const TCounters&
+{
+    return MainCounters_;
+}
+
+template <class TKey, class TValue, class THash>
 auto TAsyncSlruCacheBase<TKey, TValue, THash>::GetSmallGhostCounters() const -> const TCounters&
 {
     return SmallGhostCounters_;
@@ -1240,6 +1328,19 @@ void TAsyncSlruCacheBase<TKey, TValue, THash>::TGhostShard::EndInsert(TValuePtr
     YT_VERIFY(!item->Inserted);
     this->UpdateCookie(item, /*countDelta*/ -1, -item->CachedWeight);
 
+    if (RejectOversizedItems_ && this->IsOversized(weight, /*cookieWeight*/ 0)) {
+        EraseOrCrash(ItemMap_, key);
+
+        Counters_->MissedWeightCounter.Increment(weight);
+        Counters_->AsyncHitWeightCounter.Increment(weight * item->AsyncHitCount.load());
+        Counters_->RejectedOversizedCounter.Increment();
+        Counters_->RejectedOversizedWeightCounter.Increment(weight);
+
+        guard.Release();
+        delete item;
+        return;
+    }
+
     item->Value = std::move(value);
     item->Inserted = true;
 
@@ -1267,6 +1368,14 @@ void TAsyncSlruCacheBase<TKey, TValue, THash>::TGhostShard::Resurrect(const TVal
         return;
     }
 
+    if (RejectOversizedItems_ && this->IsOversized(weight, /*cookieWeight*/ 0)) {
+        Counters_->SyncHitWeightCounter.Increment(weight);
+        Counters_->SyncHitCounter.Increment();
+        Counters_->RejectedOversizedCounter.Increment();
+        Counters_->RejectedOversizedWeightCounter.Increment(weight);
+        return;
+    }
+
     auto* item = new TGhostItem(key);
     item->Value = value;
     item->Inserted = true;
@@ -1376,9 +1485,10 @@ void TAsyncSlruCacheBase<TKey, TValue, THash>::TGhostShard::UpdateCookieWeight(c
 }
 
 template <class TKey, class TValue, class THash>
-void TAsyncSlruCacheBase<TKey, TValue, THash>::TGhostShard::Reconfigure(i64 capacity, double youngerSizeFraction)
+void TAsyncSlruCacheBase<TKey, TValue, THash>::TGhostShard::Reconfigure(i64 capacity, double youngerSizeFraction, bool rejectOversizedItems)
 {
     auto writerGuard = WriterGuard(SpinLock_);
+    RejectOversizedItems_ = rejectOversizedItems;
     TAsyncSlruCacheListManager<TGhostItem, TGhostShard>::Reconfigure(capacity, youngerSizeFraction);
     this->DrainTouchBuffer();
     Trim(writerGuard);
diff --git a/yt/yt/core/misc/async_slru_cache.h b/yt/yt/core/misc/async_slru_cache.h
index 096931035b3..4d9a7ae8cd6 100644
--- a/yt/yt/core/misc/async_slru_cache.h
+++ b/yt/yt/core/misc/async_slru_cache.h
@@ -87,6 +87,8 @@ public:
 
     TIntrusiveListWithAutoDelete<TItem, TDelete> TrimNoDelete();
 
+    bool IsOversized(i64 weight, i64 cookieWeight) const;
+
     bool TouchItem(TItem* item);
 
     //! Drains touch buffer. You MUST call this function before trying to remove anything from the
@@ -146,6 +148,9 @@ private:
 //! pointer to this value), it returns back to the cache. This behavior may be overloaded by
 //! overriding IsResurrectionSupported() function.
 //!
+//! When oversized-item rejection is enabled, a value that cannot survive cache trimming is not
+//! admitted or resurrected. It is still returned to the caller or delivered to concurrent waiters.
+//!
 //! This cache is quite complex and has many invariants. Read about them below and change the
 //! code carefully.
 template <class TKey, class TValue, class THash = THash<TKey>>
@@ -176,6 +181,7 @@ public:
         void UpdateWeight(i64 newWeight);
 
         void Cancel(const TError& error);
+
         void EndInsert(TValuePtr value);
 
     private:
@@ -272,9 +278,12 @@ protected:
         NProfiling::TCounter SyncHitCounter;
         NProfiling::TCounter AsyncHitCounter;
         NProfiling::TCounter MissedCounter;
+        NProfiling::TCounter RejectedOversizedCounter;
+        NProfiling::TCounter RejectedOversizedWeightCounter;
     };
 
     //! For testing purposes only.
+    const TCounters& GetMainCounters() const;
     const TCounters& GetSmallGhostCounters() const;
     const TCounters& GetLargeGhostCounters() const;
 
@@ -367,7 +376,7 @@ private:
 
         using TAsyncSlruCacheListManager<TGhostItem, TGhostShard>::SetTouchBufferCapacity;
 
-        void Reconfigure(i64 capacity, double youngerSizeFraction);
+        void Reconfigure(i64 capacity, double youngerSizeFraction, bool rejectOversizedItems);
 
         DEFINE_BYVAL_RW_PROPERTY(TCounters*, Counters);
 
@@ -377,6 +386,7 @@ private:
         YT_DECLARE_SPIN_LOCK(NThreading::TReaderWriterSpinLock, SpinLock_);
 
         THashMap<TKey, TGhostItem*, THash> ItemMap_;
+        bool RejectOversizedItems_ = false;
 
         template <class THeterogenousKey>
         bool DoLookup(const THeterogenousKey& key, bool allowAsyncHits);
@@ -442,7 +452,7 @@ private:
     std::atomic<int> Size_ = 0;
     std::atomic<i64> Capacity_;
 
-    TCounters Counters_;
+    TCounters MainCounters_;
     TCounters SmallGhostCounters_;
     TCounters LargeGhostCounters_;
 
@@ -455,12 +465,13 @@ private:
     std::atomic<i64> CookieWeightCounter_ = 0;
 
     std::atomic<bool> GhostCachesEnabled_;
+    std::atomic<bool> RejectOversizedItems_;
 
     template <class THeterogenousKey>
     TShard* GetShardByKey(const THeterogenousKey& key) const;
 
     template <class THeterogenousKey>
-    TValueFuture DoLookup(TShard* shard, const THeterogenousKey& key);
+    TValueFuture DoLookup(TShard* shard, const THeterogenousKey& key, bool resurrectGhostCachesOnRejectedValue);
 
     void DoTryRemove(const TKey& key, const TValuePtr& value, bool forbidResurrection);
 
diff --git a/yt/yt/core/misc/cache_config.cpp b/yt/yt/core/misc/cache_config.cpp
index f77a5c51a2f..d6fd0825e01 100644
--- a/yt/yt/core/misc/cache_config.cpp
+++ b/yt/yt/core/misc/cache_config.cpp
@@ -36,6 +36,8 @@ void TSlruCacheConfig::Register(TRegistrar registrar)
         .GreaterThanOrEqual(0.0);
     registrar.Parameter("enable_ghost_caches", &TThis::EnableGhostCaches)
         .Default(true);
+    registrar.Parameter("reject_oversized_items", &TThis::RejectOversizedItems)
+        .Default(false);
 
     registrar.Postprocessor([] (TThis* config) {
         if (!IsPowerOf2(config->ShardCount)) {
@@ -57,6 +59,8 @@ void TSlruCacheDynamicConfig::Register(TRegistrar registrar)
         .InRange(0.0, 1.0);
     registrar.Parameter("enable_ghost_caches", &TThis::EnableGhostCaches)
         .Default(true);
+    registrar.Parameter("reject_oversized_items", &TThis::RejectOversizedItems)
+        .Default();
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/core/misc/cache_config.h b/yt/yt/core/misc/cache_config.h
index 7d90bd2fe13..67429b0502b 100644
--- a/yt/yt/core/misc/cache_config.h
+++ b/yt/yt/core/misc/cache_config.h
@@ -33,6 +33,9 @@ struct TSlruCacheConfig
     //! re-enabled again (i.e. the value of this field is ignored).
     bool EnableGhostCaches;
 
+    //! If set, items that cannot survive cache trimming are not admitted.
+    bool RejectOversizedItems;
+
     static TSlruCacheConfigPtr CreateWithCapacity(i64 capacity, int shardCount = 1);
 
     REGISTER_YSON_STRUCT(TSlruCacheConfig);
@@ -58,6 +61,9 @@ struct TSlruCacheDynamicConfig
     //! re-enabled again (i.e. the value of this field is ignored).
     bool EnableGhostCaches;
 
+    //! If set, items that cannot survive cache trimming are not admitted.
+    std::optional<bool> RejectOversizedItems;
+
     REGISTER_YSON_STRUCT(TSlruCacheDynamicConfig);
 
     static void Register(TRegistrar registrar);
diff --git a/yt/yt/core/misc/unittests/async_slru_cache_ut.cpp b/yt/yt/core/misc/unittests/async_slru_cache_ut.cpp
index 4c06b2fb6e9..57e49f5a344 100644
--- a/yt/yt/core/misc/unittests/async_slru_cache_ut.cpp
+++ b/yt/yt/core/misc/unittests/async_slru_cache_ut.cpp
@@ -14,6 +14,7 @@
 
 #include <util/generic/strbuf.h>
 
+#include <array>
 #include <random>
 
 namespace NYT {
@@ -66,6 +67,8 @@ public:
         i64 SyncHit;
         i64 AsyncHit;
         i64 Missed;
+        i64 RejectedOversized;
+        i64 RejectedOversizedWeight;
 
         TCountersState operator-(const TCountersState& other) const
         {
@@ -75,11 +78,18 @@ public:
                 .MissedWeight = MissedWeight - other.MissedWeight,
                 .SyncHit = SyncHit - other.SyncHit,
                 .AsyncHit = AsyncHit - other.AsyncHit,
-                .Missed = Missed - other.Missed
+                .Missed = Missed - other.Missed,
+                .RejectedOversized = RejectedOversized - other.RejectedOversized,
+                .RejectedOversizedWeight = RejectedOversizedWeight - other.RejectedOversizedWeight
             };
         }
     };
 
+    TCountersState ReadMainCounters() const
+    {
+        return ReadCounters(GetMainCounters());
+    }
+
     TCountersState ReadSmallGhostCounters() const
     {
         return ReadCounters(GetSmallGhostCounters());
@@ -104,7 +114,9 @@ protected:
             .MissedWeight = TTesting::ReadCounter(counters.MissedWeightCounter),
             .SyncHit = TTesting::ReadCounter(counters.SyncHitCounter),
             .AsyncHit = TTesting::ReadCounter(counters.AsyncHitCounter),
-            .Missed = TTesting::ReadCounter(counters.MissedCounter)
+            .Missed = TTesting::ReadCounter(counters.MissedCounter),
+            .RejectedOversized = TTesting::ReadCounter(counters.RejectedOversizedCounter),
+            .RejectedOversizedWeight = TTesting::ReadCounter(counters.RejectedOversizedWeightCounter)
         };
     }
 };
@@ -344,6 +356,7 @@ TEST(TAsyncSlruCacheTest, UpdateWeight)
 {
     const int cacheSize = 10;
     auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
     auto cache = New<TSimpleSlruCache>(config);
 
     for (int i = 0; i < cacheSize; ++i) {
@@ -384,8 +397,8 @@ TEST(TAsyncSlruCacheTest, UpdateWeight)
         // But now '0' should be moved to 'youngest' after Trim() call.
         // Second insert should delete '0' and insert '1' because it's newer.
         cookie = cache->BeginInsert(1);
-        // Cookie is not active because we still hold value and it can be resurrected.
-        EXPECT_FALSE(cookie.IsActive());
+        EXPECT_TRUE(cookie.IsActive());
+        cookie.EndInsert(value);
 
         // '0' is deleted, because it is too big.
         EXPECT_EQ(cache->Find(0), nullptr);
@@ -551,10 +564,11 @@ TEST(TAsyncSlruCacheTest, AddRemoveStressTest)
     threadPool->Shutdown();
 }
 
-TEST(TAsyncSlruCacheTest, AddThenImmediatelyRemove)
+TEST(TAsyncSlruCacheTest, OversizedValueIsRejectedWithoutCallbacks)
 {
     constexpr int cacheSize = 1;
     auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
     auto cache = New<TCountingSlruCache>(std::move(config));
 
     auto persistentValue = New<TSimpleCachedValue>(
@@ -566,8 +580,8 @@ TEST(TAsyncSlruCacheTest, AddThenImmediatelyRemove)
         auto cookie = cache->BeginInsert(0);
         cookie.EndInsert(persistentValue);
         EXPECT_EQ(cache->GetItemCount(), 0);
-        EXPECT_EQ(cache->GetTotalAdded(), 1);
-        EXPECT_EQ(cache->GetTotalRemoved(), 1);
+        EXPECT_EQ(cache->GetTotalAdded(), 0);
+        EXPECT_EQ(cache->GetTotalRemoved(), 0);
     }
 
     {
@@ -579,15 +593,14 @@ TEST(TAsyncSlruCacheTest, AddThenImmediatelyRemove)
         cookie.EndInsert(temporaryValue);
         temporaryValue.Reset();
         EXPECT_EQ(cache->GetItemCount(), 0);
-        EXPECT_EQ(cache->GetTotalAdded(), 2);
-        EXPECT_EQ(cache->GetTotalRemoved(), 2);
+        EXPECT_EQ(cache->GetTotalAdded(), 0);
+        EXPECT_EQ(cache->GetTotalRemoved(), 0);
     }
 
     {
-        auto value = WaitForFast(cache->Lookup(0))
-            .ValueOrThrow();
+        auto value = cache->Lookup(0);
         EXPECT_EQ(cache->GetItemCount(), 0);
-        EXPECT_EQ(value->Value, 42);
+        ASSERT_FALSE(static_cast<bool>(value));
     }
 
     {
@@ -597,6 +610,302 @@ TEST(TAsyncSlruCacheTest, AddThenImmediatelyRemove)
     }
 }
 
+TEST(TAsyncSlruCacheTest, OversizedValueDoesNotEvictCachedValues)
+{
+    constexpr int cacheSize = 1;
+    auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TCountingSlruCache>(std::move(config));
+
+    auto cachedValue = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 42,
+        /*weight*/ 1);
+
+    {
+        auto cookie = cache->BeginInsert(0);
+        cookie.EndInsert(cachedValue);
+        EXPECT_EQ(cache->GetItemCount(), 1);
+        EXPECT_EQ(cache->GetTotalAdded(), 1);
+        EXPECT_EQ(cache->GetTotalRemoved(), 0);
+    }
+
+    {
+        auto cookie = cache->BeginInsert(1);
+        auto oversizedValue = New<TSimpleCachedValue>(
+            /*key*/ 1,
+            /*value*/ 43,
+            /*weight*/ 100);
+
+        cookie.EndInsert(oversizedValue);
+
+        EXPECT_EQ(cache->GetItemCount(), 1);
+        EXPECT_EQ(cache->GetTotalAdded(), 1);
+        EXPECT_EQ(cache->GetTotalRemoved(), 0);
+        EXPECT_EQ(cache->Find(0), cachedValue);
+        EXPECT_FALSE(static_cast<bool>(cache->Lookup(1)));
+    }
+}
+
+TEST(TAsyncSlruCacheTest, OversizedValueDoesNotWashYoungerSegment)
+{
+    constexpr int cacheSize = 100;
+    auto config = CreateCacheConfig(cacheSize);
+    config->YoungerSizeFraction = 0.25;
+    config->RejectOversizedItems = true;
+    auto cache = New<TCountingSlruCache>(std::move(config));
+
+    std::vector<TSimpleCachedValuePtr> olderValues;
+    for (int index = 0; index < 3; ++index) {
+        auto value = New<TSimpleCachedValue>(
+            /*key*/ index,
+            /*value*/ 42,
+            /*weight*/ 25);
+        auto cookie = cache->BeginInsert(index);
+        cookie.EndInsert(value);
+        EXPECT_EQ(cache->Find(index), value);
+        olderValues.push_back(std::move(value));
+    }
+
+    auto youngerValue = New<TSimpleCachedValue>(
+        /*key*/ 3,
+        /*value*/ 43,
+        /*weight*/ 25);
+    auto cookie = cache->BeginInsert(3);
+    cookie.EndInsert(youngerValue);
+
+    auto oversizedValue = New<TSimpleCachedValue>(
+        /*key*/ 4,
+        /*value*/ 44,
+        /*weight*/ 40);
+    cookie = cache->BeginInsert(4);
+    cookie.EndInsert(oversizedValue);
+
+    EXPECT_EQ(cache->GetItemCount(), 4);
+    EXPECT_EQ(cache->GetTotalAdded(), 4);
+    EXPECT_EQ(cache->GetTotalRemoved(), 0);
+    for (int index = 0; index < 3; ++index) {
+        EXPECT_EQ(cache->Find(index), olderValues[index]);
+    }
+    EXPECT_EQ(cache->Find(3), youngerValue);
+    EXPECT_FALSE(static_cast<bool>(cache->Lookup(4)));
+}
+
+TEST(TAsyncSlruCacheTest, ValueLargerThanYoungerSegmentFitsInEmptyShard)
+{
+    constexpr int cacheSize = 100;
+    auto config = CreateCacheConfig(cacheSize);
+    config->YoungerSizeFraction = 0.25;
+    config->RejectOversizedItems = true;
+    auto cache = New<TCountingSlruCache>(std::move(config));
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 42,
+        /*weight*/ 40);
+    auto cookie = cache->BeginInsert(0);
+    cookie.EndInsert(value);
+
+    EXPECT_EQ(cache->GetItemCount(), 1);
+    EXPECT_EQ(cache->GetTotalAdded(), 1);
+    EXPECT_EQ(cache->GetTotalRemoved(), 0);
+    EXPECT_EQ(cache->Find(0), value);
+}
+
+TEST(TAsyncSlruCacheTest, ValueAtShardCapacityWithCookieWeightIsAdmitted)
+{
+    constexpr int cacheSize = 100;
+    auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TCountingSlruCache>(std::move(config));
+
+    auto cookie = cache->BeginInsert(/*key*/ 0, /*cookieWeight*/ cacheSize);
+    ASSERT_TRUE(cookie.IsActive());
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 42,
+        /*weight*/ cacheSize);
+    cookie.EndInsert(value);
+
+    EXPECT_EQ(cache->GetItemCount(), 1);
+    EXPECT_EQ(cache->GetTotalAdded(), 1);
+    EXPECT_EQ(cache->GetTotalRemoved(), 0);
+    EXPECT_EQ(cache->Find(0), value);
+}
+
+TEST(TAsyncSlruCacheTest, OversizedValueAccountsForOtherCookies)
+{
+    constexpr int cacheSize = 100;
+    auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TCountingSlruCache>(std::move(config));
+
+    auto otherCookie = cache->BeginInsert(/*key*/ 0, /*cookieWeight*/ 80);
+    ASSERT_TRUE(otherCookie.IsActive());
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 1,
+        /*value*/ 42,
+        /*weight*/ 30);
+    auto cookie = cache->BeginInsert(1);
+    cookie.EndInsert(value);
+
+    EXPECT_EQ(cache->GetItemCount(), 0);
+    EXPECT_EQ(cache->GetTotalAdded(), 0);
+    EXPECT_EQ(cache->GetTotalRemoved(), 0);
+    EXPECT_FALSE(static_cast<bool>(cache->Lookup(1)));
+    EXPECT_TRUE(otherCookie.IsActive());
+
+    otherCookie.Cancel(TError("Cancelled"));
+}
+
+TEST(TAsyncSlruCacheTest, OversizedValueIsIsolatedToShard)
+{
+    constexpr int cacheSize = 100;
+    constexpr int shardCount = 2;
+    constexpr int shardCapacity = cacheSize / shardCount;
+    auto config = CreateCacheConfig(cacheSize);
+    config->ShardCount = shardCount;
+    config->RejectOversizedItems = true;
+    auto cache = New<TCountingSlruCache>(std::move(config));
+
+    std::array<TSimpleCachedValuePtr, shardCount> cachedValues;
+    for (int key = 0; key < shardCount; ++key) {
+        cachedValues[key] = New<TSimpleCachedValue>(
+            key,
+            /*value*/ 42,
+            /*weight*/ shardCapacity);
+        auto cookie = cache->BeginInsert(key);
+        ASSERT_TRUE(cookie.IsActive());
+        cookie.EndInsert(cachedValues[key]);
+    }
+
+    auto cookie = cache->BeginInsert(/*key*/ shardCount);
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(New<TSimpleCachedValue>(
+        /*key*/ shardCount,
+        /*value*/ 43,
+        /*weight*/ shardCapacity + 1));
+
+    EXPECT_EQ(cache->GetItemCount(), shardCount);
+    EXPECT_EQ(cache->GetTotalAdded(), shardCount);
+    EXPECT_EQ(cache->GetTotalRemoved(), 0);
+    for (int key = 0; key < shardCount; ++key) {
+        EXPECT_EQ(cache->Find(key), cachedValues[key]);
+    }
+    EXPECT_FALSE(static_cast<bool>(cache->Lookup(shardCount)));
+}
+
+TEST(TAsyncSlruCacheTest, OversizedValueIsDeliveredToWaiters)
+{
+    auto config = CreateCacheConfig(/*cacheSize*/ 1);
+    config->RejectOversizedItems = true;
+    auto cache = New<TSimpleSlruCache>(std::move(config));
+
+    auto insertCookie = cache->BeginInsert(0);
+    ASSERT_TRUE(insertCookie.IsActive());
+
+    auto waitingCookie = cache->BeginInsert(0);
+    ASSERT_FALSE(waitingCookie.IsActive());
+    auto valueFuture = waitingCookie.GetValue();
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 42,
+        /*weight*/ 100);
+    insertCookie.EndInsert(value);
+
+    EXPECT_EQ(WaitForFast(valueFuture).ValueOrThrow(), value);
+    EXPECT_EQ(cache->Find(0), nullptr);
+}
+
+TEST(TAsyncSlruCacheTest, DisableOversizedItemRejectionDynamically)
+{
+    auto config = CreateCacheConfig(/*cacheSize*/ 1);
+    config->RejectOversizedItems = true;
+    auto cache = New<TCountingSlruCache>(std::move(config));
+
+    auto dynamicConfig = New<TSlruCacheDynamicConfig>();
+    dynamicConfig->RejectOversizedItems = false;
+    cache->Reconfigure(dynamicConfig);
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 42,
+        /*weight*/ 100);
+    auto cookie = cache->BeginInsert(0);
+    cookie.EndInsert(value);
+
+    EXPECT_EQ(cache->GetItemCount(), 0);
+    EXPECT_EQ(cache->GetTotalAdded(), 1);
+    EXPECT_EQ(cache->GetTotalRemoved(), 1);
+    EXPECT_EQ(WaitForFast(cache->Lookup(0)).ValueOrThrow(), value);
+}
+
+TEST(TAsyncSlruCacheTest, EnableOversizedItemRejectionDynamically)
+{
+    auto config = CreateCacheConfig(/*cacheSize*/ 1);
+    config->RejectOversizedItems = false;
+    auto cache = New<TCountingSlruCache>(std::move(config));
+
+    auto dynamicConfig = New<TSlruCacheDynamicConfig>();
+    dynamicConfig->RejectOversizedItems = true;
+    cache->Reconfigure(dynamicConfig);
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 42,
+        /*weight*/ 100);
+    auto cookie = cache->BeginInsert(0);
+    cookie.EndInsert(value);
+
+    EXPECT_EQ(cache->GetItemCount(), 0);
+    EXPECT_EQ(cache->GetTotalAdded(), 0);
+    EXPECT_EQ(cache->GetTotalRemoved(), 0);
+    EXPECT_FALSE(static_cast<bool>(cache->Lookup(0)));
+}
+
+TEST(TAsyncSlruCacheTest, OversizedValueDoesNotWashCacheOnResurrection)
+{
+    constexpr int initialCacheSize = 100;
+    constexpr int reconfiguredCacheSize = 30;
+    auto config = CreateCacheConfig(initialCacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TSimpleSlruCache>(std::move(config), TProfiler{"/cache"});
+
+    auto oversizedValue = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 42,
+        /*weight*/ 40);
+    auto cookie = cache->BeginInsert(oversizedValue->GetKey());
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(oversizedValue);
+
+    auto dynamicConfig = New<TSlruCacheDynamicConfig>();
+    dynamicConfig->Capacity = reconfiguredCacheSize;
+    cache->Reconfigure(dynamicConfig);
+    EXPECT_EQ(cache->GetSize(), 0);
+
+    auto cachedValue = New<TSimpleCachedValue>(
+        /*key*/ 1,
+        /*value*/ 43,
+        /*weight*/ reconfiguredCacheSize);
+    cookie = cache->BeginInsert(cachedValue->GetKey());
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(cachedValue);
+
+    auto oldCounters = cache->ReadMainCounters();
+    EXPECT_EQ(WaitForFast(cache->Lookup(oversizedValue->GetKey())).ValueOrThrow(), oversizedValue);
+
+    EXPECT_EQ(cache->GetSize(), 1);
+    EXPECT_EQ(cache->Find(cachedValue->GetKey()), cachedValue);
+
+    auto counters = cache->ReadMainCounters() - oldCounters;
+    EXPECT_EQ(counters.RejectedOversized, 1);
+    EXPECT_EQ(counters.RejectedOversizedWeight, oversizedValue->Weight);
+}
+
 TEST(TAsyncSlruCacheTest, TouchRemovedValue)
 {
     constexpr int cacheSize = 100;
@@ -903,13 +1212,13 @@ TEST(TAsyncSlruGhostCacheTest, MoveAssignCookie)
             /*value*/ 42,
             /*weight*/ 1));
     }
-    {
-        auto cookie = cache->BeginInsert(43);
+    for (int index = 43; index < 47; ++index) {
+        auto cookie = cache->BeginInsert(index);
         ASSERT_TRUE(cookie.IsActive());
         cookie.EndInsert(New<TSimpleCachedValue>(
-            /*key*/ 43,
+            /*key*/ index,
             /*value*/ 100500,
-            /*weight*/ 101));
+            /*weight*/ cacheSize / 4));
     }
 
     for (int index = 0; index < 5; ++index) {
@@ -950,6 +1259,293 @@ TEST(TAsyncSlruGhostCacheTest, MoveAssignCookie)
     }
 }
 
+TEST(TAsyncSlruGhostCacheTest, OversizedValueIsTracked)
+{
+    constexpr int cacheSize = 100;
+    constexpr int oversizedValueWeight = cacheSize + 1;
+    auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TSimpleSlruCache>(std::move(config), TProfiler{"/cache"});
+
+    auto oldMainCounters = cache->ReadMainCounters();
+    auto oldSmallCounters = cache->ReadSmallGhostCounters();
+    auto oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    auto cookie = cache->BeginInsert(43);
+    ASSERT_TRUE(cookie.IsActive());
+
+    auto waitingCookie = cache->BeginInsert(43);
+    ASSERT_FALSE(waitingCookie.IsActive());
+    auto valueFuture = waitingCookie.GetValue();
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 43,
+        /*value*/ 100500,
+        /*weight*/ oversizedValueWeight);
+    cookie.EndInsert(value);
+
+    EXPECT_EQ(cache->GetSize(), 0);
+    EXPECT_EQ(WaitForFast(valueFuture).ValueOrThrow(), value);
+
+    auto mainCount = cache->ReadMainCounters() - oldMainCounters;
+    auto smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    auto largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(mainCount.AsyncHit, 1);
+    EXPECT_EQ(mainCount.AsyncHitWeight, oversizedValueWeight);
+    EXPECT_EQ(mainCount.Missed, 1);
+    EXPECT_EQ(mainCount.MissedWeight, oversizedValueWeight);
+    EXPECT_EQ(mainCount.RejectedOversized, 1);
+    EXPECT_EQ(mainCount.RejectedOversizedWeight, oversizedValueWeight);
+    EXPECT_EQ(smallCount.AsyncHit, 1);
+    EXPECT_EQ(smallCount.AsyncHitWeight, oversizedValueWeight);
+    EXPECT_EQ(smallCount.Missed, 1);
+    EXPECT_EQ(smallCount.MissedWeight, oversizedValueWeight);
+    EXPECT_EQ(smallCount.RejectedOversized, 1);
+    EXPECT_EQ(smallCount.RejectedOversizedWeight, oversizedValueWeight);
+    EXPECT_EQ(largeCount.AsyncHit, 1);
+    EXPECT_EQ(largeCount.AsyncHitWeight, oversizedValueWeight);
+    EXPECT_EQ(largeCount.Missed, 1);
+    EXPECT_EQ(largeCount.MissedWeight, oversizedValueWeight);
+    EXPECT_EQ(largeCount.RejectedOversized, 0);
+    EXPECT_EQ(largeCount.RejectedOversizedWeight, 0);
+
+    oldSmallCounters = cache->ReadSmallGhostCounters();
+    oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    YT_UNUSED_FUTURE(cache->Lookup(43));
+
+    smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(smallCount.Missed, 1);
+    EXPECT_EQ(smallCount.SyncHit, 0);
+    EXPECT_EQ(largeCount.Missed, 0);
+    EXPECT_EQ(largeCount.SyncHit, 1);
+    EXPECT_EQ(largeCount.SyncHitWeight, oversizedValueWeight);
+}
+
+TEST(TAsyncSlruGhostCacheTest, OversizedValueIsRejectedOnResurrection)
+{
+    constexpr int cacheSize = 10;
+    constexpr int oversizedValueWeight = 15;
+    auto config = CreateCacheConfig(cacheSize);
+    auto cache = New<TSimpleSlruCache>(std::move(config), TProfiler{"/cache"});
+
+    auto oversizedValue = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 100500,
+        /*weight*/ oversizedValueWeight);
+    auto cookie = cache->BeginInsert(oversizedValue->GetKey());
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(oversizedValue);
+
+    cookie = cache->BeginInsert(1);
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(New<TSimpleCachedValue>(
+        /*key*/ 1,
+        /*value*/ 42,
+        /*weight*/ 2 * cacheSize));
+
+    EXPECT_EQ(cache->GetSize(), 0);
+
+    auto dynamicConfig = New<TSlruCacheDynamicConfig>();
+    dynamicConfig->RejectOversizedItems = true;
+    cache->Reconfigure(dynamicConfig);
+
+    auto oldMainCounters = cache->ReadMainCounters();
+    auto oldSmallCounters = cache->ReadSmallGhostCounters();
+    auto oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    EXPECT_EQ(WaitForFast(cache->Lookup(oversizedValue->GetKey())).ValueOrThrow(), oversizedValue);
+    EXPECT_EQ(cache->GetSize(), 0);
+
+    auto mainCount = cache->ReadMainCounters() - oldMainCounters;
+    auto smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    auto largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(mainCount.RejectedOversized, 1);
+    EXPECT_EQ(mainCount.RejectedOversizedWeight, oversizedValueWeight);
+    EXPECT_EQ(smallCount.RejectedOversized, 1);
+    EXPECT_EQ(smallCount.RejectedOversizedWeight, oversizedValueWeight);
+    EXPECT_EQ(largeCount.RejectedOversized, 0);
+    EXPECT_EQ(largeCount.RejectedOversizedWeight, 0);
+
+    oldSmallCounters = cache->ReadSmallGhostCounters();
+    oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    YT_UNUSED_FUTURE(cache->Lookup(oversizedValue->GetKey()));
+
+    smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(smallCount.Missed, 1);
+    EXPECT_EQ(largeCount.SyncHit, 1);
+}
+
+TEST(TAsyncSlruGhostCacheTest, DisableOversizedItemRejectionDynamically)
+{
+    constexpr int cacheSize = 10;
+    constexpr int oversizedValueWeight = 15;
+    auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TSimpleSlruCache>(std::move(config), TProfiler{"/cache"});
+
+    auto dynamicConfig = New<TSlruCacheDynamicConfig>();
+    dynamicConfig->RejectOversizedItems = false;
+    cache->Reconfigure(dynamicConfig);
+
+    auto oldMainCounters = cache->ReadMainCounters();
+    auto oldSmallCounters = cache->ReadSmallGhostCounters();
+    auto oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 100500,
+        /*weight*/ oversizedValueWeight);
+    auto cookie = cache->BeginInsert(value->GetKey());
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(value);
+
+    EXPECT_EQ(cache->GetSize(), 0);
+
+    auto mainCount = cache->ReadMainCounters() - oldMainCounters;
+    auto smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    auto largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(mainCount.RejectedOversized, 0);
+    EXPECT_EQ(smallCount.RejectedOversized, 0);
+    EXPECT_EQ(largeCount.RejectedOversized, 0);
+
+    oldSmallCounters = cache->ReadSmallGhostCounters();
+    oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    YT_UNUSED_FUTURE(cache->Lookup(value->GetKey()));
+
+    smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(smallCount.Missed, 1);
+    EXPECT_EQ(largeCount.SyncHit, 1);
+}
+
+TEST(TAsyncSlruGhostCacheTest, OversizedValueAccountsForOtherCookies)
+{
+    constexpr int cacheSize = 10;
+    auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TSimpleSlruCache>(std::move(config), TProfiler{"/cache"});
+
+    auto otherCookie = cache->BeginInsert(/*key*/ 0, /*cookieWeight*/ 3);
+    ASSERT_TRUE(otherCookie.IsActive());
+
+    auto oldMainCounters = cache->ReadMainCounters();
+    auto oldSmallCounters = cache->ReadSmallGhostCounters();
+    auto oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 1,
+        /*value*/ 100500,
+        /*weight*/ 4);
+    auto cookie = cache->BeginInsert(value->GetKey());
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(value);
+
+    EXPECT_EQ(cache->Find(value->GetKey()), value);
+
+    auto mainCount = cache->ReadMainCounters() - oldMainCounters;
+    auto smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    auto largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(mainCount.RejectedOversized, 0);
+    EXPECT_EQ(smallCount.RejectedOversized, 1);
+    EXPECT_EQ(smallCount.RejectedOversizedWeight, value->Weight);
+    EXPECT_EQ(largeCount.RejectedOversized, 0);
+
+    oldSmallCounters = cache->ReadSmallGhostCounters();
+    oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    YT_UNUSED_FUTURE(cache->Lookup(value->GetKey()));
+
+    smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(smallCount.Missed, 1);
+    EXPECT_EQ(largeCount.SyncHit, 1);
+
+    otherCookie.Cancel(TError("Cancelled"));
+}
+
+TEST(TAsyncSlruGhostCacheTest, ValueAtSmallGhostCapacityIsAdmitted)
+{
+    constexpr int cacheSize = 10;
+    constexpr int smallGhostCapacity = cacheSize / 2;
+    auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TSimpleSlruCache>(std::move(config), TProfiler{"/cache"});
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 100500,
+        /*weight*/ smallGhostCapacity);
+    auto cookie = cache->BeginInsert(value->GetKey());
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(value);
+
+    auto oldSmallCounters = cache->ReadSmallGhostCounters();
+    auto oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    YT_UNUSED_FUTURE(cache->Lookup(value->GetKey()));
+
+    auto smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    auto largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(smallCount.SyncHit, 1);
+    EXPECT_EQ(smallCount.RejectedOversized, 0);
+    EXPECT_EQ(largeCount.SyncHit, 1);
+    EXPECT_EQ(largeCount.RejectedOversized, 0);
+}
+
+TEST(TAsyncSlruGhostCacheTest, ValueAtLargeGhostCapacityIsAdmitted)
+{
+    constexpr int cacheSize = 10;
+    constexpr int largeGhostCapacity = 2 * cacheSize;
+    auto config = CreateCacheConfig(cacheSize);
+    config->RejectOversizedItems = true;
+    auto cache = New<TSimpleSlruCache>(std::move(config), TProfiler{"/cache"});
+
+    auto oldMainCounters = cache->ReadMainCounters();
+    auto oldSmallCounters = cache->ReadSmallGhostCounters();
+    auto oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    auto value = New<TSimpleCachedValue>(
+        /*key*/ 0,
+        /*value*/ 100500,
+        /*weight*/ largeGhostCapacity);
+    auto cookie = cache->BeginInsert(value->GetKey());
+    ASSERT_TRUE(cookie.IsActive());
+    cookie.EndInsert(value);
+
+    auto mainCount = cache->ReadMainCounters() - oldMainCounters;
+    auto smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    auto largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(mainCount.RejectedOversized, 1);
+    EXPECT_EQ(smallCount.RejectedOversized, 1);
+    EXPECT_EQ(largeCount.RejectedOversized, 0);
+
+    oldSmallCounters = cache->ReadSmallGhostCounters();
+    oldLargeCounters = cache->ReadLargeGhostCounters();
+
+    YT_UNUSED_FUTURE(cache->Lookup(value->GetKey()));
+
+    smallCount = cache->ReadSmallGhostCounters() - oldSmallCounters;
+    largeCount = cache->ReadLargeGhostCounters() - oldLargeCounters;
+
+    EXPECT_EQ(smallCount.Missed, 1);
+    EXPECT_EQ(largeCount.SyncHit, 1);
+}
+
 TEST(TAsyncSlruGhostCacheTest, Disable)
 {
     constexpr int cacheSize = 100;