From ef76796a1d743ec8db9b6481eb357949b4bec67a Mon Sep 17 00:00:00 2001
From: Vitaliy Idelson <v.idelson@mindsw.io>
Date: Thu, 12 Nov 2026 15:09:33 +0300
Subject: [PATCH] Fix keepalive during major recalc PG

---
 src/client/messenger.cpp    | 25 +++++++++++
 src/client/messenger.h      |  2 +
 src/test/test_msgr_stop.cpp | 82 +++++++++++++++++++++++++++++++++++++
 3 files changed, 109 insertions(+)

diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp
index e6d53ac6..ea93c59c 100644
--- a/src/client/messenger.cpp
+++ b/src/client/messenger.cpp
@@ -94,8 +94,15 @@ void osd_messenger_t::init()
     {
         init_iothreads();
     }
+    clock_gettime(CLOCK_MONOTONIC, &keepalive_last);
     keepalive_timer_id = tfd->set_timer(1000, true, [this](int)
     {
+        timespec now;
+        clock_gettime(CLOCK_MONOTONIC, &now);
+        if (keepalive_catch_up(now))
+        {
+            return;
+        }
         std::vector<uint64_t> clients_to_stop;
         std::vector<osd_op_t*> ops_to_send;
         auto cl_it = clients.begin();
@@ -175,6 +182,24 @@ void osd_messenger_t::init()
     });
 }
 
+bool osd_messenger_t::keepalive_catch_up(const timespec & now)
+{
+    int64_t since_ms = (int64_t)(now.tv_sec - keepalive_last.tv_sec)*1000 +
+        (now.tv_nsec - keepalive_last.tv_nsec)/1000000;
+    keepalive_last = now;
+    if (since_ms < 100)
+    {
+        // Catch-up call of the same wakeup
+        return true;
+    }
+    if (since_ms >= 2000 && log_level > 0)
+    {
+        // The timer period is 1 s, the rest is the stall
+        fprintf(stderr, "Event loop was blocked for about %jd ms, keepalive counted it as one tick\n",
+            (intmax_t)(since_ms - 1000));
+    }
+    return false;
+}
 osd_messenger_t::~osd_messenger_t()
 {
     if (keepalive_timer_id >= 0)
diff --git a/src/client/messenger.h b/src/client/messenger.h
index ed7e05ea..2d964779 100644
--- a/src/client/messenger.h
+++ b/src/client/messenger.h
@@ -219,6 +219,7 @@ protected:
     friend struct osd_client_t;
 
     int keepalive_timer_id = -1;
+    timespec keepalive_last = {};
 
     uint32_t receive_buffer_size = 0;
     int peer_connect_interval = 0;
@@ -314,6 +315,7 @@ public:
     void connect_peer(osd_num_t osd_num, json11::Json peer_state);
     void stop_client(uint64_t client_id, bool force_delete = false);
     void destroy_client(osd_client_t *cl);
+    bool keepalive_catch_up(const timespec & now);
     void outbox_push(osd_op_t *cur_op);
     std::function<void(osd_op_t*)> exec_op;
     std::function<void(osd_num_t)> repeer_pgs;
diff --git a/src/test/test_msgr_stop.cpp b/src/test/test_msgr_stop.cpp
index adb1a487..cd8159bf 100644
--- a/src/test/test_msgr_stop.cpp
+++ b/src/test/test_msgr_stop.cpp
@@ -9,6 +9,7 @@
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <fcntl.h>
+#include <poll.h>
 #include <unistd.h>
 #include <string.h>
 #include <stdio.h>
@@ -162,8 +163,89 @@ static void test_stop_client_with_stuck_send(ring_loop_t *ringloop, bool zerocop
     printf("test_stop_client_with_stuck_send(%s) passed\n", zerocopy ? "zerocopy" : "copy");
 }
 
+static timespec ts_ms(int64_t ms)
+{
+    return (timespec){ .tv_sec = ms/1000, .tv_nsec = (ms%1000)*1000000 };
+}
+
+// Keepalive must count at most one tick per event loop wakeup (USTOR-4408)
+static void test_keepalive_catch_up()
+{
+    printf("test_keepalive_catch_up\n");
+    osd_messenger_t *msgr = new osd_messenger_t();
+    msgr->parse_config(json11::Json::object { { "log_level", 1 } });
+    struct { int64_t at_ms; bool skip; } steps[] = {
+        { 100000, false }, // first call
+        { 101000, false }, // regular 1s ticks
+        { 102000, false },
+        { 103600, false }, // tick delayed by 600 ms
+        { 104000, false }, // next on-time tick 400 ms later is counted too
+        { 115000, false }, // event loop was blocked for 11s: one tick
+        { 115000, true },  // catch-up calls of the same wakeup
+        { 115001, true },
+        { 115002, true },
+        { 116000, false }, // regular ticks again
+        { 119000, false }, // chronic stalls: one tick per stall,
+        { 119001, true },
+        { 122000, false }, // so a dead peer is still detected
+        { 122001, true },
+        { 125000, false },
+    };
+    for (auto & s: steps)
+    {
+        bool skip = msgr->keepalive_catch_up(ts_ms(s.at_ms));
+        if (skip != s.skip)
+        {
+            printf("  at %jd ms: expected %s, got %s\n", (intmax_t)s.at_ms,
+                s.skip ? "skip" : "count", skip ? "skip" : "count");
+            assert(skip == s.skip);
+        }
+    }
+    delete msgr;
+}
+
+// The same with the real timerfd_manager: after a 3s stall it fires the 1s timer
+// several times in one wakeup, only one of them must be counted
+static void test_keepalive_after_stall()
+{
+    printf("test_keepalive_after_stall\n");
+    int timer_fd = -1;
+    std::function<void(int, int)> timer_handler;
+    timerfd_manager_t *tfd = new timerfd_manager_t([&](int fd, bool wr, std::function<void(int, int)> cb)
+    {
+        timer_fd = fd;
+        timer_handler = cb;
+    });
+    osd_messenger_t *msgr = new osd_messenger_t();
+    // As init() does
+    timespec now;
+    clock_gettime(CLOCK_MONOTONIC, &now);
+    msgr->keepalive_catch_up(now);
+    int fired = 0, counted = 0;
+    tfd->set_timer(1000, true, [&](int)
+    {
+        timespec now;
+        clock_gettime(CLOCK_MONOTONIC, &now);
+        fired++;
+        if (!msgr->keepalive_catch_up(now))
+            counted++;
+    });
+    sleep(3);
+    pollfd pfd = { .fd = timer_fd, .events = POLLIN };
+    int r = poll(&pfd, 1, 2000);
+    assert(r == 1);
+    timer_handler(timer_fd, POLLIN);
+    printf("  timer fired %d times in one wakeup, %d counted\n", fired, counted);
+    assert(fired >= 2);
+    assert(counted == 1);
+    delete msgr;
+    delete tfd;
+}
+
 int main(int narg, char *args[])
 {
+    test_keepalive_catch_up();
+    test_keepalive_after_stall();
     ring_loop_t *ringloop;
     try
     {
-- 
2.47.3