Skip to content

Hopefully fix flaky overlay test. #5365

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
graydon wants to merge 1 commit into
base: master
Choose a base branch
from
+38 −52
Open

Hopefully fix flaky overlay test. #5365

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 38 additions & 52 deletions src/overlay/test/OverlayTests.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -545,7 +545,7 @@ TEST_CASE("flow control byte capacity", "[overlay][flowcontrol]")
}

TEST_CASE("flow control total byte capacity throttles non-flood traffic",
"[overlay][flowcontrol][!hide]")
"[overlay][flowcontrol]")
{
SCPQuorumSet qSet;
qSet.threshold = 1;
Expand All @@ -563,7 +563,7 @@ TEST_CASE("flow control total byte capacity throttles non-flood traffic",

auto const msgSize = FlowControlCapacity::msgBodySize(*scpQSetMsg);
uint64_t constexpr MESSAGES_TO_FLOOR = 3;
uint64_t constexpr MESSAGES_PER_BURST = MESSAGES_TO_FLOOR + 1;
size_t constexpr NUM_CYCLES = 100;
auto const byteCapacity = msgSize * MESSAGES_TO_FLOOR - 1;
auto const totalCapacity =
FlowControlByteCapacity::BYTE_CAPACITY_READ_FLOOR + byteCapacity;
Expand Down Expand Up @@ -621,33 +621,21 @@ TEST_CASE("flow control total byte capacity throttles non-flood traffic",
.getOverlayMetrics()
.mConnectionReadThrottle.count();
};
auto waitFor = [&](std::string const& phase, auto&& predicate) {
auto timeout = std::chrono::seconds(10);
auto start = std::chrono::steady_clock::now();

while (!predicate() &&
std::chrono::steady_clock::now() - start <= timeout)
{
if (simulation->crankAllNodes() == 0)
{
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
auto getRecvCount = [&]() {
return app2->getOverlayManager()
.getOverlayMetrics()
.mRecvSCPQuorumSetTimer.count();
};

auto capacity = getReceiverCapacity();
CAPTURE(phase);
CAPTURE(capacity.mTotalCapacity);
CAPTURE(totalCapacity);
CAPTURE(capacity.mFloodCapacity);
CAPTURE(receiver->getFlowControl()->isThrottled());
CAPTURE(receiver->getFlowControl()->canRead());
CAPTURE(sender->isConnectedForTesting());
CAPTURE(receiver->isConnectedForTesting());
CAPTURE(getReadThrottleCount());
simulation->crankUntil(
[&]() {
return getReceiverCapacity().mTotalCapacity == totalCapacity &&
!receiver->getFlowControl()->isThrottled();
},
std::chrono::seconds(10), false);
auto const initialCapacity = getReceiverCapacity();

REQUIRE(predicate());
};
auto waitForReceiverThrottle = [&]() {
auto waitForThrottle = [&](size_t cycle) {

@marta-lokhova marta-lokhova Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is std::this_thread::sleep_for(std::chrono::milliseconds(1)); correct here? Does it work with Simulation class, which cranks virtual time separately?

It's not a common pattern across our test suite, so I wanted to confirm that we're confident this is the right fix.

@marta-lokhova marta-lokhova Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@graydon I don't want to block this change since CI is unstable - we can merge as-is if you want. However, I'm still not sure if we should sleep at all. Wouldn't anything time-based like this continue to cause flakiness depending on how fast messages get processed/tasks are scheduled on main?

One way around that is to make the test simpler: we can just let it crank for a while, then check in metrics that core throttled connections the number of times we expect and that peers are still connected. Of course, we can't assert concrete checks like "reading capacity is N", but presumably the invariance in the code already does the heavy lifting for us (e.g. capacity can't go negative, capacity can't go over the limit, etc). I would only add an assert in the test that reading capacity is higher than flood capacity, to make sure that if we there's a bug and flood capacity is touched for some reason, we trigger an error of sorts. Up to you though.

auto timeout = std::chrono::seconds(10);
auto start = std::chrono::steady_clock::now();

Expand All @@ -658,6 +646,7 @@ TEST_CASE("flow control total byte capacity throttles non-flood traffic",
}

auto capacity = getReceiverCapacity();
CAPTURE(cycle);
CAPTURE(capacity.mTotalCapacity);
CAPTURE(totalCapacity);
CAPTURE(capacity.mFloodCapacity);
Expand All @@ -668,44 +657,41 @@ TEST_CASE("flow control total byte capacity throttles non-flood traffic",
CAPTURE(getReadThrottleCount());

REQUIRE(receiver->getFlowControl()->isThrottled());
};
auto drainReceiver = [&]() {
waitFor("receiver drains", [&]() {
return getReceiverCapacity().mTotalCapacity == totalCapacity &&
!receiver->getFlowControl()->isThrottled();
});
REQUIRE(getReceiverCapacity().mTotalCapacity == totalCapacity);
REQUIRE(!receiver->getFlowControl()->isThrottled());
};

drainReceiver();
auto const initialFloodCapacity = getReceiverCapacity().mFloodCapacity;
auto requireReceiverTotalCapacity = [&](uint64_t total) {
auto capacity = getReceiverCapacity();
REQUIRE(capacity.mTotalCapacity == total);
REQUIRE(capacity.mFloodCapacity == initialFloodCapacity);
REQUIRE(capacity.mTotalCapacity ==
FlowControlByteCapacity::BYTE_CAPACITY_READ_FLOOR - 1);
REQUIRE(capacity.mFloodCapacity == initialCapacity.mFloodCapacity);
};

requireReceiverTotalCapacity(totalCapacity);
REQUIRE(initialCapacity.mTotalCapacity == totalCapacity);
REQUIRE(receiver->getFlowControl()->canRead());
REQUIRE(!receiver->getFlowControl()->isThrottled());
REQUIRE(initialCapacity.mFloodCapacity ==
app2->getOverlayManager().getFlowControlFloodByteCapacity());

for (size_t i = 0; i < 100; ++i)
for (size_t cycle = 0; cycle < NUM_CYCLES; ++cycle)
{
CAPTURE(i);
drainReceiver();
CAPTURE(cycle);
auto const initialReadThrottleCount = getReadThrottleCount();
CAPTURE(initialReadThrottleCount);
for (size_t j = 0; j < MESSAGES_PER_BURST; ++j)
auto const initialRecvCount = getRecvCount();
for (size_t i = 0; i < MESSAGES_TO_FLOOR; ++i)
{
sender->sendMessage(scpQSetMsg);
}

waitForReceiverThrottle();
drainReceiver();
waitForThrottle(cycle);
simulation->crankUntil(
[&]() {
return getRecvCount() == initialRecvCount + MESSAGES_TO_FLOOR;

@marta-lokhova marta-lokhova Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Isn't this a race? The metric is updated before msg destructor is called, which releases capacity.

},
std::chrono::seconds(10), false);

auto const finalCapacity = getReceiverCapacity();
REQUIRE(getRecvCount() == initialRecvCount + MESSAGES_TO_FLOOR);
REQUIRE(getReadThrottleCount() > initialReadThrottleCount);
requireReceiverTotalCapacity(totalCapacity);
REQUIRE(finalCapacity.mTotalCapacity == initialCapacity.mTotalCapacity);
REQUIRE(finalCapacity.mFloodCapacity == initialCapacity.mFloodCapacity);
REQUIRE(receiver->getFlowControl()->canRead());
REQUIRE(!receiver->getFlowControl()->isThrottled());
REQUIRE(sender->isConnectedForTesting());
REQUIRE(receiver->isConnectedForTesting());
}
Expand Down
Loading