From e30f743b5cc2928420ce2e8224e04dcbc89e85c1 Mon Sep 17 00:00:00 2001
From: Dominik Charousset <dominik@charousset.de>
Date: Mon, 13 Feb 2023 19:59:20 +0100
Subject: [PATCH] Fix UB when using prefix_and_tail

(cherry picked from commit 68aed3b2943137a8bab46cac41c2400ac93737fa)
(cherry picked from commit 4c3f803b0b3d7ecc802de2b9642b0ec7a40cfe10)
---
 libcaf_core/caf/flow/op/from_steps.hpp      | 3 +++
 libcaf_core/caf/flow/op/prefix_and_tail.hpp | 2 +-
 libcaf_core/caf/flow/op/ucast.hpp           | 5 +++++
 3 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/libcaf_core/caf/flow/op/from_steps.hpp b/libcaf_core/caf/flow/op/from_steps.hpp
index efcf300bca..8c0d330d7b 100644
--- a/libcaf_core/caf/flow/op/from_steps.hpp
+++ b/libcaf_core/caf/flow/op/from_steps.hpp
@@ -214,6 +214,9 @@ class from_steps_sub : public detail::plain_ref_counted,
         buf_.pop_front();
         --demand_;
         out_.on_next(item);
+        // Note: on_next() may call dispose() and set out_ to nullptr.
+        if (!out_)
+          return;
       }
       if (in_) {
         pull();
diff --git a/libcaf_core/caf/flow/op/prefix_and_tail.hpp b/libcaf_core/caf/flow/op/prefix_and_tail.hpp
index 0a6e19b30d..22334b1598 100644
--- a/libcaf_core/caf/flow/op/prefix_and_tail.hpp
+++ b/libcaf_core/caf/flow/op/prefix_and_tail.hpp
@@ -121,7 +121,7 @@ class prefix_and_tail_sub : public detail::plain_ref_counted,
   }
 
   void request(size_t demand) override {
-    // Only called by the out_, never by the sink_. The latter triggers
+    // Only called by out_, never by sink_ (triggers on_sink_demand_change()).
     prefix_demand_ += demand;
     if (sub_ && !requested_prefix_) {
       sub_.request(prefix_size_);
diff --git a/libcaf_core/caf/flow/op/ucast.hpp b/libcaf_core/caf/flow/op/ucast.hpp
index 3d306db091..9db4350593 100644
--- a/libcaf_core/caf/flow/op/ucast.hpp
+++ b/libcaf_core/caf/flow/op/ucast.hpp
@@ -123,6 +123,9 @@ class ucast_sub_state : public detail::plain_ref_counted {
       auto got_some = demand > 0 && !buf.empty();
       for (bool run = got_some; run; run = demand > 0 && !buf.empty()) {
         out.on_next(buf.front());
+        // Note: on_next may call dispose().
+        if (disposed)
+          return;
         buf.pop_front();
         --demand;
       }
@@ -166,6 +169,8 @@ class ucast_sub : public subscription::impl_base {
   }
 
   void request(size_t n) override {
+    if (!state_)
+      return;
     state_->demand += n;
     if (state_->when_demand_changed)
       state_->when_demand_changed.run();