cluster: avoid duplicate partition registration callbacks

If registration occurs while a partition is co_await starting, it will
be present in the ntp_table and the upcalls will happen both inline with
registration and again after the partition is started.

This patch skips callback at time of registration if the partition is in
the process of starting, and relies on the callback that is invoked
after the partition is started.

Signed-off-by: Noah Watkins <[email protected]>

src/v/cluster/partition.h

@@ -348,6 +348,9 @@ class partition : public ss::enable_lw_shared_from_this<partition> {
       partition_properties_stm::writes_disabled disable,
       model::timeout_clock::time_point deadline);
 
+    bool started() const noexcept { return _started; }
+    void mark_started() noexcept { _started = true; }
+
 private:
     ss::future<result<ssx::rwlock_unit>> hold_writes_enabled();
 
@@ -410,6 +413,8 @@ class partition : public ss::enable_lw_shared_from_this<partition> {
     // exclusive ("write") for enabling/disabling writes
     ssx::rwlock _produce_lock;
 
+    bool _started{false};
+
     friend std::ostream& operator<<(std::ostream& o, const partition& x);
 };
 } // namespace cluster

src/v/cluster/partition_manager.cc

@@ -297,6 +297,12 @@ ss::future<consensus_ptr> partition_manager::manage(
 
     co_await p->start(_stm_registry, xst_state);
 
+    // this is not done in partition::start itself because the purpose of this
+    // flag is to operate in an uninterruptible context with watcher
+    // notification below to avoid registration while this fiber is blocked,
+    // leading to double notifications.
+    p->mark_started();
+
     _manage_watchers.notify(p->ntp(), p);
 
     co_return c;

src/v/cluster/partition_manager.h

@@ -122,7 +122,9 @@ class partition_manager
             cb(std::move(p));
         });
         for (auto& e : _ntp_table) {
-            init.notify(e.first, e.second);
+            if (e.second->started()) {
+                init.notify(e.first, e.second);
+            }
         }
 
         // now setup the permenant callback for new partitions
@@ -134,7 +136,9 @@ class partition_manager
         init.register_notify(
           ns, [&cb](ss::lw_shared_ptr<partition> p) { cb(std::move(p)); });
         for (auto& e : _ntp_table) {
-            init.notify(e.first, e.second);
+            if (e.second->started()) {
+                init.notify(e.first, e.second);
+            }
         }
         return _manage_watchers.register_notify(ns, std::move(cb));
     }