SERVER-66023 Create a DelayableTimeoutCallback class to avoid constantly resetting timers.

(cherry picked from commit 466ddbc1784709f3421ed9ccc485210b1fe94e6e)
(cherry picked from commit 87d6ec40779e3750a123efd941e16d159bf938bb)

4 files changed, 724 insertions, 0 deletions

diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 10783f244cd..3bebfe10f56 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -1612,6 +1612,7 @@ if wiredtiger:
             'abstract_async_component_test.cpp',
             'apply_ops_test.cpp',
             'check_quorum_for_config_change_test.cpp',
+            'delayable_timeout_callback_test.cpp',
             'drop_pending_collection_reaper_test.cpp',
             'idempotency_document_structure_test.cpp',
             'idempotency_update_sequence_test.cpp',
@@ -1701,6 +1702,7 @@ if wiredtiger:
             '$BUILD_DIR/mongo/util/concurrency/thread_pool',
             'abstract_async_component',
             'data_replicator_external_state_mock',
+            'delayable_timeout_callback',
             'drop_pending_collection_reaper',
             'idempotency_test_fixture',
             'idempotency_test_util',
@@ -1977,3 +1979,14 @@ env.Library(
         'oplog_entry',
     ],
 )
+
+env.Library(
+    target='delayable_timeout_callback',
+    source=[
+        'delayable_timeout_callback.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/executor/task_executor_interface',
+    ],
+)
diff --git a/src/mongo/db/repl/delayable_timeout_callback.cpp b/src/mongo/db/repl/delayable_timeout_callback.cpp
new file mode 100644
index 00000000000..d84d7099779
--- /dev/null
+++ b/src/mongo/db/repl/delayable_timeout_callback.cpp
@@ -0,0 +1,192 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kReplication
+
+#include "mongo/db/repl/delayable_timeout_callback.h"
+#include "mongo/logv2/log.h"
+
+namespace mongo {
+namespace repl {
+
+DelayableTimeoutCallback::~DelayableTimeoutCallback() {
+    cancel();
+}
+
+void DelayableTimeoutCallback::cancel() {
+    stdx::lock_guard lk(_mutex);
+    _cancel(lk);
+}
+
+void DelayableTimeoutCallback::_cancel(WithLock) {
+    if (_cbHandle) {
+        _executor->cancel(_cbHandle);
+        _cbHandle = executor::TaskExecutor::CallbackHandle();
+        _nextCall = Date_t();
+    }
+    invariant(_nextCall == Date_t());
+}
+
+Date_t DelayableTimeoutCallback::getNextCall() const {
+    stdx::lock_guard lk(_mutex);
+    return _nextCall;
+}
+
+bool DelayableTimeoutCallback::isActive() const {
+    return getNextCall() != Date_t();
+}
+
+Status DelayableTimeoutCallback::scheduleAt(Date_t when) {
+    stdx::lock_guard lk(_mutex);
+    return _scheduleAt(lk, when);
+}
+
+Status DelayableTimeoutCallback::_scheduleAt(WithLock lk, Date_t when) {
+    if (_cbHandle && when < _nextCall) {
+        LOGV2_DEBUG(6602300,
+                    3,
+                    "Moving a delayable timeout call backwards, which is inefficient",
+                    "timerName"_attr = _timerName,
+                    "when"_attr = when,
+                    "nextCall"_attr = _nextCall);
+        _cancel(lk);
+    }
+    return _delayUntil(lk, when);
+}
+
+Status DelayableTimeoutCallback::delayUntil(Date_t when) {
+    stdx::lock_guard lk(_mutex);
+    return _delayUntil(lk, when);
+}
+
+Status DelayableTimeoutCallback::_delayUntil(WithLock lk, Date_t when) {
+    if (!_cbHandle) {
+        // No timeout is active; just schedule it
+        return _reschedule(lk, when);
+    }
+    if (when == _nextCall) {
+        LOGV2_DEBUG(6602301,
+                    5,
+                    "'Rescheduling' to same time",
+                    "timerName"_attr = _timerName,
+                    "when"_attr = when,
+                    "nextCall"_attr = _nextCall);
+    }
+    _nextCall = when;
+    return Status::OK();
+}
+
+void DelayableTimeoutCallback::_handleTimeout(const executor::TaskExecutor::CallbackArgs& args) {
+    {
+        stdx::lock_guard lk(_mutex);
+        if (args.myHandle != _cbHandle) {
+            // This is normal when scheduleAt() or cancel() is used.
+            LOGV2_DEBUG(6602302,
+                        5,
+                        "DelayableTimeoutCallback::_handleTimeout got a timeout after a new handle "
+                        "was scheduled",
+                        "timerName"_attr = _timerName);
+            return;
+        }
+        Date_t now = _executor->now();
+        if (args.status == ErrorCodes::CallbackCanceled) {
+            // If args.status is CallbackCanceled yet the handles matched, that means the
+            // executor canceled the callback itself, probably as part of shutdown.  We
+            // do not want to reschedule in this case, nor call the callback.
+            _cbHandle = executor::TaskExecutor::CallbackHandle();
+            _nextCall = Date_t();
+            return;
+        } else if (_nextCall > now) {
+            Status status = _reschedule(lk, _nextCall);
+            if (!status.isOK()) {
+                LOGV2_DEBUG(6602303,
+                            2,
+                            "DelayableTimeoutCallback::_handleTimeout unable to schedule",
+                            "timerName"_attr = _timerName,
+                            "error"_attr = status);
+                fassert(6602305, status == ErrorCodes::ShutdownInProgress);
+            }
+            return;
+        }
+        _cbHandle = executor::TaskExecutor::CallbackHandle();
+        _nextCall = Date_t();
+    }
+    _callback(args);
+}
+
+Status DelayableTimeoutCallback::_reschedule(WithLock, Date_t when) {
+    // We clear _cbHandle and _nextCall in advance so if scheduleWorkAt fails for any reason
+    // (including by exception), the invariant that _cbHandle and _nextCall are clear when no
+    // callback is scheduled is maintained.
+    _cbHandle = executor::TaskExecutor::CallbackHandle();
+    _nextCall = Date_t();
+    auto cbh = _executor->scheduleWorkAt(
+        when, [this](const executor::TaskExecutor::CallbackArgs& args) { _handleTimeout(args); });
+    if (cbh == ErrorCodes::ShutdownInProgress) {
+        return cbh.getStatus();
+    }
+    _nextCall = when;
+    _cbHandle = fassert(6602304, cbh);
+    return Status::OK();
+}
+
+void DelayableTimeoutCallbackWithJitter::_resetRandomization(WithLock) {
+    _lastRandomizationTime = Date_t();
+    _currentJitter = Milliseconds(0);
+}
+
+Status DelayableTimeoutCallbackWithJitter::scheduleAt(Date_t when) {
+    stdx::lock_guard lk(_mutex);
+    _resetRandomization(lk);
+    return _scheduleAt(lk, when);
+}
+
+Status DelayableTimeoutCallbackWithJitter::delayUntil(Date_t when) {
+    stdx::lock_guard lk(_mutex);
+    _resetRandomization(lk);
+    return _delayUntil(lk, when);
+}
+
+Status DelayableTimeoutCallbackWithJitter::delayUntilWithJitter(Date_t when,
+                                                                Milliseconds jitterUpperBound) {
+    if (jitterUpperBound == Milliseconds::zero())
+        return delayUntil(when);
+    stdx::lock_guard lk(_mutex);
+    Date_t now = _getExecutor()->now();
+    Milliseconds elapsed = now - _lastRandomizationTime;
+    if (_lastRandomizationTime == Date_t() || elapsed < Milliseconds::zero() ||
+        elapsed >= jitterUpperBound || jitterUpperBound < _currentJitter) {
+        _lastRandomizationTime = now;
+        _currentJitter = Milliseconds(_randomSource(durationCount<Milliseconds>(jitterUpperBound)));
+    }
+    return _delayUntil(lk, when + _currentJitter);
+}
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/delayable_timeout_callback.h b/src/mongo/db/repl/delayable_timeout_callback.h
new file mode 100644
index 00000000000..d1cb8e29f95
--- /dev/null
+++ b/src/mongo/db/repl/delayable_timeout_callback.h
@@ -0,0 +1,159 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+#pragma once
+
+#include <string>
+
+#include "mongo/base/status.h"
+#include "mongo/executor/task_executor.h"
+#include "mongo/util/time_support.h"
+
+namespace mongo {
+namespace repl {
+
+/**
+ * The DelayableTimeoutCallback is a utility class which allows a callback to be scheduled on an
+ * executor at a given time, and then that time pushed back (later) arbitrarily often without
+ * rescheduling the call on the executor.  The callback is never called with CallbackCanceled or
+ * ShutdownInProgress.
+ *
+ * All methods are thread safe, though isActive() and getNextCall() may return stale information
+ * if external synchronization is not used.  The callback is called without any locks held.
+ */
+class DelayableTimeoutCallback {
+public:
+    /**
+     * Creates a DelayableTimeoutCallback with the given executor and callback function.  The
+     * DelayableTimeoutCallback is inactive (the callback is not scheduled) when constructed.
+     */
+    DelayableTimeoutCallback(executor::TaskExecutor* executor,
+                             executor::TaskExecutor::CallbackFn callback,
+                             std::string timerName = std::string())
+        : _executor(executor), _callback(std::move(callback)), _timerName(std::move(timerName)){};
+
+    ~DelayableTimeoutCallback();
+
+    /**
+     * If the timeout is scheduled, cancel it.  The callback function will not be called.
+     */
+    void cancel();
+
+    /**
+     * Schedule the timeout to occur at "when", regardless of if or when it is already scheduled.
+     * If it is already scheduled to occur after "when", it is canceled and rescheduled
+     *
+     * Returns status of the attempt to schedule on the executor.
+     */
+    Status scheduleAt(Date_t when);
+
+    /**
+     * Schedule the timeout to occur at "when" if it is not scheduled or scheduled to occur before
+     * "when".  If it is already scheduled to occur before "when", this call has no effect.
+     *
+     * Returns status of the attempt to schedule on the executor.
+     */
+    Status delayUntil(Date_t when);
+
+    /**
+     * Returns whether the callback is scheduled at all.
+     */
+    bool isActive() const;
+
+    /**
+     * Returns when the next call to the passed-in callback will be made, or Date_t() if inactive.
+     */
+    Date_t getNextCall() const;
+
+protected:
+    void _cancel(WithLock);
+    Status _scheduleAt(WithLock, Date_t when);
+    Status _delayUntil(WithLock, Date_t when);
+    executor::TaskExecutor* _getExecutor() {
+        return _executor;
+    }
+
+    mutable Mutex _mutex = MONGO_MAKE_LATCH("DelayableTimeoutCallback");
+
+private:
+    void _handleTimeout(const executor::TaskExecutor::CallbackArgs& cbData);
+    Status _reschedule(WithLock, Date_t when);
+
+    executor::TaskExecutor* _executor;
+    executor::TaskExecutor::CallbackHandle _cbHandle;
+    const executor::TaskExecutor::CallbackFn _callback;
+    Date_t _nextCall;
+
+    // Timer name is used only for logging.
+    const std::string _timerName;
+};
+
+/**
+ * DelayableTimeoutCallbackWithJitter is a slight variation on DelayableTimeoutCallback
+ * which adds some additional random time to delays.  Since the callback may be delayed at
+ * intervals much shorter than the random time, this would naively result in the timeout
+ * either being moved backwards often, or if we forbid moving it backwards, ending up quickly
+ * moving to the maximum jitter (which isn't very random).  To avoid that, we only recompute
+ * the jitter every maximum jitter interval -- e.g. if the max jitter is 10 seconds and we
+ * add 3 seconds jitter at time T, we will add 3 seconds jitter to every subsequent call until
+ * time T + 10.
+ *
+ * The typical purpose of the jitter is to prevent two timers receiving delay calls at the same
+ * times from firing at the same time.
+ *
+ * Synchronization of the randomSource is up to the caller; it is provided externally to
+ * avoid having a separate random number generator per timer.  The randomSource function will
+ * be called with the maximum jitter value passed to delayUntilWithJitter; it should return
+ * a value in the range [0, maxJitter) or [0, maxJitter] depending on what you want the
+ * actual jitter range to be.
+ */
+class DelayableTimeoutCallbackWithJitter : public DelayableTimeoutCallback {
+public:
+    using RandomSource = std::function<int64_t(int64_t)>;
+
+    DelayableTimeoutCallbackWithJitter(executor::TaskExecutor* executor,
+                                       executor::TaskExecutor::CallbackFn callback,
+                                       RandomSource randomSource,
+                                       std::string timerName = std::string())
+        : DelayableTimeoutCallback(executor, std::move(callback), timerName),
+          _randomSource(std::move(randomSource)) {}
+
+    Status scheduleAt(Date_t when);
+    Status delayUntil(Date_t when);
+    Status delayUntilWithJitter(Date_t when, Milliseconds maxJitter);
+
+private:
+    void _resetRandomization(WithLock);
+
+    RandomSource _randomSource;
+    Date_t _lastRandomizationTime;
+    Milliseconds _currentJitter;
+};
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/delayable_timeout_callback_test.cpp b/src/mongo/db/repl/delayable_timeout_callback_test.cpp
new file mode 100644
index 00000000000..b293d2793bb
--- /dev/null
+++ b/src/mongo/db/repl/delayable_timeout_callback_test.cpp
@@ -0,0 +1,360 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/repl/delayable_timeout_callback.h"
+#include "mongo/executor/thread_pool_task_executor_test_fixture.h"
+
+namespace mongo {
+namespace repl {
+template <typename T>
+class DelayableTimeoutCallbackBaseTest : public unittest::Test {
+protected:
+    void setUp() override {
+        auto network = std::make_unique<executor::NetworkInterfaceMock>();
+        _net = network.get();
+        _executor = makeSharedThreadPoolTestExecutor(std::move(network));
+        _executor->startup();
+        createDelayableTimeoutCallback();
+    }
+
+    void tearDown() override {
+        _delayableTimeoutCallback = boost::none;
+        _executor->shutdown();
+        _executor->join();
+        _executor.reset();
+    }
+
+    void callback(const mongo::executor::TaskExecutor::CallbackArgs& cbData) {
+        callbackRan++;
+        if (!cbData.status.isOK()) {
+            callbackRanWithError++;
+        }
+    }
+
+    void createDelayableTimeoutCallback() {
+        MONGO_UNREACHABLE;
+    }
+
+
+protected:
+    boost::optional<T> _delayableTimeoutCallback;
+    executor::NetworkInterfaceMock* _net;
+    std::shared_ptr<executor::TaskExecutor> _executor;
+    int callbackRan = 0;
+    int callbackRanWithError = 0;
+};
+
+template <>
+void DelayableTimeoutCallbackBaseTest<DelayableTimeoutCallback>::createDelayableTimeoutCallback() {
+    _delayableTimeoutCallback.emplace(
+        _executor.get(), [this](const mongo::executor::TaskExecutor::CallbackArgs& cbData) {
+            this->callback(cbData);
+        });
+}
+
+template <>
+void DelayableTimeoutCallbackBaseTest<
+    DelayableTimeoutCallbackWithJitter>::createDelayableTimeoutCallback() {
+    _delayableTimeoutCallback.emplace(
+        _executor.get(),
+        [this](const mongo::executor::TaskExecutor::CallbackArgs& cbData) {
+            this->callback(cbData);
+        },
+        [](int64_t limit) {
+            static int64_t notVeryRandom = 0;
+            notVeryRandom += 10;
+            return notVeryRandom % limit;
+        });
+}
+
+typedef DelayableTimeoutCallbackBaseTest<DelayableTimeoutCallback> DelayableTimeoutCallbackTest;
+typedef DelayableTimeoutCallbackBaseTest<DelayableTimeoutCallbackWithJitter>
+    DelayableTimeoutCallbackWithJitterTest;
+
+TEST_F(DelayableTimeoutCallbackTest, ScheduleAtSchedulesFirstCallback) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, DelayUntilSchedulesFirstCallback) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(2)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, ScheduleAtMovesCallbackLater) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+
+    ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2)));
+
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, DelayUntilMovesCallbackLater) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(2)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+
+    ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(2)));
+
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, ScheduleAtMovesCallbackEarlier) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(3)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+
+    ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(1)));
+
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, DelayUntilDoesNotMoveCallbackEarlier) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(3)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+
+    ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(1)));
+
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, ScheduleAtInPastRunsImmediately) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    // Make sure there's a past to schedule in.
+    _net->runUntil(_net->now() + Days(1));
+
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+    ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() - Seconds(1)));
+
+    // Needed to trigger anything scheduled.
+    _net->runReadyNetworkOperations();
+
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, DelayUntilInPastRunsImmediately) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    // Make sure there's a past to schedule in.
+    _net->runUntil(_net->now() + Days(1));
+
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+    ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() - Seconds(1)));
+
+    // Needed to trigger anything scheduled.
+    _net->runReadyNetworkOperations();
+
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, Cancellation) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    _delayableTimeoutCallback->cancel();
+    ASSERT_EQ(0, callbackRan);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_EQ(0, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackTest, Shutdown) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(0, callbackRan);
+    _net->runUntil(_net->now() + Seconds(1));
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+
+    _executor->shutdown();
+
+    // This makes sure the executor processes the shutdown.
+    _net->runReadyNetworkOperations();
+
+    ASSERT_EQ(0, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackWithJitterTest, DelayUntilWithJitter) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10),
+                                                              Milliseconds(100)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    // Our "random" generator is just 10,20,30,...
+    ASSERT_EQ(_net->now() + Milliseconds(10010), _delayableTimeoutCallback->getNextCall());
+    ASSERT_EQ(0, callbackRan);
+
+    // Setting it again in the same time shouldn't change jitter.
+    ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10),
+                                                              Milliseconds(100)));
+    ASSERT_EQ(_net->now() + Milliseconds(10010), _delayableTimeoutCallback->getNextCall());
+    ASSERT_EQ(0, callbackRan);
+
+    // Move forward less than the max jitter shouldn't change jitter.
+    for (int i = 0; i < 3; i++) {
+        _net->runUntil(_net->now() + Milliseconds(25));
+        ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10),
+                                                                  Milliseconds(100)));
+        ASSERT_EQ(_net->now() + Milliseconds(10010), _delayableTimeoutCallback->getNextCall());
+        ASSERT_EQ(0, callbackRan);
+    }
+
+    // Move forward to the max jitter should recalculate jitter.
+    _net->runUntil(_net->now() + Milliseconds(25));
+    ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10),
+                                                              Milliseconds(100)));
+    ASSERT_EQ(_net->now() + Milliseconds(10020), _delayableTimeoutCallback->getNextCall());
+    ASSERT_EQ(0, callbackRan);
+
+    // Setting max jitter to less than actual jitter should recalculate jitter.
+    ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10),
+                                                              Milliseconds(19)));
+    // Jitter value will be 30 % 19 = 11
+    ASSERT_EQ(_net->now() + Milliseconds(10011), _delayableTimeoutCallback->getNextCall());
+    ASSERT_EQ(0, callbackRan);
+
+    _net->runUntil(_net->now() + Milliseconds(10011));
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+TEST_F(DelayableTimeoutCallbackWithJitterTest, DelayUntilWithZeroJitter) {
+    executor::NetworkInterfaceMock::InNetworkGuard guard(_net);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+
+    ASSERT_OK(
+        _delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10), Seconds(0)));
+
+    ASSERT_TRUE(_delayableTimeoutCallback->isActive());
+    ASSERT_EQ(_net->now() + Milliseconds(10000), _delayableTimeoutCallback->getNextCall());
+    ASSERT_EQ(0, callbackRan);
+
+    _net->runUntil(_net->now() + Milliseconds(10000));
+    ASSERT_EQ(1, callbackRan);
+    ASSERT_EQ(0, callbackRanWithError);
+    ASSERT_FALSE(_delayableTimeoutCallback->isActive());
+}
+
+}  // namespace repl
+}  // namespace mongo