transform: abort siblings when producer loop fails

`run_all_producers()` fanned out one producer loop per output via
`ss::parallel_for_each()`, which captures the first exception but waits
for every loop to finish before resolving. A producer loop only exits
once its abort source fires, so when a single producer threw (e.g. a
transient "not a leader for partition" during a leadership transfer) the
surviving loops spun forever. parallel_for_each stayed pending, the
exception was never observed, and state::errored never fired -- so the
processor's reported state stayed stuck at running with no progress on
the affected output until an external pause/resume.

Single-output transforms were unaffected because `parallel_for_each()`
over one element resolves exceptionally immediately.

Drive the producer loops off a composite abort source that fires when
either the processor stops (_as) or a producer fails (a local source).
The first producer to fail records its exception and aborts the local
source, which unwinds the sibling producer loops promptly; the captured
exception is then rethrown so it is reported exactly once as
state::errored, and the manager restarts the processor.

Also updates the repro test from the previous commit to assert the
failure is reported for any output count, and adds a regression test that
fails a producer, restarts, and asserts a clean restart with no spurious
second error.

Author: WillemKauf

src/v/transform/BUILD

@@ -95,6 +95,7 @@ redpanda_cc_library(
         "//src/v/model:batch_compression",
         "//src/v/random:time_jitter",
         "//src/v/rpc",
+        "//src/v/ssx:abort_source",
         "//src/v/ssx:future_util",
         "//src/v/ssx:sformat",
         "//src/v/utils:backoff_policy",

src/v/transform/tests/test_fixture.cc

@@ -43,6 +43,11 @@ void fake_sink::fail_writes() {
     _cond_var.broadcast();
 }
 
+void fake_sink::resume_writes() {
+    _fail = false;
+    _cond_var.broadcast();
+}
+
 class read_timed_out : public ss::condition_variable_timed_out {
     const char* what() const noexcept override {
         return "waiting for read timed out";

src/v/transform/tests/test_fixture.h

@@ -115,6 +115,12 @@ class fake_sink : public sink {
      */
     void fail_writes();
 
+    /**
+     * Stop failing writes, mimicking the transient produce failure (e.g. a
+     * leadership transfer) resolving so the sink can make progress again.
+     */
+    void resume_writes();
+
 private:
     ss::chunked_fifo<model::record> _records;
     ss::condition_variable _cond_var;

src/v/transform/tests/transform_processor_test.cc

@@ -224,6 +224,9 @@ class ProcessorTestFixture : public ::testing::TestWithParam<fixture_param> {
     void fail_sink(model::output_topic_index idx) {
         _sinks[idx()]->fail_writes();
     }
+    void recover_sink(model::output_topic_index idx) {
+        _sinks[idx()]->resume_writes();
+    }
     bool processor_running() const { return _p->is_running(); }
 
     std::vector<model::output_topic_index> output_topics() const {
@@ -367,16 +370,24 @@ TEST_P(ProcessorTestFixture, LagOverflowBug) {
     EXPECT_EQ(lag(), 0);
 }
 
-// Reproduces the "stuck transform" bug. When one output's producer fails to
-// produce (e.g. a transient "not a leader for partition"), a SINGLE-output
-// transform reports the error (state::errored), so the manager would restart
-// it. But with MULTIPLE outputs, the surviving producer loop(s) keep
-// `ss::parallel_for_each` (run_all_producers) pending forever, so the exception
-// is never observed: state::errored never fires, the processor still reports
-// "running", and it makes no progress until an external stop (== an `rpk
-// transform pause`/`resume`).
-TEST_P(ProcessorTestFixture, ProduceFailureWedgesMultiOutputButErrorsSingle) {
-    const size_t n_out = GetParam().meta.output_topics.size();
+// Regression test for the "stuck transform" bug. When one output's producer
+// fails to produce (e.g. a transient "not a leader for partition"), the failure
+// must be reported as state::errored so the manager restarts the processor,
+// regardless of how many outputs the transform has.
+//
+// The bug: with MULTIPLE outputs, `run_all_producers()` fanned out via
+// `ss::parallel_for_each()`, which captures the first exception but waits for
+// every loop to finish before resolving. Nothing aborted the shared abort
+// source, so the surviving producer loop(s) spin forever, pinning
+// parallel_for_each pending: state::errored never fired, the processor stayed
+// "running", and it made no progress until an external stop (== an `rpk
+// transform pause`/`resume`). Single-output transforms were unaffected because
+// parallel_for_each over one element resolves exceptionally right away.
+//
+// The fix drives the producer loops off a composite abort source, so the first
+// failing producer unwinds its siblings (without touching the processor's own
+// abort source) and propagates exactly one error.
+TEST_P(ProcessorTestFixture, ProduceFailureIsReportedForAnyOutputCount) {
     set_tee_output();
 
     // Healthy baseline so the pipeline is flowing and committed.
@@ -392,26 +403,59 @@ TEST_P(ProcessorTestFixture, ProduceFailureWedgesMultiOutputButErrorsSingle) {
     push_batch(make_records(1));
     tests::drain_task_queue().get();
 
-    // The loops are never aborted, so the processor still reports running.
+    // The producers all unwind instead of wedging. The processor still reports
+    // running (its abort source is untouched); like a single-output failure, it
+    // relies on the manager observing the error and restarting it.
     EXPECT_TRUE(processor_running());
     // Output 0's producer died, so nothing is ever written there.
     EXPECT_TRUE(sink_empty(model::output_topic_index(0)));
+    // The failure is reported exactly once, no matter the output count: the
+    // consumer/transform loops keep running on the untouched abort source, so
+    // the manager restarts the processor without spurious duplicate errors.
+    EXPECT_EQ(error_count(), 1u)
+      << "produce failure should be reported as errored exactly once";
+}
 
-    if (n_out == 1) {
-        // parallel_for_each(1 element) resolves exceptionally, the error
-        // propagates, and the manager would restart the processor.
-        EXPECT_GE(error_count(), 1u)
-          << "single-output produce failure should be reported as errored";
-    } else {
-        // The surviving producer loop(s) keep parallel_for_each pending, so the
-        // error is swallowed -- THE BUG.
-        EXPECT_EQ(error_count(), 0u)
-          << "multi-output produce failure is silently swallowed (stuck "
-             "transform bug)";
-        // Healthy outputs keep flowing while output 0 is wedged.
-        for (size_t i = 1; i < n_out; ++i) {
-            EXPECT_FALSE(sink_empty(model::output_topic_index(i)));
-        }
+// A produce failure must leave the processor cleanly restartable: the manager's
+// recovery path stops and then restarts the same processor instance, so stop()
+// has to tear down (and the wedge fix must not have aborted the processor's own
+// abort source, which would short-circuit stop() and leave the engine started
+// for the restart to double-start). Exercise that full cycle and assert the
+// restart is clean and the processor resumes producing.
+TEST_P(ProcessorTestFixture, RecoversFromProduceFailureViaRestart) {
+    set_tee_output();
+
+    // Healthy baseline so the pipeline is flowing and committed.
+    auto baseline = make_records(1);
+    push_batch(baseline);
+    for (auto o : output_topics()) {
+        EXPECT_THAT(read_records(o, 1), SameRecords(baseline));
+    }
+    ASSERT_TRUE(wait_for_all_committed());
+
+    // A producer fails and the error is reported, exactly what the manager
+    // observes before it restarts.
+    fail_sink(model::output_topic_index(0));
+    push_batch(make_records(1));
+    tests::drain_task_queue().get();
+    ASSERT_EQ(error_count(), 1u);
+
+    // Recovery: the transient failure clears and the manager stops and restarts
+    // the same processor instance.
+    recover_sink(model::output_topic_index(0));
+    restart();
+    tests::drain_task_queue().get();
+
+    // A clean restart produces no further errors.
+    ASSERT_EQ(error_count(), 1u)
+      << "restart after a produce failure should not report a new error";
+    EXPECT_TRUE(processor_running());
+
+    // The restarted processor resumes producing to every output.
+    auto resumed = make_records(1);
+    push_batch(resumed);
+    for (auto o : output_topics()) {
+        EXPECT_FALSE(read_records(o, 1).empty());
     }
 }
 

src/v/transform/transform_processor.cc

@@ -18,6 +18,7 @@
 #include "model/timestamp.h"
 #include "model/transform.h"
 #include "random/simple_time_jitter.h"
+#include "ssx/abort_source.h"
 #include "ssx/future-util.h"
 #include "wasm/engine.h"
 
@@ -342,19 +343,50 @@ ss::future<> processor::run_transform_loop() {
 ss::future<> processor::run_all_producers(
   absl::flat_hash_map<model::output_topic_index, kafka::offset>
     latest_committed) {
-    return ss::parallel_for_each(
-      _outputs, [this, committed = std::move(latest_committed)](auto& entry) {
+    // parallel_for_each captures the first exception but waits for every loop
+    // to finish before resolving, and a producer loop only exits when its abort
+    // source fires. Drive the loops off a composite source that aborts when
+    // either the processor stops (_as) or a producer fails (local_producer_as,
+    // our "a producer failed" signal), so the first failure unwinds its
+    // siblings promptly instead of leaving them spinning forever.
+    ss::abort_source local_producer_as;
+    ssx::composite_abort_source producers_as(_as, local_producer_as);
+
+    std::exception_ptr failure;
+    co_await ss::parallel_for_each(
+      _outputs,
+      [this,
+       &failure,
+       &local_producer_as,
+       &producers_as,
+       committed = std::move(latest_committed)](auto& entry) {
           output& out = entry.second;
           return run_producer_loop(
-            out.index, &out.queue, out.sink.get(), committed.at(out.index));
+                   out.index,
+                   &out.queue,
+                   out.sink.get(),
+                   committed.at(out.index),
+                   producers_as.as())
+            .handle_exception([&failure,
+                               &local_producer_as](std::exception_ptr ep) {
+                if (!local_producer_as.abort_requested()) {
+                    failure = std::move(ep);
+                    local_producer_as.request_abort_ex(
+                      std::make_exception_ptr(processor_shutdown_exception()));
+                }
+            });
       });
+    if (failure) {
+        std::rethrow_exception(failure);
+    }
 }
 
 ss::future<> processor::run_producer_loop(
   model::output_topic_index index,
   transfer_queue<transformed_output>* queue,
   sink* sink,
-  kafka::offset last_committed) {
+  kafka::offset last_committed,
+  ss::abort_source& as) {
     vlog(
       _logger.debug,
       "starting producer {} - last committed: {}",
@@ -364,8 +396,8 @@ ss::future<> processor::run_producer_loop(
     // to suppress records until we've reached the previous offset we've
     // committed.
     bool suppress = true;
-    while (!_as.abort_requested()) {
-        auto popped = co_await queue->pop_all(&_as);
+    while (!as.abort_requested()) {
+        auto popped = co_await queue->pop_all(&as);
         if (popped.empty()) {
             continue;
         }

src/v/transform/transform_processor.h

@@ -114,7 +114,8 @@ class processor {
       model::output_topic_index,
       transfer_queue<transformed_output>*,
       sink*,
-      kafka::offset);
+      kafka::offset,
+      ss::abort_source&);
     ss::future<> poll_sleep();
     ss::future<absl::flat_hash_map<model::output_topic_index, kafka::offset>>
     load_latest_committed();