Use read-only TEvGetBlock requests for lease confirmations (#17479)

Author: snaury

ydb/core/base/tablet.h

@@ -64,6 +64,8 @@ struct TEvTablet {
         EvPromoteToLeader,
         EvFGcAck, // from user tablet to follower
         EvLeaseDropped,
+        EvConfirmLeader, // from user tablet to sys tablet
+        EvConfirmLeaderResult, // from sys tablet to user tablet
 
         EvTabletDead = EvBoot + 1024,
         EvFollowerUpdateState, // notifications to guardian
@@ -882,6 +884,30 @@ struct TEvTablet {
             Record.SetTabletID(tabletId);
         }
     };
+
+    struct TEvConfirmLeader : TEventLocal<TEvConfirmLeader, EvConfirmLeader> {
+        ui64 TabletID;
+        ui32 Generation;
+
+        TEvConfirmLeader(ui64 tabletId, ui32 generation)
+            : TabletID(tabletId)
+            , Generation(generation)
+        {}
+    };
+
+    struct TEvConfirmLeaderResult : TEventLocal<TEvConfirmLeaderResult, EvConfirmLeaderResult> {
+        ui64 TabletID;
+        ui32 Generation;
+        NKikimrProto::EReplyStatus Status;
+        TString ErrorReason;
+
+        TEvConfirmLeaderResult(ui64 tabletId, ui32 generation, NKikimrProto::EReplyStatus status = NKikimrProto::OK, TString errorReason = {})
+            : TabletID(tabletId)
+            , Generation(generation)
+            , Status(status)
+            , ErrorReason(std::move(errorReason))
+        {}
+    };
 };
 
 IActor* CreateTabletKiller(ui64 tabletId, ui32 nodeId = 0, ui32 maxGeneration = Max<ui32>());

ydb/core/tablet/tablet_sys.cpp

@@ -1102,6 +1102,56 @@ void TTablet::Handle(TEvTablet::TEvCommit::TPtr &ev) {
     HandleNext(ev);
 }
 
