Sync branches 260115-0051

8 files changed, 95 insertions, 29 deletions

diff --git a/library/cpp/mime/types/mime.cpp b/library/cpp/mime/types/mime.cpp
index 144ae392b27..993c7039106 100644
--- a/library/cpp/mime/types/mime.cpp
+++ b/library/cpp/mime/types/mime.cpp
@@ -103,6 +103,7 @@ const TMimeTypes::TRecord TMimeTypes::Records[] = {
     {MIME_CSV, "text/csv\0", "csv\0"},
     {MIME_VIDEO_MP4, "video/mp4\0", "mp4\0"},
     {MIME_VIDEO_AVI, "video/x-msvideo\0", "avi\0"},
+    {MIME_IMAGE_AVIF, "image/avif\0", "avif\0"},
     {MIME_MAX, nullptr, nullptr},
 
     // Additional records
@@ -258,4 +259,5 @@ const char* MimeNames[MIME_MAX] = {
     "csv",     // MIME_CSV             // 48
     "mp4",     // MIME_VIDEO_MP4       // 49
     "avi",     // MIME_VIDEO_AVI       // 50
+    "avif",    // MIME_IMAGE_AVIF      // 51
 };
diff --git a/library/cpp/mime/types/mime.h b/library/cpp/mime/types/mime.h
index 8eaedd001b6..e4fd024a98c 100644
--- a/library/cpp/mime/types/mime.h
+++ b/library/cpp/mime/types/mime.h
@@ -62,6 +62,7 @@ enum MimeTypes {
     MIME_CSV = 48,
     MIME_VIDEO_MP4 = 49,
     MIME_VIDEO_AVI = 50,
+    MIME_IMAGE_AVIF = 51,
     MIME_MAX
 };
 
diff --git a/library/cpp/threading/hot_swap/hot_swap.cpp b/library/cpp/threading/hot_swap/hot_swap.cpp
index f8127ae45f8..ba2601e2b0d 100644
--- a/library/cpp/threading/hot_swap/hot_swap.cpp
+++ b/library/cpp/threading/hot_swap/hot_swap.cpp
@@ -4,18 +4,17 @@
 
 namespace NHotSwapPrivate {
     void TWriterLock::Acquire() noexcept {
-        AtomicIncrement(ReadersCount);
+        ++ReadersCount;
     }
 
     void TWriterLock::Release() noexcept {
-        AtomicDecrement(ReadersCount);
+        --ReadersCount;
     }
 
     void TWriterLock::WaitAllReaders() const noexcept {
-        TAtomicBase cnt = AtomicGet(ReadersCount);
-        while (cnt > 0) {
+        for (i32 cnt = ReadersCount.load(); cnt > 0;) {
             SpinLockPause();
-            cnt = AtomicGet(ReadersCount);
+            cnt = ReadersCount.load();
             Y_ASSERT(cnt >= 0);
         }
     }
diff --git a/library/cpp/threading/hot_swap/hot_swap.h b/library/cpp/threading/hot_swap/hot_swap.h
index e155614b533..5bee1ae25ab 100644
--- a/library/cpp/threading/hot_swap/hot_swap.h
+++ b/library/cpp/threading/hot_swap/hot_swap.h
@@ -2,12 +2,12 @@
 
 #include <util/generic/cast.h>
 #include <util/generic/ptr.h>
-#include <util/generic/utility.h>
-#include <library/cpp/deprecated/atomic/atomic.h>
 #include <util/system/guard.h>
 #include <util/system/spinlock.h>
 #include <util/system/yassert.h>
 
+#include <atomic>
+
 namespace NHotSwapPrivate {
     // Special guard object for THotSwap
     class TWriterLock {
@@ -19,7 +19,7 @@ namespace NHotSwapPrivate {
         void WaitAllReaders() const noexcept;
 
     private:
-        TAtomic ReadersCount = 0;
+        std::atomic<i32> ReadersCount = 0;
     };
 
 }
@@ -52,8 +52,7 @@ public:
     using TPtr = TIntrusivePtr<T, Ops>;
 
 public:
-    THotSwap() noexcept {
-    }
+    THotSwap() noexcept = default;
 
     explicit THotSwap(T* p) noexcept {
         AtomicStore(p);
@@ -70,7 +69,8 @@ public:
     }
 
     THotSwap(THotSwap&& other) noexcept {
-        DoSwap(RawPtr, other.RawPtr); // we don't need thread safety, because both objects are local
+        RawPtr.store(other.RawPtr.load()); // we don't need thread safety, because both objects are local
+        other.RawPtr.store(0);
     }
 
     ~THotSwap() noexcept {
@@ -86,7 +86,7 @@ public:
     ///
     /// @returns          Current value of stored object
     TPtr AtomicLoad() const noexcept {
-        const TAtomicBase lockIndex = GetLockIndex();
+        const ui32 lockIndex = GetLockIndex();
         auto guard = Guard(WriterLocks[lockIndex]); // non-blocking (for other AtomicLoad()'s) guard
         return GetRawPtr();
     }
@@ -105,24 +105,24 @@ public:
 
 private:
     T* GetRawPtr() const noexcept {
-        return reinterpret_cast<T*>(AtomicGet(RawPtr));
+        return reinterpret_cast<T*>(RawPtr.load());
     }
 
-    TAtomicBase GetLockIndex() const noexcept {
-        return AtomicGet(LockIndex);
+    ui32 GetLockIndex() const noexcept {
+        return LockIndex.load();
     }
 
-    TAtomicBase SwitchLockIndex() noexcept; // returns previous index value
+    ui32 SwitchLockIndex() noexcept; // returns previous index value
     void SwitchRawPtr(T* from, T* to) noexcept;
     void WaitReaders() noexcept;
 
 private:
-    TAtomic RawPtr = 0; // T* // Pointer to current value
-    static_assert(sizeof(TAtomic) == sizeof(T*), "TAtomic can't represent a pointer value");
+    std::atomic<intptr_t> RawPtr = 0; // T* // Pointer to current value
+    static_assert(sizeof(std::atomic<intptr_t>) == sizeof(T*), "std::atomic<intptr_t> can't represent a pointer value");
 
     TAdaptiveLock UpdateMutex;                           // Guarantee that AtomicStore() will be one at a time
     mutable NHotSwapPrivate::TWriterLock WriterLocks[2]; // Guarantee that AtomicStore() will wait for all concurrent AtomicLoad()'s completion
-    TAtomic LockIndex = 0;
+    std::atomic<ui32> LockIndex = 0;
 };
 
 // Atomic operations of AtomicLoad:
@@ -167,10 +167,10 @@ void THotSwap<T, Ops>::AtomicStore(T* p) noexcept {
 }
 
 template <class T, class Ops>
-TAtomicBase THotSwap<T, Ops>::SwitchLockIndex() noexcept {
-    const TAtomicBase prevIndex = AtomicGet(LockIndex);
+ui32 THotSwap<T, Ops>::SwitchLockIndex() noexcept {
+    const ui32 prevIndex = LockIndex.load();
     Y_ASSERT(prevIndex == 0 || prevIndex == 1);
-    AtomicSet(LockIndex, prevIndex ^ 1);
+    LockIndex.store(prevIndex ^ 1u);
     return prevIndex;
 }
 
@@ -185,7 +185,7 @@ void THotSwap<T, Ops>::SwitchRawPtr(T* from, T* to) noexcept {
     if (to)
         Ops::Ref(to); // Ref() for new value
 
-    AtomicSet(RawPtr, reinterpret_cast<TAtomicBase>(to));
+    RawPtr.store(reinterpret_cast<intptr_t>(to));
 
     if (from)
         Ops::UnRef(from); // Unref() for old value
diff --git a/library/cpp/threading/hot_swap/ya.make b/library/cpp/threading/hot_swap/ya.make
index 73cc1694117..e86105d59dc 100644
--- a/library/cpp/threading/hot_swap/ya.make
+++ b/library/cpp/threading/hot_swap/ya.make
@@ -5,8 +5,4 @@ SRCS(
     hot_swap.cpp
 )
 
-PEERDIR(
-    library/cpp/deprecated/atomic
-)
-
 END()
diff --git a/library/cpp/unified_agent_client/atomic_throttlers.h b/library/cpp/unified_agent_client/atomic_throttlers.h
new file mode 100644
index 00000000000..293c4429627
--- /dev/null
+++ b/library/cpp/unified_agent_client/atomic_throttlers.h
@@ -0,0 +1,48 @@
+#pragma once
+
+#include <atomic>
+#include <chrono>
+
+namespace NUnifiedAgent {
+
+class AtomicOneSecondThrottler {
+   public:
+    // The current time in seconds by AtomicOneSecondThrottler version.
+    uint64_t static inline Now() {
+        return std::chrono::duration_cast<std::chrono::seconds>(
+                   std::chrono::steady_clock::now().time_since_epoch())
+                   .count()
+               // Add one second, because the steady clock starts after the
+               // system reboot, and at the first moment, it's zero, and the
+               // last operation is also zero.
+               + 1;
+    }
+    // Check if we may do the operation.
+    inline bool May() noexcept {
+        return LastOp_.load(std::memory_order_acquire) < Now();
+    }
+    //  Check if we can do it right now. Book the limits.
+    inline bool Do() noexcept {
+        auto now = Now();
+        auto last_op_expected = LastOp_.load(std::memory_order_acquire);
+        // Early check that we can do operation.
+        if (last_op_expected >= now) {
+            // It means that somebody has already taken our second, or maybe
+            // this thread is stuck in the past.
+            return false;
+        }
+        // Place our second in the LastOp_,
+        // if false, that means somebody replaced last_op_expected value.
+        return LastOp_.compare_exchange_strong(last_op_expected, now,
+                                               std::memory_order_acq_rel);
+    }
+
+   private:
+    // Last operation time
+    std::atomic<uint64_t> LastOp_{0};
+    static_assert(decltype(LastOp_)::is_always_lock_free,
+                  "AtomicOneSecondThrottler::LastOp_ must be always lock-free");
+};
+
+}  // namespace NUnifiedAgent
+
diff --git a/library/cpp/unified_agent_client/client_impl.cpp b/library/cpp/unified_agent_client/client_impl.cpp
index bd2859f4661..2e350d9a133 100644
--- a/library/cpp/unified_agent_client/client_impl.cpp
+++ b/library/cpp/unified_agent_client/client_impl.cpp
@@ -1,5 +1,6 @@
 #include "client_impl.h"
 #include "helpers.h"
+#include "atomic_throttlers.h"
 
 #include <contrib/libs/grpc/include/grpc/grpc.h>
 #include <contrib/libs/grpc/src/core/lib/gpr/string.h>
@@ -531,13 +532,26 @@ namespace NUnifiedAgent::NPrivate {
             }
             if (InflightBytes.load() + messageSize > MaxInflightBytes) {
                 g.Release();
-                YLOG_ERR(Sprintf("max inflight of [%zu] bytes reached, [%zu] bytes dropped",
-                    MaxInflightBytes, messageSize));
+
+                // Update counters first before potentially slow logging operations
                 --Counters->InflightMessages;
                 Counters->InflightBytes -= messageSize;
                 ++Counters->DroppedMessages;
                 Counters->DroppedBytes += messageSize;
                 ++Counters->ErrorsCount;
+
+                // Check log level and increment error counter before throttling check.
+                // We use non-silent mode (false) to ensure ALL errors are counted in metrics,
+                // even if they don't get logged due to throttling. This provides accurate
+                // monitoring of the real error rate.
+                if (auto* log = Logger.Accept(TLOG_ERR, false); log != nullptr) {
+                    if (InflightErrorThrottler.Do()) {
+                        Logger.Log(*log, TLOG_ERR,
+                            std::format("max inflight of [{}] bytes reached, [{}] bytes dropped",
+                                MaxInflightBytes, messageSize));
+                    }
+                }
+
                 return;
             }
             InflightBytes.fetch_add(messageSize);
@@ -1281,3 +1295,4 @@ namespace NUnifiedAgent {
         return MakeIntrusive<NPrivate::TClient>(parameters, forkProtector);
     }
 }
+
diff --git a/library/cpp/unified_agent_client/client_impl.h b/library/cpp/unified_agent_client/client_impl.h
index aa9d808b6e7..f590f4ab865 100644
--- a/library/cpp/unified_agent_client/client_impl.h
+++ b/library/cpp/unified_agent_client/client_impl.h
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <library/cpp/unified_agent_client/atomic_throttlers.h>
 #include <library/cpp/unified_agent_client/client.h>
 #include <library/cpp/unified_agent_client/client_proto_weighing.h>
 #include <library/cpp/unified_agent_client/counters.h>
@@ -279,6 +280,10 @@ namespace NUnifiedAgent::NPrivate {
         TAdaptiveLock Lock;
         size_t MaxInflightBytes;
         TFMaybe<size_t> AgentMaxReceiveMessage;
+        // Per-session throttler for max inflight error logging.
+        // Must be per-session (not static) to ensure each session can log
+        // independently during test iterations where sessions are created/destroyed rapidly.
+        AtomicOneSecondThrottler InflightErrorThrottler;
     };
 
     class TGrpcCall final: public TAtomicRefCount<TGrpcCall> {