From fb57ccd229f4f8c357b28e68e8712379b96258fb Mon Sep 17 00:00:00 2001
From: Bharath Vissapragada <bharathv@redpanda.com>
Date: Fri, 24 Jan 2025 17:33:55 -0800
Subject: [PATCH 1/2] rm_stm/apply: hold gate during apply

Currently apply fiber can continue to run (and possibly add new
producers to _producers map) as the state machine is shutting down.
This can manifest in weird crashes as the clean up destroys the
_producers without deregistering properly.

First manifestation

Iterator invalidation in reset_producers() as it loops thru _producers
with scheduling points while state machine apply adds new producers

future<> rm_stm::stop() {
.....
    co_await _gate.close();
    co_await reset_producers();  <---- interferes with state machine apply
    _metrics.clear();
    co_await raft::persisted_stm<>::stop();
.....

Second manifestation

Crashes: every producer creation registers with an intrusive list in
producer_state_manager using a safe link. Now, if a new producer is
registered after reset_producers, the map is destroyed in the state
machine destructor without unlinking from the producer_state_manager
and the safe_link fires an assert.

This bug has been there forever from what I can tell, perhaps got
worsened with recent changes that added more scheduling points in the
surrounding code.
---
 src/v/cluster/rm_stm.cc | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/v/cluster/rm_stm.cc b/src/v/cluster/rm_stm.cc
index d8696c5384504..5cfb568045711 100644
--- a/src/v/cluster/rm_stm.cc
+++ b/src/v/cluster/rm_stm.cc
@@ -1676,6 +1676,7 @@ void rm_stm::apply_fence(model::producer_identity pid, model::record_batch b) {
 }
 
 ss::future<> rm_stm::do_apply(const model::record_batch& b) {
+    auto holder = _gate.hold();
     const auto& hdr = b.header();
     const auto bid = model::batch_identity::from(hdr);
 

From 873b28214f238b4d902234ba42c8a4c4c564ac29 Mon Sep 17 00:00:00 2001
From: Bharath Vissapragada <bharathv@redpanda.com>
Date: Fri, 24 Jan 2025 17:35:20 -0800
Subject: [PATCH 2/2] rm_stm/logging: more logging

---
 src/v/cluster/rm_stm.cc | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/v/cluster/rm_stm.cc b/src/v/cluster/rm_stm.cc
index 5cfb568045711..c286a2af0882e 100644
--- a/src/v/cluster/rm_stm.cc
+++ b/src/v/cluster/rm_stm.cc
@@ -180,6 +180,7 @@ rm_stm::maybe_create_producer(model::producer_identity pid) {
             return std::make_pair(producer, producer_previously_known::yes);
         }
     }
+    vlog(_ctx_log.trace, "creating producer for pid: {}", pid);
     auto producer = ss::make_lw_shared<producer_state>(
       _ctx_log, pid, _raft->group(), [this](model::producer_identity pid) {
           cleanup_producer_state(pid);
@@ -238,6 +239,7 @@ void rm_stm::cleanup_producer_state(model::producer_identity pid) noexcept {
 };
 
 ss::future<> rm_stm::reset_producers() {
+    vlog(_ctx_log.trace, "reseting producers");
     // note: must always be called under exlusive write lock to
     // avoid concurrrent state changes to _producers.
     co_await ss::max_concurrent_for_each(