+void TTablet::Handle(TEvTablet::TEvConfirmLeader::TPtr &ev) {
+    auto *msg = ev->Get();
+
+    Y_ABORT_UNLESS(msg->TabletID == TabletID() && msg->Generation == StateStorageInfo.KnownGeneration,
+        "ev.TabletID=%ld, tablet=%ld, ev.Generation=%d, KnownGeneration=%d",
+        msg->TabletID, TabletID(), msg->Generation, StateStorageInfo.KnownGeneration);
+
+    ui64 reqId = ++ConfirmLeaderCounter;
+    ConfirmLeaderRequests[reqId] = { ev->Sender, ev->Cookie };
+
+    const auto *channelInfo = Info->ChannelInfo(0);
+    Y_ABORT_UNLESS(channelInfo, "Unexpected failure to find group for channel 0");
+    const auto *x = channelInfo->LatestEntry();
+    Y_ABORT_UNLESS(x->FromGeneration <= StateStorageInfo.KnownGeneration);
+
+    SendToBSProxy(ActorContext(), x->GroupID, new TEvBlobStorage::TEvGetBlock(TabletID(), TInstant::Max()), reqId);
+}
+
+void TTablet::Handle(TEvBlobStorage::TEvGetBlockResult::TPtr &ev) {
+    auto *msg = ev->Get();
+
+    auto it = ConfirmLeaderRequests.find(ev->Cookie);
+    if (it == ConfirmLeaderRequests.end()) {
+        BLOG_ERROR("Unexpected TEvGetBlockResult with cookie " << ev->Cookie << " without a pending request", "TSYS33");
+        return;
+    }
+
+    auto status = msg->Status;
+    if (status == NKikimrProto::OK && msg->BlockedGeneration >= StateStorageInfo.KnownGeneration) {
+        status = NKikimrProto::BLOCKED;
+    }
+
+    if (status != NKikimrProto::OK) {
+        // We want to stop after current graph is committed
+        if (BlobStorageStatus == NKikimrProto::OK) {
+            BlobStorageStatus = status;
+            BlobStorageErrorStep = Graph.NextEntry;
+            BlobStorageErrorReason = std::move(msg->ErrorReason);
+        }
+
+        CheckBlobStorageError();
+        return;
+    }
+
+    auto addr = std::move(it->second);
+    ConfirmLeaderRequests.erase(it);
+
+    Send(addr.Sender, new TEvTablet::TEvConfirmLeaderResult(TabletID(), StateStorageInfo.KnownGeneration), 0, addr.Cookie);
+}
+
 bool TTablet::HandleNext(TEvTablet::TEvCommit::TPtr &ev) {
     TEvTablet::TEvCommit *msg = ev->Get();
 
@@ -1795,6 +1845,16 @@ void TTablet::CancelTablet(TEvTablet::TEvTabletDead::EReason reason, const TStri
     if (UserTablet)
         Send(UserTablet, new TEvTablet::TEvTabletDead(TabletID(), reason, reportedGeneration));
 
+    if (BlobStorageStatus != NKikimrProto::OK) {
+        // Note: probably unnecessary after TEvTabletDead
+        for (const auto& pr : ConfirmLeaderRequests) {
+            Send(pr.second.Sender,
+                new TEvTablet::TEvConfirmLeaderResult(TabletID(), StateStorageInfo.KnownGeneration, BlobStorageStatus, BlobStorageErrorReason),
+                0, pr.second.Cookie);
+        }
+        ConfirmLeaderRequests.clear();
+    }
+
     if (StateStorageGuardian)
         Send(StateStorageGuardian, new TEvents::TEvPoisonPill());
 

ydb/core/tablet/tablet_sys.h

@@ -15,6 +15,11 @@ class TTablet : public TActor<TTablet> {
     using TTabletStateInfo = NKikimrWhiteboard::TTabletStateInfo;
     using ETabletState = TTabletStateInfo::ETabletState;
 
+    struct TRequestAddr {
+        TActorId Sender;
+        ui64 Cookie;
+    };
+
     struct TStateStorageInfo {
         TActorId ProxyID;
 
@@ -255,6 +260,10 @@ class TTablet : public TActor<TTablet> {
     TString BlobStorageErrorReason;
     bool BlobStorageErrorReported = false;
 
+    // Leader confirmation requests
+    THashMap<ui64, TRequestAddr> ConfirmLeaderRequests;
+    ui64 ConfirmLeaderCounter = 0;
+
     ui64 TabletID() const;
 
     void ReportTabletStateChange(ETabletState state);
@@ -318,6 +327,8 @@ class TTablet : public TActor<TTablet> {
 
     void Handle(TEvTablet::TEvGcForStepAckRequest::TPtr& ev);
 
+    void Handle(TEvTablet::TEvConfirmLeader::TPtr &ev);
+    void Handle(TEvBlobStorage::TEvGetBlockResult::TPtr &ev);
     void Handle(TEvTablet::TEvCommit::TPtr &ev);
     bool HandleNext(TEvTablet::TEvCommit::TPtr &ev);
     void Handle(TEvTablet::TEvAux::TPtr &ev);
@@ -512,6 +523,7 @@ class TTablet : public TActor<TTablet> {
     STATEFN(StateActivePhase) {
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvTablet::TEvCommit, Handle);
+            hFunc(TEvTablet::TEvConfirmLeader, Handle);
             hFunc(TEvTablet::TEvAux, Handle);
             hFunc(TEvTablet::TEvPreCommit, Handle);
             hFunc(TEvTablet::TEvTabletActive, HandleByLeader);
@@ -536,6 +548,7 @@ class TTablet : public TActor<TTablet> {
             hFunc(TEvInterconnect::TEvNodeDisconnected, HandleByLeader);
             hFunc(TEvents::TEvUndelivered, HandleByLeader);
             hFunc(TEvBlobStorage::TEvCollectGarbageResult, Handle);
+            hFunc(TEvBlobStorage::TEvGetBlockResult, Handle);
             hFunc(TEvTablet::TEvGcForStepAckRequest, Handle);
         }
     }

ydb/core/tablet_flat/flat_executor.cpp

@@ -1646,6 +1646,63 @@ void TExecutor::ApplyExternalPartSwitch(TPendingPartSwitch &partSwitch) {
     }
 }
 
+void TExecutor::LeaseConfirmed(ui64 confirmedCookie) {
+    bool leaseUpdated = false;
+    while (!LeaseCommits.empty()) {
+        auto& l = LeaseCommits.front();
+        if (l.Cookie <= confirmedCookie) {
+            LeaseEnd = Max(LeaseEnd, l.LeaseEnd);
+
+            auto callbacks = std::move(l.Callbacks);
+            LeaseCommitsByEnd.erase(l.ByEndIterator);
+            LeaseCommits.pop_front();
+
+            for (auto& callback : callbacks) {
+                callback();
+            }
+
+            leaseUpdated = true;
+        } else {
+            break;
+        }
+    }
+
+    if (leaseUpdated && LeaseCommits.empty()) {
+        if (LeaseDurationIncreases < 2) {
+            // Calculate how much of a lease is left after a full round trip
+            // When we are left with less than a third of lease duration we want
+            // to increase lease duration so we would have enough time for
+            // processing read-only requests without additional commits
+            TMonotonic ts = AppData()->MonotonicTimeProvider->Now();
+            if ((LeaseEnd - ts) < LeaseDuration / 3) {
+                LeaseDuration *= 2;
+                LeaseDurationUpdated = true;
+                ++LeaseDurationIncreases;
+            }
+        }
+
+        // We want to schedule a new commit before the lease expires
+        if (!LeaseExtendPending) {
+            Schedule(LeaseEnd - LeaseDuration / 3, new TEvPrivate::TEvLeaseExtend);
+            LeaseExtendPending = true;
+        }
+    }
+}
+
+TExecutor::TLeaseCommit* TExecutor::AddLeaseConfirm() {
+    if (!LeaseEnabled || Y_UNLIKELY(LeaseDropped)) {
+        return nullptr;
+    }
+
+    TMonotonic ts = AppData()->MonotonicTimeProvider->Now();
+    TLeaseCommit* lease = &LeaseCommits.emplace_back(0, ts, ts + LeaseDuration, ++LeaseCommitsCounter);
+    lease->ByEndIterator = LeaseCommitsByEnd.emplace(lease->LeaseEnd, lease);
+
+    Send(Owner->Tablet(), new TEvTablet::TEvConfirmLeader(Owner->TabletID(), Generation()), 0, lease->Cookie);
+
+    return lease;
+}
+
 TExecutor::TLeaseCommit* TExecutor::AttachLeaseCommit(TLogCommit* commit, bool force) {
     if (!LeaseEnabled || Y_UNLIKELY(LeaseDropped)) {
         return nullptr;
@@ -1661,17 +1718,17 @@ TExecutor::TLeaseCommit* TExecutor::AttachLeaseCommit(TLogCommit* commit, bool f
         Y_ENSURE(ok);
 
         commit->Metadata.emplace_back(ui32(NBoot::ELogCommitMeta::LeaseInfo), std::move(data));
-        LeaseDurationUpdated = false;
     }
 
     TMonotonic ts = AppData()->MonotonicTimeProvider->Now();
-    TLeaseCommit* lease = &LeaseCommits.emplace_back(commit->Step, ts, ts + LeaseDuration);
+    TLeaseCommit* lease = &LeaseCommits.emplace_back(commit->Step, ts, ts + LeaseDuration, ++LeaseCommitsCounter);
+    LeaseCommitsByStep.PushBack(lease);
 
     // It may happen in the future that LeaseDuration is decreased by this
     // commit, in which case new leader might read and use it, and may not wait
     // longer than the new LeaseEnd. If there are commits currently in flight
     // make sure to truncate their lease extensions to the new LeaseEnd.
-    if (force || LeaseDurationUpdated) {
+    if (LeaseDurationUpdated) {
         auto it = LeaseCommitsByEnd.upper_bound(lease->LeaseEnd);
         while (it != LeaseCommitsByEnd.end()) {
             TLeaseCommit* other = it->second;
@@ -1681,9 +1738,11 @@ TExecutor::TLeaseCommit* TExecutor::AttachLeaseCommit(TLogCommit* commit, bool f
         }
         // Currently confirmed lease may become truncated as well
         LeaseEnd = Min(LeaseEnd, lease->LeaseEnd);
+        LeaseDurationUpdated = false;
     }
 
     lease->ByEndIterator = LeaseCommitsByEnd.emplace(lease->LeaseEnd, lease);
+
     return lease;
 }
 
@@ -1707,16 +1766,22 @@ TExecutor::TLeaseCommit* TExecutor::EnsureReadOnlyLease(TMonotonic at) {
     if (itAfter != LeaseCommitsByEnd.end()) {
         lease = itAfter->second;
     } else if (!LeaseDropped) {
-        LogicRedo->FlushBatchedLog();
+        if (LeaseDurationUpdated || !LeasePersisted) {
+            // We need to make a real commit
+            LogicRedo->FlushBatchedLog();
 
-        auto commit = CommitManager->Begin(true, ECommit::Misc, {});
+            auto commit = CommitManager->Begin(true, ECommit::Misc, {});
 
-        lease = AttachLeaseCommit(commit.Get(), /* force */ true);
+            lease = AttachLeaseCommit(commit.Get(), /* force */ true);
 
-        CommitManager->Commit(commit);
+            CommitManager->Commit(commit);
 
-        if (LogicSnap->MayFlush(false)) {
-            MakeLogSnapshot();
+            if (LogicSnap->MayFlush(false)) {
+                MakeLogSnapshot();
+            }
+        } else {
+            // We want a lightweight confirmation
+            lease = AddLeaseConfirm();
         }
     }
 
@@ -3052,6 +3117,20 @@ void TExecutor::Handle(TEvPrivate::TEvLeaseExtend::TPtr &, const TActorContext &
     EnsureReadOnlyLease(LeaseEnd);
 }
 
+void TExecutor::Handle(TEvTablet::TEvConfirmLeaderResult::TPtr &ev) {
+    auto *msg = ev->Get();
+
+    // Note: lease confirmation error will currenly cause tablet to stop anyway
+    if (msg->Status != NKikimrProto::OK) {
+        if (auto logl = Logger->Log(ELnLev::Error)) {
+            logl << NFmt::Do(*this) << " Broken on lease confirmation";
+        }
+        return Broken();
+    }
+
+    LeaseConfirmed(ev->Cookie);
+}
+
 void TExecutor::Handle(TEvTablet::TEvCommitResult::TPtr &ev, const TActorContext &ctx) {
     TEvTablet::TEvCommitResult *msg = ev->Get();
 
@@ -3078,39 +3157,12 @@ void TExecutor::Handle(TEvTablet::TEvCommitResult::TPtr &ev, const TActorContext
             << " for step " << step;
     }
 
-    if (!LeaseCommits.empty()) {
-        auto& l = LeaseCommits.front();
-        Y_ENSURE(step <= l.Step);
-        if (step == l.Step) {
+    if (!LeaseCommitsByStep.Empty()) {
+        auto* l = LeaseCommitsByStep.Front();
+        Y_ENSURE(step <= l->Step);
+        if (step == l->Step) {
             LeasePersisted = true;
-            LeaseEnd = Max(LeaseEnd, l.LeaseEnd);
-
-            auto callbacks = std::move(l.Callbacks);
-            LeaseCommitsByEnd.erase(l.ByEndIterator);
-            LeaseCommits.pop_front();
-
-            for (auto& callback : callbacks) {
-                callback();
-            }
-
-            if (LeaseDurationIncreases < 2 && LeaseCommits.empty()) {
-                // Calculate how much of a lease is left after a full round trip
-                // When we are left with less than a third of lease duration we want
-                // to increase lease duration so we would have enough time for
-                // processing read-only requests without additional commits
-                TMonotonic ts = AppData()->MonotonicTimeProvider->Now();
-                if ((LeaseEnd - ts) < LeaseDuration / 3) {
-                    LeaseDuration *= 2;
-                    LeaseDurationUpdated = true;
-                    ++LeaseDurationIncreases;
-                }
-            }
-
-            // We want to schedule a new commit before the lease expires
-            if (LeaseCommits.empty() && !LeaseExtendPending) {
-                Schedule(LeaseEnd - LeaseDuration / 3, new TEvPrivate::TEvLeaseExtend);
-                LeaseExtendPending = true;
-            }
+            LeaseConfirmed(l->Cookie);
         }
     }
 
@@ -4154,6 +4206,7 @@ STFUNC(TExecutor::StateWork) {
         hFunc(NSharedCache::TEvUpdated, Handle);
         HFunc(TEvTablet::TEvDropLease, Handle);
         HFunc(TEvTablet::TEvCommitResult, Handle);
+        hFunc(TEvTablet::TEvConfirmLeaderResult, Handle);
         hFunc(TEvTablet::TEvCheckBlobstorageStatusResult, Handle);
         hFunc(TEvBlobStorage::TEvCollectGarbageResult, Handle);
         HFunc(TEvBlobStorage::TEvGetResult, Handle);