Merge 11640a8 into ffa96b1

Author: ijon

ydb/core/testlib/tablet_helpers.cpp

@@ -1151,6 +1151,7 @@ namespace NKikimr {
  HFunc(TEvHive::TEvAdoptTablet, Handle);
  HFunc(TEvHive::TEvDeleteTablet, Handle);
  HFunc(TEvHive::TEvDeleteOwnerTablets, Handle);
+ HFunc(TEvHive::TEvStopTablet, Handle);
  HFunc(TEvHive::TEvRequestHiveInfo, Handle);
  HFunc(TEvHive::TEvInitiateTabletExternalBoot, Handle);
  HFunc(TEvHive::TEvUpdateTabletsObject, Handle);
@@ -1179,7 +1180,7 @@ namespace NKikimr {
 
  void Handle(TEvHive::TEvCreateTablet::TPtr& ev, const TActorContext& ctx) {
  LOG_INFO_S(ctx, NKikimrServices::HIVE, "[" << TabletID() << "] TEvCreateTablet, msg: " << ev->Get()->Record.ShortDebugString());
- Cout << "FAKEHIVE " << TabletID() << " TEvCreateTablet " << ev->Get()->Record.ShortDebugString() << Endl;
+ Cerr << "FAKEHIVE " << TabletID() << " TEvCreateTablet " << ev->Get()->Record.ShortDebugString() << Endl;
  NKikimrProto::EReplyStatus status = NKikimrProto::OK;
  const std::pair<ui64, ui64> key(ev->Get()->Record.GetOwner(), ev->Get()->Record.GetOwnerIdx());
  const auto type = ev->Get()->Record.GetTabletType();
@@ -1256,6 +1257,9 @@ namespace NKikimr {
  auto& boundChannels = ev->Get()->Record.GetBindedChannels();
  it->second.BoundChannels.assign(boundChannels.begin(), boundChannels.end());
  it->second.ChannelsProfile = ev->Get()->Record.GetChannelsProfile();
+
+ it->second.State = ETabletState::ReadyToWork;
+ it->second.ObjectDomain = TSubDomainKey(ev->Get()->Record.GetObjectDomain());
  }
 
  ctx.Send(ev->Sender, new TEvHive::TEvCreateTabletReply(status, key.first,
@@ -1355,7 +1359,7 @@ namespace NKikimr {
  void Handle(TEvHive::TEvDeleteTablet::TPtr &ev, const TActorContext &ctx) {
  LOG_INFO_S(ctx, NKikimrServices::HIVE, "[" << TabletID() << "] TEvDeleteTablet, msg: " << ev->Get()->Record.ShortDebugString());
  NKikimrHive::TEvDeleteTablet& rec = ev->Get()->Record;
- Cout << "FAKEHIVE " << TabletID() << " TEvDeleteTablet " << rec.ShortDebugString() << Endl;
+ Cerr << "FAKEHIVE " << TabletID() << " TEvDeleteTablet " << rec.ShortDebugString() << Endl;
  TVector<ui64> deletedIdx;
  for (size_t i = 0; i < rec.ShardLocalIdxSize(); ++i) {
  auto id = std::make_pair<ui64, ui64>(rec.GetShardOwnerId(), rec.GetShardLocalIdx(i));
@@ -1368,7 +1372,7 @@ namespace NKikimr {
  void Handle(TEvHive::TEvDeleteOwnerTablets::TPtr &ev, const TActorContext &ctx) {
  LOG_INFO_S(ctx, NKikimrServices::HIVE, "[" << TabletID() << "] TEvDeleteOwnerTablets, msg: " << ev->Get()->Record);
  NKikimrHive::TEvDeleteOwnerTablets& rec = ev->Get()->Record;
- Cout << "FAKEHIVE " << TabletID() << " TEvDeleteOwnerTablets " << rec.ShortDebugString() << Endl;
+ Cerr << "FAKEHIVE " << TabletID() << " TEvDeleteOwnerTablets " << rec.ShortDebugString() << Endl;
  auto ownerId = rec.GetOwner();
  TVector<ui64> toDelete;
 
@@ -1400,6 +1404,34 @@ namespace NKikimr {
  ctx.Send(ev->Sender, new TEvHive::TEvDeleteOwnerTabletsReply(NKikimrProto::OK, TabletID(), ownerId, rec.GetTxId()));
  }
 
+ void StopTablet(const ui64& tabletId, const TActorContext &ctx) {
+ auto ownerIt = State->TabletIdToOwner.find(tabletId);
+ if (ownerIt == State->TabletIdToOwner.end()) {
+ return;
+ }
+ auto it = State->Tablets.find(ownerIt->second);
+ if (it == State->Tablets.end()) {
+ return;
+ }
+
+ TFakeHiveTabletInfo& tabletInfo = it->second;
+
+ // Very similar to DeleteTablet but don't actually removes tablet
+ // Kill the tablet and don't restart it
+ TActorId bootstrapperActorId = tabletInfo.BootstrapperActorId;
+ ctx.Send(bootstrapperActorId, new TEvBootstrapper::TEvStandBy());
+
+ tabletInfo.State = ETabletState::Stopped;
+ }
+
+ void Handle(TEvHive::TEvStopTablet::TPtr &ev, const TActorContext &ctx) {
+ LOG_INFO_S(ctx, NKikimrServices::HIVE, "[" << TabletID() << "] TEvStopTablet, msg: " << ev->Get()->Record.ShortDebugString());
+ NKikimrHive::TEvStopTablet& rec = ev->Get()->Record;
+ Cerr << "FAKEHIVE " << TabletID() << " TEvStopTablet " << rec.ShortDebugString() << Endl;
+ StopTablet(rec.GetTabletID(), ctx);
+ ctx.Send(ev->Sender, new TEvHive::TEvStopTabletResult(NKikimrProto::OK, rec.GetTabletID()));
+ }
+
  void Handle(TEvHive::TEvRequestHiveInfo::TPtr &ev, const TActorContext &ctx) {
  LOG_INFO_S(ctx, NKikimrServices::HIVE, "[" << TabletID() << "] TEvRequestHiveInfo, msg: " << ev->Get()->Record.ShortDebugString());
  const auto& record = ev->Get()->Record;
@@ -1451,15 +1483,15 @@ namespace NKikimr {
 
  void Handle(TEvHive::TEvUpdateDomain::TPtr &ev, const TActorContext &ctx) {
  LOG_INFO_S(ctx, NKikimrServices::HIVE, "[" << TabletID() << "] TEvUpdateDomain, msg: " << ev->Get()->Record.ShortDebugString());
- 
+
  const TSubDomainKey subdomainKey(ev->Get()->Record.GetDomainKey());
  NHive::TDomainInfo& domainInfo = State->Domains[subdomainKey];
  if (ev->Get()->Record.HasServerlessComputeResourcesMode()) {
  domainInfo.ServerlessComputeResourcesMode = ev->Get()->Record.GetServerlessComputeResourcesMode();
  } else {
  domainInfo.ServerlessComputeResourcesMode.Clear();
  }
- 
+
  auto response = std::make_unique<TEvHive::TEvUpdateDomainReply>();
  response->Record.SetTxId(ev->Get()->Record.GetTxId());
  response->Record.SetOrigin(TabletID());
@@ -1511,7 +1543,8 @@ namespace NKikimr {
  tabletInfo.SetTabletID(tabletId);
  if (info) {
  tabletInfo.SetTabletType(info->Type);
- tabletInfo.SetState(200); // THive::ReadyToWork
+ tabletInfo.SetState(ui32(info->State)); // THive::ETabletState::*
+ tabletInfo.MutableObjectDomain()->CopyFrom(info->ObjectDomain);
 
  // TODO: fill other fields when needed
  }

ydb/core/testlib/tablet_helpers.h

@@ -126,7 +126,7 @@ namespace NKikimr {
  : DomainKey(domainKey)
  {}
  };
- 
+
  struct TEvRequestDomainInfoReply: public TEventLocal<TEvRequestDomainInfoReply, EvRequestDomainInfoReply> {
  NHive::TDomainInfo DomainInfo;
 
@@ -137,10 +137,20 @@ namespace NKikimr {
 
  };
 
+
+ // partial mirror of NHive::ETabletState states from ydb/core/mind/hive/hive.h
+ enum class ETabletState : ui64 {
+ Unknown = 0, // THive::ETabletState::Unknown
+ Stopped = 100, // THive::ETabletState::Stopped
+ ReadyToWork = 200, // THive::ETabletState::ReadyToWork
+ };
+
  struct TFakeHiveTabletInfo {
  const TTabletTypes::EType Type;
  const ui64 TabletId;
  TActorId BootstrapperActorId;
+ ETabletState State = ETabletState::Unknown;
+ TSubDomainKey ObjectDomain; // what subdomain tablet belongs to
 
  TChannelsBindings BoundChannels;
  ui32 ChannelsProfile;

ydb/core/tx/coordinator/mediator_queue.cpp

@@ -20,6 +20,16 @@ static constexpr size_t ConfirmedStepsToFlush = 2;
 // the number of rows as large transactions are problematic to commit.
 static constexpr size_t ConfirmedParticipantsToFlush = 10'000;
 
+// Coordinator must reconnect with mediator as quickly as possible if connection is lost.
+// Retry policy prevents the reconnect loop from becoming too aggressive.
+static NTabletPipe::TClientRetryPolicy MediatorSyncRetryPolicy{
+ .RetryLimitCount = std::numeric_limits<ui32>::max(),
+ .MinRetryTime = TDuration::MilliSeconds(1),
+ .MaxRetryTime = TDuration::MilliSeconds(10),
+ .BackoffMultiplier = 2,
+ .DoFirstRetryInstantly = true,
+};
+
 void TMediatorStep::SerializeTo(TEvTxCoordinator::TEvCoordinatorStep *msg) const {
  for (const TTx &tx : Transactions) {
  NKikimrTx::TCoordinatorTransaction *x = msg->Record.AddTransactions();
@@ -50,6 +60,7 @@ class TTxCoordinatorMediatorQueue : public TActorBootstrapped<TTxCoordinatorMedi
  size_t ConfirmedParticipants = 0;
  size_t ConfirmedSteps = 0;
 
+
  void Die(const TActorContext &ctx) override {
  if (PipeClient) {
  NTabletPipe::CloseClient(ctx, PipeClient);
@@ -66,7 +77,7 @@ class TTxCoordinatorMediatorQueue : public TActorBootstrapped<TTxCoordinatorMedi
  PipeClient = TActorId();
  }
 
- PipeClient = ctx.RegisterWithSameMailbox(NTabletPipe::CreateClient(ctx.SelfID, Mediator));
+ PipeClient = ctx.RegisterWithSameMailbox(NTabletPipe::CreateClient(ctx.SelfID, Mediator, MediatorSyncRetryPolicy));
 
  LOG_DEBUG_S(ctx, NKikimrServices::TX_COORDINATOR_MEDIATOR_QUEUE, "Actor# " << ctx.SelfID.ToString()
  << " tablet# " << Coordinator << " SEND EvCoordinatorSync to# " << Mediator << " Mediator");

ydb/core/tx/schemeshard/schemeshard__clean_pathes.cpp

@@ -224,6 +224,13 @@ struct TSchemeShard::TTxCleanDroppedSubDomains : public TTransactionBase<TScheme
  << ", at schemeshard: "<< Self->TabletID());
  Self->PersistRemoveSubDomain(db, pathId);
  ++RemovedCount;
+
+ // This is for tests, so that tests could wait for actual lifetime end of a subdomain.
+ // It's kinda ok to reply from execute, and actually required for tests with reboots
+ // (to not lose event on a tablet reboot).
+ {
+ ctx.Send(Self->SelfId(), new TEvPrivate::TEvTestNotifySubdomainCleanup(pathId));
+ }
  } else {
  // Probably never happens, but better safe than sorry.
  ++SkippedCount;
@@ -237,12 +244,13 @@ struct TSchemeShard::TTxCleanDroppedSubDomains : public TTransactionBase<TScheme
  Y_ABORT_UNLESS(Self->CleanDroppedSubDomainsInFly);
 
  if (RemovedCount || SkippedCount) {
- LOG_NOTICE_S(ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
- "TTxCleanDroppedSubDomains Complete"
- << ", done PersistRemoveSubDomain for " << RemovedCount << " paths"
- << ", skipped " << SkippedCount
- << ", left " << Self->CleanDroppedSubDomainsCandidates.size() << " candidates"
- << ", at schemeshard: "<< Self->TabletID());
+ LOG_NOTICE_S(ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "TTxCleanDroppedSubDomains Complete"
+ << ", done PersistRemoveSubDomain for " << RemovedCount << " subdomains"
+ << ", skipped " << SkippedCount
+ << ", left " << Self->CleanDroppedSubDomainsCandidates.size() << " candidates"
+ << ", at schemeshard: "<< Self->TabletID()
+ );
+
  }
 
  if (!Self->CleanDroppedSubDomainsCandidates.empty()) {

ydb/core/tx/schemeshard/schemeshard__init.cpp

@@ -674,6 +674,25 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
  return true;
  }
 
+ bool LoadSystemShardsToDelete(NIceDb::TNiceDb& db, TShardsToDeleteRows& shardsToDelete) const {
+ {
+ auto rowSet = db.Table<Schema::SystemShardsToDelete>().Range().Select();
+ if (!rowSet.IsReady()) {
+ return false;
+ }
+ while (!rowSet.EndOfSet()) {
+ const auto shardIdx = Self->MakeLocalId(rowSet.GetValue<Schema::SystemShardsToDelete::ShardIdx>());
+ shardsToDelete.emplace_back(shardIdx);
+
+ if (!rowSet.Next()) {
+ return false;
+ }
+ }
+ }
+
+ return true;
+ }
+
  typedef std::tuple<TOperationId, TShardIdx, TTxState::ETxState> TTxShardRec;
  typedef TVector<TTxShardRec> TTxShardsRows;
 
@@ -3790,6 +3809,23 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
  }
  }
 
+ // Read system shards to delete
+ {
+ TShardsToDeleteRows shardsToDelete;
+ if (!LoadSystemShardsToDelete(db, shardsToDelete)) {
+ return false;
+ }
+
+ LOG_NOTICE_S(ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
+ "TTxInit for SystemShardToDelete"
+ << ", read records: " << shardsToDelete.size()
+ << ", at schemeshard: " << Self->TabletID());
+
+ for (auto& rec: shardsToDelete) {
+ OnComplete.DeleteSystemShard(std::get<0>(rec));
+ }
+ }
+
  // Read backup settings
  {
  TBackupSettingsRows backupSettings;
@@ -3845,7 +3881,7 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
  auto desc = tableInfo->BackupSettings.MutableTable();
  Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(*desc, tableDesc));
  }
- 
+
  if (changefeedUnderlyingTopics) {
  NKikimrSchemeOp::TChangefeedUnderlyingTopics wrapperOverTopics;
  Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(wrapperOverTopics, changefeedUnderlyingTopics));
@@ -4525,12 +4561,12 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
  try {
  fill(buildInfo);
  } catch (const std::exception& exc) {
- LOG_ERROR_S(ctx, NKikimrServices::BUILD_INDEX, 
+ LOG_ERROR_S(ctx, NKikimrServices::BUILD_INDEX,
  "Init " << stepName << " unhandled exception, id#" << buildInfo.Id
  << " " << TypeName(exc) << ": " << exc.what() << Endl
  << TBackTrace::FromCurrentException().PrintToString()
  << ", TIndexBuildInfo: " << buildInfo);
- 
+
  // in-memory volatile state:
  buildInfo.IsBroken = true;
  buildInfo.AddIssue(TStringBuilder() << "Init " << stepName << " unhandled exception " << exc.what());
@@ -4541,7 +4577,7 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
  const auto* buildInfoPtr = Self->IndexBuilds.FindPtr(id);
  Y_ASSERT(buildInfoPtr);
  if (!buildInfoPtr) {
- LOG_ERROR_S(ctx, NKikimrServices::BUILD_INDEX, 
+ LOG_ERROR_S(ctx, NKikimrServices::BUILD_INDEX,
  "Init " << stepName << " BuildInfo not found: id#" << id);
  return;
  }

ydb/core/tx/schemeshard/schemeshard__operation_drop_extsubdomain.cpp

@@ -10,13 +10,46 @@ namespace {
 using namespace NKikimr;
 using namespace NSchemeShard;
 
-class TDeletePrivateShards: public TDeleteParts {
+class TDeleteSubdomainSystemShards: public TSubOperationState {
+protected:
+ const TOperationId OperationId;
+
+ TString DebugHint() const override {
+ return TStringBuilder() << "TDeleteSubdomainSystemShards" << " opId# " << OperationId << " ";
+ }
+
 public:
- explicit TDeletePrivateShards(const TOperationId& id)
- : TDeleteParts(id, TTxState::Done)
+ explicit TDeleteSubdomainSystemShards(const TOperationId& id)
+ : OperationId(id)
  {
  IgnoreMessages(DebugHint(), AllIncomingEvents());
  }
+
+ bool ProgressState(TOperationContext& context) override {
+ LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "[" << context.SS->SelfTabletId() << "] " << DebugHint() << "ProgressState");
+
+ const auto* txState = context.SS->FindTx(OperationId);
+ Y_ABORT_UNLESS(txState);
+ Y_ABORT_UNLESS(txState->TxType == TTxState::TxForceDropExtSubDomain);
+
+ auto subdomain = context.SS->SubDomains.at(txState->TargetPathId);
+ Y_ABORT_UNLESS(subdomain);
+
+ // Initiate asynchronous deletion of system shards
+ if (subdomain->GetSharedHive()) {
+ for (const auto& shard : txState->Shards) {
+ context.OnComplete.DeleteShard(shard.Idx);
+ }
+ } else {
+ for (const auto& shard : txState->Shards) {
+ context.OnComplete.DeleteSystemShard(shard.Idx);
+ }
+ }
+
+ NIceDb::TNiceDb db(context.GetDB());
+ context.SS->ChangeTxState(db, OperationId, TTxState::Done);
+ return true;
+ }
 };
 
 class TDeleteExternalShards: public TSubOperationState {
@@ -237,7 +270,7 @@ class TDropExtSubdomain: public TSubOperation {
  case TTxState::DeleteExternalShards:
  return MakeHolder<TDeleteExternalShards>(OperationId);
  case TTxState::DeletePrivateShards:
- return MakeHolder<TDeletePrivateShards>(OperationId);
+ return MakeHolder<TDeleteSubdomainSystemShards>(OperationId);
  case TTxState::Done:
  return MakeHolder<TDone>(OperationId);
  default: