Merge 8300489 into 83145b0

Author: CyberROFL

ydb/core/tx/replication/controller/controller.cpp

@@ -344,7 +344,9 @@ void TController::Handle(TEvService::TEvWorkerStatus::TPtr& ev, const TActorCont
         }
         break;
     case NKikimrReplication::TEvWorkerStatus::STOPPED:
-        MaybeRemoveWorker(id, ctx);
+        if (!MaybeRemoveWorker(id, ctx)) {
+            BootQueue.insert(id);
+        }
         break;
     default:
         CLOG_W(ctx, "Unknown worker status"
@@ -589,10 +591,13 @@ void TController::RemoveWorker(const TWorkerId& id, const TActorContext& ctx) {
     target->Progress(ctx);
 }
 
-void TController::MaybeRemoveWorker(const TWorkerId& id, const TActorContext& ctx) {
-    if (RemoveQueue.contains(id)) {
-        RemoveWorker(id, ctx);
+bool TController::MaybeRemoveWorker(const TWorkerId& id, const TActorContext& ctx) {
+    if (!RemoveQueue.contains(id)) {
+        return false;
     }
+
+    RemoveWorker(id, ctx);
+    return true;
 }
 
 void TController::Handle(TEvInterconnect::TEvNodeDisconnected::TPtr& ev, const TActorContext& ctx) {

ydb/core/tx/replication/controller/controller_impl.h

@@ -100,7 +100,7 @@ class TController
     void BootWorker(ui32 nodeId, const TWorkerId& id, const NKikimrReplication::TRunWorkerCommand& cmd);
     void StopWorker(ui32 nodeId, const TWorkerId& id);
     void RemoveWorker(const TWorkerId& id, const TActorContext& ctx);
-    void MaybeRemoveWorker(const TWorkerId& id, const TActorContext& ctx);
+    bool MaybeRemoveWorker(const TWorkerId& id, const TActorContext& ctx);
 
     // local transactions
     class TTxInitSchema;

ydb/core/tx/replication/service/service.cpp

@@ -45,21 +45,47 @@ class TSessionInfo {
         return Workers.contains(id);
     }
 
-    void RegisterWorker(IActorOps* ops, const TWorkerId& id, IActor* actor) {
+    bool HasWorker(const TActorId& id) const {
+        return ActorIdToWorkerId.contains(id);
+    }
+
+    TActorId GetWorkerActorId(const TWorkerId& id) const {
+        auto it = Workers.find(id);
+        Y_ABORT_UNLESS(it != Workers.end());
+        return it->second;
+    }
+
+    TActorId RegisterWorker(IActorOps* ops, const TWorkerId& id, IActor* actor) {
         auto res = Workers.emplace(id, ops->Register(actor));
         Y_ABORT_UNLESS(res.second);
 
+        const auto actorId = res.first->second;
+        ActorIdToWorkerId.emplace(actorId, id);
+
         ops->Send(ActorId, new TEvService::TEvWorkerStatus(id, NKikimrReplication::TEvWorkerStatus::RUNNING));
+        return actorId;
     }
 
     void StopWorker(IActorOps* ops, const TWorkerId& id) {
         auto it = Workers.find(id);
         Y_ABORT_UNLESS(it != Workers.end());
 
         ops->Send(it->second, new TEvents::TEvPoison());
+        ops->Send(ActorId, new TEvService::TEvWorkerStatus(id, NKikimrReplication::TEvWorkerStatus::STOPPED));
+
+        ActorIdToWorkerId.erase(it->second);
         Workers.erase(it);
+    }
 
-        ops->Send(ActorId, new TEvService::TEvWorkerStatus(id, NKikimrReplication::TEvWorkerStatus::STOPPED));
+    void StopWorker(IActorOps* ops, const TActorId& id) {
+        auto it = ActorIdToWorkerId.find(id);
+        Y_ABORT_UNLESS(it != ActorIdToWorkerId.end());
+
+        // actor already stopped
+        ops->Send(ActorId, new TEvService::TEvWorkerStatus(it->second, NKikimrReplication::TEvWorkerStatus::STOPPED));
+
+        Workers.erase(it->second);
+        ActorIdToWorkerId.erase(it);
     }
 
     void SendStatus(IActorOps* ops) const {
@@ -83,6 +109,7 @@ class TSessionInfo {
     TActorId ActorId;
     ui64 Generation;
     THashMap<TWorkerId, TActorId> Workers;
+    THashMap<TActorId, TWorkerId> ActorIdToWorkerId;
 
 }; // TSessionInfo
 
@@ -252,7 +279,9 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
         // TODO: validate settings
         const auto& readerSettings = cmd.GetRemoteTopicReader();
         const auto& writerSettings = cmd.GetLocalTableWriter();
-        session.RegisterWorker(this, id, CreateWorker(ReaderFn(readerSettings), WriterFn(writerSettings)));
+        const auto actorId = session.RegisterWorker(this, id,
+            CreateWorker(SelfId(), ReaderFn(readerSettings), WriterFn(writerSettings)));
+        WorkerActorIdToSession[actorId] = controller.GetTabletId();
     }
 
     void Handle(TEvService::TEvStopWorker::TPtr& ev) {
@@ -284,10 +313,43 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
         if (session.HasWorker(id)) {
             LOG_I("Stop worker"
                 << ": worker# " << id);
+            WorkerActorIdToSession.erase(session.GetWorkerActorId(id));
             session.StopWorker(this, id);
         }
     }
 
+    void Handle(TEvWorker::TEvGone::TPtr& ev) {
+        LOG_T("Handle " << ev->Get()->ToString());
+
+        auto wit = WorkerActorIdToSession.find(ev->Sender);
+        if (wit == WorkerActorIdToSession.end()) {
+            LOG_W("Unknown worker has gone"
+                << ": worker# " << ev->Sender);
+            return;
+        }
+
+        auto it = Sessions.find(wit->second);
+        if (it == Sessions.end()) {
+            LOG_E("Cannot find session"
+                << ": worker# " << ev->Sender
+                << ", session# " << wit->second);
+            return;
+        }
+
+        auto& session = it->second;
+        if (!session.HasWorker(ev->Sender)) {
+            LOG_E("Cannot find worker"
+                << ": worker# " << ev->Sender
+                << ", session# " << wit->second);
+            return;
+        }
+
+        LOG_I("Worker has gone"
+            << ": worker# " << ev->Sender);
+        WorkerActorIdToSession.erase(ev->Sender);
+        session.StopWorker(this, ev->Sender);
+    }
+
     void PassAway() override {
         if (auto actorId = std::exchange(BoardPublisher, {})) {
             Send(actorId, new TEvents::TEvPoison());
@@ -319,6 +381,7 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
             hFunc(TEvService::TEvHandshake, Handle);
             hFunc(TEvService::TEvRunWorker, Handle);
             hFunc(TEvService::TEvStopWorker, Handle);
+            hFunc(TEvWorker::TEvGone, Handle);
             sFunc(TEvents::TEvPoison, PassAway);
         }
     }
@@ -328,6 +391,7 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
     TActorId BoardPublisher;
     THashMap<ui64, TSessionInfo> Sessions;
     THashMap<TCredentialsKey, TActorId> YdbProxies;
+    THashMap<TActorId, ui64> WorkerActorIdToSession;
 
 }; // TReplicationService
 

ydb/core/tx/replication/service/table_writer.cpp

@@ -139,6 +139,9 @@ class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter> {
     }
 
     void Leave(bool hardError = false) {
+        LOG_I("Leave"
+            << ": hard error# " << hardError);
+
         Send(Parent, new NChangeExchange::TEvChangeExchangePrivate::TEvGone(TabletId, hardError));
         PassAway();
     }
@@ -495,6 +498,9 @@ class TLocalTableWriter
     }
 
     void Leave(TEvWorker::TEvGone::EStatus status) {
+        LOG_I("Leave"
+            << ": status# " << status);
+
         Send(Worker, new TEvWorker::TEvGone(status));
         PassAway();
     }

ydb/core/tx/replication/service/topic_reader.cpp

@@ -76,7 +76,9 @@ class TRemoteTopicReader: public TActor<TRemoteTopicReader> {
     }
 
     void Leave(TEvWorker::TEvGone::EStatus status) {
-        LOG_I("Leave");
+        LOG_I("Leave"
+            << ": status# " << status);
+
         Send(Worker, new TEvWorker::TEvGone(status));
         PassAway();
     }

ydb/core/tx/replication/service/worker.cpp

@@ -191,12 +191,15 @@ class TWorker: public TActorBootstrapped<TWorker> {
             }
             [[fallthrough]];
         default:
-            return Leave();
+            return Leave(status);
         }
     }
 
-    void Leave() {
-        // TODO: signal to parent
+    void Leave(TEvWorker::TEvGone::EStatus status) {
+        LOG_I("Leave"
+            << ": status# " << status);
+
+        Send(Parent, new TEvWorker::TEvGone(status));
         PassAway();
     }
 
@@ -213,8 +216,12 @@ class TWorker: public TActorBootstrapped<TWorker> {
         return NKikimrServices::TActivity::REPLICATION_WORKER;
     }
 
-    explicit TWorker(std::function<IActor*(void)>&& createReaderFn, std::function<IActor*(void)>&& createWriterFn)
-        : Reader(std::move(createReaderFn))
+    explicit TWorker(
+            const TActorId& parent, 
+            std::function<IActor*(void)>&& createReaderFn,
+            std::function<IActor*(void)>&& createWriterFn)
+        : Parent(parent)
+        , Reader(std::move(createReaderFn))
         , Writer(std::move(createWriterFn))
     {
     }
@@ -239,14 +246,19 @@ class TWorker: public TActorBootstrapped<TWorker> {
 
 private:
     static constexpr ui32 MaxAttempts = 3;
+    const TActorId Parent;
     mutable TMaybe<TString> LogPrefix;
     TActorInfo Reader;
     TActorInfo Writer;
     THolder<TEvWorker::TEvData> InFlightData;
 };
 
-IActor* CreateWorker(std::function<IActor*(void)>&& createReaderFn, std::function<IActor*(void)>&& createWriterFn) {
-    return new TWorker(std::move(createReaderFn), std::move(createWriterFn));
+IActor* CreateWorker(
+        const TActorId& parent,
+        std::function<IActor*(void)>&& createReaderFn,
+        std::function<IActor*(void)>&& createWriterFn)
+{
+    return new TWorker(parent, std::move(createReaderFn), std::move(createWriterFn));
 }
 
 }

ydb/core/tx/replication/service/worker.h

@@ -59,7 +59,10 @@ struct TEvWorker {
     };
 };
 
-IActor* CreateWorker(std::function<IActor*(void)>&& createReaderFn, std::function<IActor*(void)>&& createWriterFn);
+IActor* CreateWorker(
+    const TActorId& parent,
+    std::function<IActor*(void)>&& createReaderFn,
+    std::function<IActor*(void)>&& createWriterFn);
 
 }
 

ydb/core/tx/replication/service/worker_ut.cpp

@@ -56,7 +56,7 @@ Y_UNIT_TEST_SUITE(Worker) {
             return CreateLocalTableWriter(tablePathId);
         };
 
-        auto worker = env.GetRuntime().Register(CreateWorker(std::move(createReaderFn), std::move(createWriterFn)));
+        auto worker = env.GetRuntime().Register(CreateWorker(env.GetSender(), std::move(createReaderFn), std::move(createWriterFn)));
         Y_UNUSED(worker);
 
         UNIT_ASSERT(WriteTopic(env, "/Root/topic", R"({"key":[1], "update":{"value":"10"}})"));