Skip to content

Fix cluster rejoin circuit in IC #24237

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

Merged
merged 1 commit into from
Sep 5, 2025
Merged

Fix cluster rejoin circuit in IC #24237

Show file tree
Hide file tree
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
45 changes: 34 additions & 11 deletions ydb/core/blobstorage/bridge/proxy/bridge_proxy.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -639,7 +639,7 @@ namespace NKikimr {
TBridgeInfo::TPtr BridgeInfo;

std::deque<std::unique_ptr<IEventHandle>> PendingQ;
std::map<ui32, std::deque<std::unique_ptr<IEventHandle>>> PendingByGeneration;
std::deque<std::tuple<TMonotonic, std::unique_ptr<IEventHandle>>> PendingForNextGeneration;

public:
TBridgedBlobStorageProxyActor(TIntrusivePtr<TBlobStorageGroupInfo> info)
Expand All @@ -650,6 +650,32 @@ namespace NKikimr {
void Bootstrap() {
Send(MakeBlobStorageNodeWardenID(SelfId().NodeId()), new TEvNodeWardenQueryStorageConfig(/*subscribe=*/ true));
Become(&TThis::StateWaitBridgeInfo);
HandleWakeup();
}

void HandleWakeup() {
TMonotonic dropBefore = TActivationContext::Monotonic() - TDuration::Seconds(2);
while (!PendingForNextGeneration.empty()) {
if (auto& [timestamp, ev] = PendingForNextGeneration.front(); timestamp < dropBefore) {
switch (ev->GetTypeRewrite()) {
#define MAKE_ERROR(TYPE) \
case TYPE::EventType: \
Send(ev->Sender, static_cast<TYPE*>(ev->GetBase())->MakeErrorResponse(NKikimrProto::ERROR, \
"bridge request timed out", GroupId), 0, ev->Cookie); \
break;

DSPROXY_ENUM_EVENTS(MAKE_ERROR)
#undef MAKE_ERROR
default:
Y_ABORT();
}
PendingForNextGeneration.pop_front();
} else {
break;
}
}
TActivationContext::Schedule(TDuration::Seconds(1), new IEventHandle(TEvents::TSystem::Wakeup, 0, SelfId(),
{}, nullptr, 0));
}

void PassAway() override {
Expand Down Expand Up @@ -815,7 +841,7 @@ namespace NKikimr {
const ui32 myGeneration = bridgeGroupState.GetPile(pile.BridgePileId.GetPileIndex()).GetGroupGeneration();

if (myGeneration < msg->RacingGeneration) {
PendingByGeneration[Info->GroupGeneration + 1].push_back(std::move(handle));
PendingForNextGeneration.emplace_back(TActivationContext::Monotonic(), std::move(handle));
} else if (msg->RacingGeneration < myGeneration) {
// our generation is higher than the recipient's; we have to route this message through node warden
// to ensure proxy's configuration gets in place
Expand Down Expand Up @@ -879,17 +905,13 @@ namespace NKikimr {
}

void Handle(TEvBlobStorage::TEvConfigureProxy::TPtr ev) {
Info = std::move(ev->Get()->Info);
while (!PendingByGeneration.empty()) {
auto it = PendingByGeneration.begin();
auto& [requiredGeneration, events] = *it;
if (Info->GroupGeneration < requiredGeneration) {
break;
}
for (auto& ev : events) {
auto prevInfo = std::exchange(Info, std::move(ev->Get()->Info));
Y_ABORT_UNLESS(prevInfo);
Y_ABORT_UNLESS(Info);
if (prevInfo->GroupGeneration < Info->GroupGeneration) {
for (auto& [timestamp, ev] : std::exchange(PendingForNextGeneration, {})) {
TActivationContext::Send(ev.release());
}
PendingByGeneration.erase(it);
}
}

Expand Down Expand Up @@ -923,6 +945,7 @@ namespace NKikimr {
hFunc(TEvBlobStorage::TEvConfigureProxy, Handle)
hFunc(TEvNodeWardenStorageConfig, Handle)
cFunc(TEvents::TSystem::Poison, PassAway)
cFunc(TEvents::TSystem::Wakeup, HandleWakeup)
)

#undef HANDLE_RESULT
Expand Down
63 changes: 29 additions & 34 deletions ydb/core/blobstorage/nodewarden/distconf_connectivity.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,51 +260,46 @@ namespace NKikimr::NStorage {
return TStringBuilder() << "peer storage config invalid: " << *error;
} else if (auto error = ValidateClusterState(*StorageConfig)) {
return TStringBuilder() << "local storage config invalid: " << *error;
}

// local/peer side is returning from DISCONNECTED state, validate cluster histories to ensure there were no
// definite split brain
if (auto error = ValidateClusterStateDetails(config)) {
} else if (auto error = ValidateClusterStateDetails(config)) {
return TStringBuilder() << "peer cluster state history invalid: " << *error;
} else if (auto error = ValidateClusterStateDetails(*StorageConfig)) {
return TStringBuilder() << "local cluster state history invalid: " << *error;
} else if (auto error = CheckHistoryCompatibility(*StorageConfig, config)) {
// histories are incompatible, connection won't ever be possible
return error;
}

const auto *peerPile = BridgeInfo->GetPile(peerBridgePileId);
if (peerPile == BridgeInfo->SelfNodePile) {
// no extra checks when connecting nodes from the same pile
return std::nullopt;
}
std::optional<TString> error;

if (StorageConfig->GetGeneration() == config.GetGeneration() && StorageConfig->GetFingerprint() != config.GetFingerprint()) {
return "config fingerprint mismatch";
}
const auto *peerPile = BridgeInfo->GetPile(peerBridgePileId);
if (peerPile != BridgeInfo->SelfNodePile) { // extra checks when connecting between different piles
if (auto error = CheckHistoryCompatibility(*StorageConfig, config)) {
// histories are incompatible, connection won't ever be possible
return error;
}

const NKikimrBlobStorage::TStorageConfig& newerConfig =
StorageConfig->GetGeneration() < config.GetGeneration()
? config
: *StorageConfig;
if (StorageConfig->GetGeneration() == config.GetGeneration() && StorageConfig->GetFingerprint() != config.GetFingerprint()) {
return "config fingerprint mismatch";
}

const auto& cs = newerConfig.GetClusterState();
const NKikimrBlobStorage::TStorageConfig& newerConfig =
StorageConfig->GetGeneration() < config.GetGeneration()
? config
: *StorageConfig;

std::optional<TString> error;
const auto& cs = newerConfig.GetClusterState();

if (!NBridge::PileStateTraits(cs.GetPerPileState(peerBridgePileId.GetPileIndex())).AllowsConnection) {
error = "peer is not allowed to connect";
} else if (!NBridge::PileStateTraits(cs.GetPerPileState(SelfBridgePileId.GetPileIndex())).AllowsConnection) {
error = "local node is not allowed to accept peer";
}
if (!NBridge::PileStateTraits(cs.GetPerPileState(peerBridgePileId.GetPileIndex())).AllowsConnection) {
error = "peer is not allowed to connect";
} else if (!NBridge::PileStateTraits(cs.GetPerPileState(SelfBridgePileId.GetPileIndex())).AllowsConnection) {
error = "local node is not allowed to accept peer";
}

if (!error) {
const auto& myClusterState = StorageConfig->GetClusterState();
const auto& peerClusterState = config.GetClusterState();
if (myClusterState.GetGeneration() < peerClusterState.GetGeneration()) {
error = "local cluster state is obsolete";
} else if (peerClusterState.GetGeneration() < myClusterState.GetGeneration()) {
error = "peer cluster state is obsolete";
if (!error) {
const auto& myClusterState = StorageConfig->GetClusterState();
const auto& peerClusterState = config.GetClusterState();
if (myClusterState.GetGeneration() < peerClusterState.GetGeneration()) {
error = "local cluster state is obsolete";
} else if (peerClusterState.GetGeneration() < myClusterState.GetGeneration()) {
error = "peer cluster state is obsolete";
}
}
}

Expand Down
2 changes: 1 addition & 1 deletion ydb/core/blobstorage/nodewarden/node_warden_pdisk.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -576,7 +576,7 @@ namespace NKikimr::NStorage {
Y_ABORT_UNLESS(PDiskKey.NodeId == SelfId().NodeId());
Send(MakeBlobStoragePDiskID(PDiskKey.NodeId, PDiskKey.PDiskId), ConvertedEv.release(),
IEventHandle::FlagTrackDelivery);
Become(&TThis::StateFunc, TDuration::Seconds(10), new TEvents::TEvWakeup);
Become(&TThis::StateFunc, TDuration::Seconds(30), new TEvents::TEvWakeup);
}

void Handle(TEvents::TEvUndelivered::TPtr /*ev*/) {
Expand Down
Loading