diff --git a/cpp/mrc/include/mrc/edge/edge_builder.hpp b/cpp/mrc/include/mrc/edge/edge_builder.hpp
index 78e88b577..e4bd0e571 100644
--- a/cpp/mrc/include/mrc/edge/edge_builder.hpp
+++ b/cpp/mrc/include/mrc/edge/edge_builder.hpp
@@ -285,8 +285,11 @@ class DeferredWritableMultiEdge : public MultiEdgeHolder<std::size_t, T>,
                                   public DeferredWritableMultiEdgeBase
 {
   public:
-    DeferredWritableMultiEdge(determine_indices_fn_t indices_fn = nullptr, bool deep_copy = false) :
-      m_indices_fn(std::move(indices_fn))
+    DeferredWritableMultiEdge(determine_indices_fn_t indices_fn = nullptr,
+                              bool deep_copy                    = false,
+                              std::string name                  = std::string()) :
+      m_indices_fn(std::move(indices_fn)),
+      MultiEdgeHolder<std::size_t, T>(std::move(name))
     {
         // // Generate warning if deep_copy = True but type does not support it
         // if constexpr (!std::is_copy_constructible_v<T>)
diff --git a/cpp/mrc/include/mrc/edge/edge_holder.hpp b/cpp/mrc/include/mrc/edge/edge_holder.hpp
index 0262a7e71..6486d955d 100644
--- a/cpp/mrc/include/mrc/edge/edge_holder.hpp
+++ b/cpp/mrc/include/mrc/edge/edge_holder.hpp
@@ -37,7 +37,9 @@
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <sstream>
 #include <stdexcept>
+#include <string>
 #include <type_traits>
 #include <typeindex>
 #include <utility>
@@ -50,7 +52,8 @@ template <typename T>
 class EdgeHolder
 {
   public:
-    EdgeHolder() = default;
+    EdgeHolder(std::string name = std::string()) : m_name(std::move(name)){};
+
     virtual ~EdgeHolder()
     {
         // Drop any edge connections before this object goes out of scope. This should execute any disconnectors
@@ -58,12 +61,35 @@ class EdgeHolder
 
         if (this->check_active_connection(false))
         {
-            LOG(FATAL) << "A node was destructed which still had dependent connections. Nodes must be kept alive while "
-                          "dependent connections are still active";
+            LOG(FATAL) << "EdgeHolder(" << this << ")[" << m_name << "] "
+                       << "A node was destructed which still had dependent connections. Nodes must be kept alive while "
+                          "dependent connections are still active\n"
+                       << this->connection_info();
         }
     }
 
+    const std::string& name() const
+    {
+        return m_name;
+    };
+
   protected:
+    std::string connection_info() const
+    {
+        std::stringstream ss;
+        ss << "m_owned_edge=" << m_owned_edge.lock() << "\tm_owned_edge_lifetime=" << m_owned_edge_lifetime
+           << "\tm_connected_edge=" << m_connected_edge;
+
+        bool is_connected = false;
+        if (m_connected_edge)
+        {
+            is_connected = m_connected_edge->is_connected();
+        }
+
+        ss << "\tis_connected=" << is_connected << "\tcheck_active_connection=" << this->check_active_connection(false);
+        return ss.str();
+    }
+
     bool check_active_connection(bool do_throw = true) const
     {
         // Alive connection exists when the lock is true, lifetime is false or a connction object has been set
@@ -155,6 +181,13 @@ class EdgeHolder
         m_connected_edge.reset();
     }
 
+    void disconnect()
+    {
+        m_connected_edge.reset();
+        m_owned_edge_lifetime.reset();
+        m_owned_edge.reset();
+    }
+
     const std::shared_ptr<Edge<T>>& get_connected_edge() const
     {
         return m_connected_edge;
@@ -188,6 +221,8 @@ class EdgeHolder
     // Holds a pointer to any set edge (different from init edge). Maintains lifetime
     std::shared_ptr<Edge<T>> m_connected_edge;
 
+    std::string m_name;
+
     // Allow edge builder to call set_edge
     friend EdgeBuilder;
 
@@ -200,10 +235,25 @@ template <typename KeyT, typename T>
 class MultiEdgeHolder
 {
   public:
-    MultiEdgeHolder()          = default;
+    MultiEdgeHolder(std::string name = std::string()) : m_name(std::move(name)){};
     virtual ~MultiEdgeHolder() = default;
 
+    const std::string& name() const
+    {
+        return m_name;
+    };
+
   protected:
+    std::string connection_info() const
+    {
+        std::stringstream ss;
+        ss << "m_edges.size()=" << m_edges.size();
+        for (const auto& [key, edge_pair] : m_edges)
+        {
+            ss << "\n\tkey=" << key << "\t" << edge_pair.connection_info();
+        }
+        return ss.str();
+    }
     void init_owned_edge(KeyT key, std::shared_ptr<Edge<T>> edge)
     {
         auto& edge_pair = this->get_edge_pair(key, true);
@@ -276,7 +326,9 @@ class MultiEdgeHolder
         {
             if (create_if_missing)
             {
-                m_edges[key] = EdgeHolder<T>();
+                std::ostringstream edge_name;
+                edge_name << m_name << "_" << key;
+                m_edges[key] = EdgeHolder<T>(edge_name.str());
                 return m_edges[key];
             }
 
@@ -321,6 +373,8 @@ class MultiEdgeHolder
     // Keeps pairs of get_edge/set_edge for each key
     std::map<KeyT, EdgeHolder<T>> m_edges;
 
+    std::string m_name;
+
     // Allow edge builder to call set_edge
     friend EdgeBuilder;
 };
diff --git a/cpp/mrc/include/mrc/manifold/composite_manifold.hpp b/cpp/mrc/include/mrc/manifold/composite_manifold.hpp
index 974729468..7d77c6285 100644
--- a/cpp/mrc/include/mrc/manifold/composite_manifold.hpp
+++ b/cpp/mrc/include/mrc/manifold/composite_manifold.hpp
@@ -59,6 +59,17 @@ class CompositeManifold : public Manifold
         mrc::make_edge(*m_ingress, *m_egress);
     }
 
+    ~CompositeManifold() override
+    {
+        shutdown();
+    };
+
+    void shutdown() final
+    {
+        m_ingress->shutdown();
+        m_egress->shutdown();
+    }
+
   protected:
     IngressT& ingress()
     {
diff --git a/cpp/mrc/include/mrc/manifold/egress.hpp b/cpp/mrc/include/mrc/manifold/egress.hpp
index 781122d61..51fe23f74 100644
--- a/cpp/mrc/include/mrc/manifold/egress.hpp
+++ b/cpp/mrc/include/mrc/manifold/egress.hpp
@@ -35,6 +35,7 @@ struct EgressDelegate
 {
     virtual ~EgressDelegate()                                                                        = default;
     virtual void add_output(const SegmentAddress& address, edge::IWritableProviderBase* output_sink) = 0;
+    virtual void shutdown(){};
 };
 
 template <typename T>
@@ -55,6 +56,13 @@ class TypedEgress : public EgressDelegate
 template <typename T>
 class RoundRobinEgress : public node::Router<SegmentAddress, T>, public TypedEgress<T>
 {
+  public:
+    void shutdown() final
+    {
+        DVLOG(10) << "Releasing edges from manifold egress";
+        node::Router<SegmentAddress, T>::release_edge_connections();
+    }
+
   protected:
     SegmentAddress determine_key_for_value(const T& t) override
     {
diff --git a/cpp/mrc/include/mrc/manifold/ingress.hpp b/cpp/mrc/include/mrc/manifold/ingress.hpp
index 060446b79..2842be4ba 100644
--- a/cpp/mrc/include/mrc/manifold/ingress.hpp
+++ b/cpp/mrc/include/mrc/manifold/ingress.hpp
@@ -23,6 +23,8 @@
 #include "mrc/node/sink_properties.hpp"
 #include "mrc/node/source_properties.hpp"
 
+#include <glog/logging.h>
+
 #include <memory>
 
 namespace mrc::manifold {
@@ -31,6 +33,7 @@ struct IngressDelegate
 {
     virtual ~IngressDelegate()                                                                       = default;
     virtual void add_input(const SegmentAddress& address, edge::IWritableAcceptorBase* input_source) = 0;
+    virtual void shutdown(){};
 };
 
 template <typename T>
@@ -51,6 +54,13 @@ class TypedIngress : public IngressDelegate
 template <typename T>
 class MuxedIngress : public node::Muxer<T>, public TypedIngress<T>
 {
+  public:
+    void shutdown() final
+    {
+        DVLOG(10) << "Releasing edges from manifold ingress";
+        node::SourceProperties<T>::release_edge_connection();
+    }
+
   protected:
     void do_add_input(const SegmentAddress& address, edge::IWritableAcceptor<T>* source) final
     {
diff --git a/cpp/mrc/include/mrc/manifold/interface.hpp b/cpp/mrc/include/mrc/manifold/interface.hpp
index 5c3d28fa4..706487091 100644
--- a/cpp/mrc/include/mrc/manifold/interface.hpp
+++ b/cpp/mrc/include/mrc/manifold/interface.hpp
@@ -27,9 +27,11 @@ struct Interface
     virtual ~Interface() = default;
 
     virtual const PortName& port_name() const = 0;
+    virtual const std::string& info() const   = 0;
 
-    virtual void start() = 0;
-    virtual void join()  = 0;
+    virtual void start()    = 0;
+    virtual void join()     = 0;
+    virtual void shutdown() = 0;
 
     virtual void add_input(const SegmentAddress& address, edge::IWritableAcceptorBase* input_source) = 0;
     virtual void add_output(const SegmentAddress& address, edge::IWritableProviderBase* output_sink) = 0;
diff --git a/cpp/mrc/include/mrc/manifold/manifold.hpp b/cpp/mrc/include/mrc/manifold/manifold.hpp
index 4cb567341..046934331 100644
--- a/cpp/mrc/include/mrc/manifold/manifold.hpp
+++ b/cpp/mrc/include/mrc/manifold/manifold.hpp
@@ -39,12 +39,12 @@ class Manifold : public Interface
     ~Manifold() override;
 
     const PortName& port_name() const final;
+    const std::string& info() const final;
+    void shutdown() override;
 
   protected:
     runnable::IRunnableResources& resources();
 
-    const std::string& info() const;
-
   private:
     void add_input(const SegmentAddress& address, edge::IWritableAcceptorBase* input_source) final;
     void add_output(const SegmentAddress& address, edge::IWritableProviderBase* output_sink) final;
diff --git a/cpp/mrc/include/mrc/node/generic_sink.hpp b/cpp/mrc/include/mrc/node/generic_sink.hpp
index de0fbec7c..05f96af6e 100644
--- a/cpp/mrc/include/mrc/node/generic_sink.hpp
+++ b/cpp/mrc/include/mrc/node/generic_sink.hpp
@@ -63,7 +63,7 @@ template <typename T>
 class GenericSinkComponent : public RxSinkComponent<T>
 {
   public:
-    GenericSinkComponent()
+    GenericSinkComponent(std::string name = std::string()) : RxSinkComponent<T>(std::move(name))
     {
         RxSinkComponent<T>::set_observer(rxcpp::make_observer_dynamic<T>(
             [this](T data) {
diff --git a/cpp/mrc/include/mrc/node/generic_source.hpp b/cpp/mrc/include/mrc/node/generic_source.hpp
index 19956d422..96406006c 100644
--- a/cpp/mrc/include/mrc/node/generic_source.hpp
+++ b/cpp/mrc/include/mrc/node/generic_source.hpp
@@ -71,7 +71,7 @@ template <typename T>
 class GenericSourceComponent : public ForwardingEgressProvider<T>
 {
   public:
-    GenericSourceComponent()           = default;
+    GenericSourceComponent(std::string name = std::string()) : m_name(std::move(name)) {}
     ~GenericSourceComponent() override = default;
 
   private:
@@ -81,6 +81,8 @@ class GenericSourceComponent : public ForwardingEgressProvider<T>
     }
 
     virtual mrc::channel::Status get_data(T& data) = 0;
+
+    std::string m_name;
 };
 
 template <typename T>
@@ -90,6 +92,10 @@ class LambdaSourceComponent : public GenericSourceComponent<T>
     using get_data_fn_t = std::function<mrc::channel::Status(T&)>;
 
     LambdaSourceComponent(get_data_fn_t get_data_fn) : m_get_data_fn(std::move(get_data_fn)) {}
+    LambdaSourceComponent(std::string name, get_data_fn_t get_data_fn) :
+      GenericSourceComponent<T>(std::move(name)),
+      m_get_data_fn(std::move(get_data_fn))
+    {}
     ~LambdaSourceComponent() override = default;
 
   private:
diff --git a/cpp/mrc/include/mrc/node/operators/broadcast.hpp b/cpp/mrc/include/mrc/node/operators/broadcast.hpp
index 553739c1c..dbb2c0598 100644
--- a/cpp/mrc/include/mrc/node/operators/broadcast.hpp
+++ b/cpp/mrc/include/mrc/node/operators/broadcast.hpp
@@ -34,6 +34,8 @@ namespace mrc::node {
 class BroadcastTypeless : public edge::IWritableProviderBase, public edge::IWritableAcceptorBase
 {
   public:
+    BroadcastTypeless(std::string name = std::string()) : m_name(std::move(name)) {}
+
     std::shared_ptr<edge::WritableEdgeHandle> get_writable_edge_handle() const override
     {
         auto* self = const_cast<BroadcastTypeless*>(this);
@@ -141,6 +143,7 @@ class BroadcastTypeless : public edge::IWritableProviderBase, public edge::IWrit
     }
 
   private:
+    std::string m_name;
     std::mutex m_mutex;
     std::vector<std::weak_ptr<edge::DeferredWritableMultiEdgeBase>> m_upstream_handles;
     std::vector<std::shared_ptr<edge::WritableEdgeHandle>> m_downstream_handles;
@@ -199,18 +202,18 @@ class Broadcast : public WritableProvider<T>, public edge::IWritableAcceptor<T>
 
     Broadcast(bool deep_copy = false)
     {
-        auto edge = std::make_shared<BroadcastEdge>(*this, deep_copy);
-
-        // Save to avoid casting
-        m_edge = edge;
+        init_edge(deep_copy);
+    }
 
-        WritableProvider<T>::init_owned_edge(edge);
+    Broadcast(std::string name, bool deep_copy = false) : m_name(std::move(name))
+    {
+        init_edge(deep_copy);
     }
 
     ~Broadcast()
     {
         // Debug print
-        VLOG(10) << "Destroying TestBroadcast";
+        VLOG(10) << "Destroying Broadcast " << m_name;
     }
 
     void set_writable_edge_handle(std::shared_ptr<edge::WritableEdgeHandle> ingress) override
@@ -227,11 +230,22 @@ class Broadcast : public WritableProvider<T>, public edge::IWritableAcceptor<T>
 
     void on_complete()
     {
-        VLOG(10) << "TestBroadcast completed";
+        VLOG(10) << "Broadcast completed " << m_name;
     }
 
   private:
+    void init_edge(bool deep_copy)
+    {
+        auto edge = std::make_shared<BroadcastEdge>(*this, deep_copy);
+
+        // Save to avoid casting
+        m_edge = edge;
+
+        WritableProvider<T>::init_owned_edge(edge);
+    }
+
     std::weak_ptr<BroadcastEdge> m_edge;
+    std::string m_name;
 };
 
 }  // namespace mrc::node
diff --git a/cpp/mrc/include/mrc/node/queue.hpp b/cpp/mrc/include/mrc/node/queue.hpp
index 81038feaf..2bb554203 100644
--- a/cpp/mrc/include/mrc/node/queue.hpp
+++ b/cpp/mrc/include/mrc/node/queue.hpp
@@ -23,17 +23,28 @@
 #include "mrc/node/sink_properties.hpp"
 #include "mrc/node/source_properties.hpp"
 
+#include <string>
+
 namespace mrc::node {
 
+using namespace std::string_literals;
+
 template <typename T>
 class Queue : public WritableProvider<T>, public ReadableProvider<T>
 {
   public:
-    Queue()
+    Queue(const std::string& name = std::string()) :
+      SinkProperties<T>(name + "-sink"s),
+      SourceProperties<T>(name + "-source"s)
     {
         this->set_channel(std::make_unique<mrc::channel::BufferedChannel<T>>());
     }
-    ~Queue() override = default;
+
+    ~Queue() override
+    {
+        SinkProperties<T>::release_edge_connection();
+        SourceProperties<T>::release_edge_connection();
+    };
 
     void set_channel(std::unique_ptr<mrc::channel::Channel<T>> channel)
     {
diff --git a/cpp/mrc/include/mrc/node/rx_node.hpp b/cpp/mrc/include/mrc/node/rx_node.hpp
index 34a48fd0d..a467dee9d 100644
--- a/cpp/mrc/include/mrc/node/rx_node.hpp
+++ b/cpp/mrc/include/mrc/node/rx_node.hpp
@@ -36,6 +36,7 @@
 #include <exception>
 #include <memory>
 #include <mutex>
+#include <string>
 
 namespace mrc::node {
 
@@ -50,11 +51,14 @@ class RxNode : public RxSinkBase<InputT>,
     // function defining the stream, i.e. operations linking Sink -> Source
     using stream_fn_t = std::function<rxcpp::observable<OutputT>(const rxcpp::observable<InputT>&)>;
 
-    RxNode();
+    RxNode(std::string name = std::string());
 
     template <typename... OpsT>
     RxNode(OpsT&&... ops);
 
+    template <typename... OpsT>
+    RxNode(std::string name, OpsT&&... ops);
+
     ~RxNode() override = default;
 
     template <typename... OpsT>
@@ -68,13 +72,20 @@ class RxNode : public RxSinkBase<InputT>,
 
     void make_stream(stream_fn_t fn);
 
+    void set_name(std::string name)
+    {
+        RxSinkBase<InputT>::m_name    = name;
+        RxSourceBase<OutputT>::m_name = std::move(name);
+    }
+
+    void on_shutdown_critical_section() final;
+
   private:
     // the following method(s) are moved to private from their original scopes to prevent access from deriving classes
     using RxSinkBase<InputT>::observable;
     using RxSourceBase<OutputT>::observer;
 
     void do_subscribe(rxcpp::composite_subscription& subscription) final;
-    void on_shutdown_critical_section() final;
 
     void on_stop(const rxcpp::subscription& subscription) override;
     void on_kill(const rxcpp::subscription& subscription) final;
@@ -85,7 +96,11 @@ class RxNode : public RxSinkBase<InputT>,
 };
 
 template <typename InputT, typename OutputT, typename ContextT>
-RxNode<InputT, OutputT, ContextT>::RxNode() :
+RxNode<InputT, OutputT, ContextT>::RxNode(std::string name) :
+  RxSinkBase<InputT>{name},
+  RxSourceBase<OutputT>{name},
+  SinkProperties<InputT>(name),
+  SourceProperties<OutputT>(name),
   m_stream([](const rxcpp::observable<InputT>& obs) {
       // Default to just returning the input
       return obs;
@@ -99,6 +114,15 @@ RxNode<InputT, OutputT, ContextT>::RxNode(OpsT&&... ops)
     pipe(std::forward<OpsT>(ops)...);
 }
 
+template <typename InputT, typename OutputT, typename ContextT>
+template <typename... OpsT>
+RxNode<InputT, OutputT, ContextT>::RxNode(std::string name, OpsT&&... ops) :
+  RxSinkBase<InputT>{name},
+  RxSourceBase<OutputT>{name}
+{
+    pipe(std::forward<OpsT>(ops)...);
+}
+
 template <typename InputT, typename OutputT, typename ContextT>
 void RxNode<InputT, OutputT, ContextT>::make_stream(stream_fn_t fn)
 {
@@ -140,7 +164,7 @@ void RxNode<InputT, OutputT, ContextT>::on_kill(const rxcpp::subscription& subsc
 template <typename InputT, typename OutputT, typename ContextT>
 void RxNode<InputT, OutputT, ContextT>::on_shutdown_critical_section()
 {
-    DVLOG(10) << runnable::Context::get_runtime_context().info() << " releasing source channel";
+    DVLOG(10) << "releasing source channel";
     RxSourceBase<OutputT>::release_edge_connection();
 }
 
@@ -182,7 +206,7 @@ class RxNodeComponent : public WritableProvider<InputT>, public WritableAcceptor
   public:
     using stream_fn_t = std::function<rxcpp::observable<OutputT>(const rxcpp::observable<InputT>&)>;
 
-    RxNodeComponent()
+    RxNodeComponent(std::string name = std::string()) : m_name(std::move(name))
     {
         auto edge = std::make_shared<EdgeRxSubscriber<InputT>>(m_subject.get_subscriber());
 
@@ -194,12 +218,23 @@ class RxNodeComponent : public WritableProvider<InputT>, public WritableAcceptor
         this->make_stream(stream_fn);
     }
 
+    RxNodeComponent(std::string name, stream_fn_t stream_fn) : RxNodeComponent(std::move(name))
+    {
+        this->make_stream(stream_fn);
+    }
+
     template <typename... OpsT>
     RxNodeComponent(OpsT&&... ops) : RxNodeComponent()
     {
         this->pipe(std::forward<OpsT>(ops)...);
     }
 
+    template <typename... OpsT>
+    RxNodeComponent(std::string name, OpsT&&... ops) : RxNodeComponent(std::move(name))
+    {
+        this->pipe(std::forward<OpsT>(ops)...);
+    }
+
     template <typename... OpsT>
     RxNodeComponent& pipe(OpsT&&... ops)
     {
@@ -247,6 +282,7 @@ class RxNodeComponent : public WritableProvider<InputT>, public WritableAcceptor
   private:
     rxcpp::subjects::subject<InputT> m_subject;
     rxcpp::subscription m_subject_subscription;
+    std::string m_name;
 };
 
 }  // namespace mrc::node
diff --git a/cpp/mrc/include/mrc/node/rx_sink.hpp b/cpp/mrc/include/mrc/node/rx_sink.hpp
index 43c7fc2dd..b93f65a07 100644
--- a/cpp/mrc/include/mrc/node/rx_sink.hpp
+++ b/cpp/mrc/include/mrc/node/rx_sink.hpp
@@ -84,7 +84,8 @@ class RxSink : public RxSinkBase<T>, public RxRunnable<ContextT>, public RxProlo
     using on_error_fn_t    = std::function<void(std::exception_ptr)>;
     using on_complete_fn_t = std::function<void()>;
 
-    RxSink()           = default;
+    RxSink(std::string name = std::string()) : RxSinkBase<T>(name), SinkProperties<T>(name) {}
+
     ~RxSink() override = default;
 
     template <typename... ArgsT>
@@ -93,6 +94,12 @@ class RxSink : public RxSinkBase<T>, public RxRunnable<ContextT>, public RxProlo
         set_observer(std::forward<ArgsT>(args)...);
     }
 
+    template <typename... ArgsT>
+    RxSink(std::string name, ArgsT&&... args) : RxSinkBase<T>(name), SinkProperties<T>(name)
+    {
+        set_observer(std::forward<ArgsT>(args)...);
+    }
+
     void set_observer(observer_t observer);
 
     template <typename... ArgsT>
@@ -174,11 +181,12 @@ class RxSinkComponent : public WritableProvider<T>
 
     RxSinkComponent()
     {
-        auto edge = std::make_shared<EdgeRxObserver<T>>();
-
-        m_sink_edge = edge;
+        init_edge();
+    }
 
-        WritableProvider<T>::init_owned_edge(edge);
+    RxSinkComponent(std::string name = std::string()) : m_name(std::move(name))
+    {
+        init_edge();
     }
 
     ~RxSinkComponent() = default;
@@ -186,12 +194,12 @@ class RxSinkComponent : public WritableProvider<T>
     template <typename... ArgsT>
     RxSinkComponent(ArgsT&&... args) : RxSinkComponent()
     {
-        // auto edge = std::make_shared<EdgeRxObserver<T>>();
-
-        // m_sink_edge = edge;
-
-        // WritableProvider<T>::init_owned_edge(edge);
+        set_observer(std::forward<ArgsT>(args)...);
+    }
 
+    template <typename... ArgsT>
+    RxSinkComponent(std::string name, ArgsT&&... args) : RxSinkComponent(std::move(name))
+    {
         set_observer(std::forward<ArgsT>(args)...);
     }
 
@@ -204,7 +212,17 @@ class RxSinkComponent : public WritableProvider<T>
     }
 
   private:
+    void init_edge()
+    {
+        auto edge = std::make_shared<EdgeRxObserver<T>>();
+
+        m_sink_edge = edge;
+
+        WritableProvider<T>::init_owned_edge(edge);
+    }
+
     std::weak_ptr<EdgeRxObserver<T>> m_sink_edge;
+    std::string m_name;
     // observer_t m_observer;
 };
 
diff --git a/cpp/mrc/include/mrc/node/rx_sink_base.hpp b/cpp/mrc/include/mrc/node/rx_sink_base.hpp
index 89d4a8c8f..294d0b98f 100644
--- a/cpp/mrc/include/mrc/node/rx_sink_base.hpp
+++ b/cpp/mrc/include/mrc/node/rx_sink_base.hpp
@@ -47,10 +47,11 @@ class RxSinkBase : public WritableProvider<T>, public ReadableAcceptor<T>, publi
     void sink_remove_watcher(std::shared_ptr<WatcherInterface> watcher);
 
   protected:
-    RxSinkBase();
+    RxSinkBase(std::string name = std::string());
     ~RxSinkBase() override = default;
 
     const rxcpp::observable<T>& observable() const;
+    std::string m_name;
 
   private:
     // this is our channel reader progress engine
@@ -61,7 +62,9 @@ class RxSinkBase : public WritableProvider<T>, public ReadableAcceptor<T>, publi
 };
 
 template <typename T>
-RxSinkBase<T>::RxSinkBase() :
+RxSinkBase<T>::RxSinkBase(std::string name) :
+  SinkChannelOwner<T>(name),
+  m_name(std::move(name)),
   m_observable(rxcpp::observable<>::create<T>([this](rxcpp::subscriber<T> s) {
       progress_engine(s);
   }))
diff --git a/cpp/mrc/include/mrc/node/rx_source.hpp b/cpp/mrc/include/mrc/node/rx_source.hpp
index 034d669ff..6e90623a3 100644
--- a/cpp/mrc/include/mrc/node/rx_source.hpp
+++ b/cpp/mrc/include/mrc/node/rx_source.hpp
@@ -38,6 +38,7 @@
 #include <memory>
 #include <mutex>
 #include <string>
+#include <utility>
 #include <vector>
 
 namespace mrc::node {
@@ -51,8 +52,9 @@ template <typename T, typename ContextT>
 class RxSource : public RxSourceBase<T>, public RxRunnable<ContextT>, public RxEpilogueTap<T>
 {
   public:
-    RxSource() = default;
+    RxSource(std::string name = std::string()) : RxSourceBase<T>(name), SourceProperties<T>(name) {}
     RxSource(rxcpp::observable<T> observable);
+    RxSource(std::string name, rxcpp::observable<T> observable);
     ~RxSource() override = default;
 
     void set_observable(rxcpp::observable<T> observable);
@@ -72,6 +74,14 @@ RxSource<T, ContextT>::RxSource(rxcpp::observable<T> observable) : RxSourceBase<
     set_observable(observable);
 }
 
+template <typename T, typename ContextT>
+RxSource<T, ContextT>::RxSource(std::string name, rxcpp::observable<T> observable) :
+  RxSourceBase<T>(name),
+  SourceProperties<T>(name)
+{
+    set_observable(observable);
+}
+
 template <typename T, typename ContextT>
 void RxSource<T, ContextT>::on_shutdown_critical_section()
 {
diff --git a/cpp/mrc/include/mrc/node/rx_source_base.hpp b/cpp/mrc/include/mrc/node/rx_source_base.hpp
index 58876d3c6..4af410e71 100644
--- a/cpp/mrc/include/mrc/node/rx_source_base.hpp
+++ b/cpp/mrc/include/mrc/node/rx_source_base.hpp
@@ -55,10 +55,11 @@ class RxSourceBase : public ReadableProvider<T>,
     void source_remove_watcher(std::shared_ptr<WatcherInterface> watcher);
 
   protected:
-    RxSourceBase();
+    RxSourceBase(std::string name = std::string());
     ~RxSourceBase() override = default;
 
     const rxcpp::observer<T>& observer() const;
+    std::string m_name;
 
   private:
     // // the following methods are moved to private from their original scopes to prevent access from deriving classes
@@ -68,7 +69,9 @@ class RxSourceBase : public ReadableProvider<T>,
 };
 
 template <typename T>
-RxSourceBase<T>::RxSourceBase() :
+RxSourceBase<T>::RxSourceBase(std::string name) :
+  SourceChannelOwner<T>(name),
+  m_name(std::move(name)),
   m_observer(rxcpp::make_observer_dynamic<T>(
       [this](T data) {
           this->watcher_epilogue(WatchableEvent::sink_on_data, true, &data);
diff --git a/cpp/mrc/include/mrc/node/sink_channel_owner.hpp b/cpp/mrc/include/mrc/node/sink_channel_owner.hpp
index 8997e3a8d..712afc014 100644
--- a/cpp/mrc/include/mrc/node/sink_channel_owner.hpp
+++ b/cpp/mrc/include/mrc/node/sink_channel_owner.hpp
@@ -42,7 +42,7 @@ class SinkChannelOwner : public virtual SinkProperties<T>
     }
 
   protected:
-    SinkChannelOwner() = default;
+    SinkChannelOwner(std::string name = std::string()) : SinkProperties<T>(std::move(name)){};
 
     void do_set_channel(edge::EdgeChannel<T>& edge_channel)
     {
diff --git a/cpp/mrc/include/mrc/node/sink_properties.hpp b/cpp/mrc/include/mrc/node/sink_properties.hpp
index af56b0fe0..5cad278da 100644
--- a/cpp/mrc/include/mrc/node/sink_properties.hpp
+++ b/cpp/mrc/include/mrc/node/sink_properties.hpp
@@ -26,6 +26,7 @@
 
 #include <memory>
 #include <stdexcept>
+#include <string>
 #include <typeindex>
 
 namespace mrc::node {
@@ -109,7 +110,7 @@ class SinkProperties : public edge::EdgeHolder<T>, public SinkPropertiesBase
     }
 
   protected:
-    SinkProperties()
+    SinkProperties(std::string name = std::string()) : edge::EdgeHolder<T>(std::move(name))
     {
         // Set the default edge to be a null one in case no connection is made
         this->init_connected_edge(std::make_shared<NullReadableEdge<T>>());
diff --git a/cpp/mrc/include/mrc/node/source_channel_owner.hpp b/cpp/mrc/include/mrc/node/source_channel_owner.hpp
index 226492e5e..823f9f21b 100644
--- a/cpp/mrc/include/mrc/node/source_channel_owner.hpp
+++ b/cpp/mrc/include/mrc/node/source_channel_owner.hpp
@@ -34,6 +34,7 @@ template <typename T>
 class SourceChannelOwner : public virtual SourceProperties<T>
 {
   public:
+    SourceChannelOwner(std::string name = std::string()) : SourceProperties<T>(std::move(name)){};
     ~SourceChannelOwner() override = default;
 
     void set_channel(std::unique_ptr<mrc::channel::Channel<T>> channel)
@@ -44,8 +45,6 @@ class SourceChannelOwner : public virtual SourceProperties<T>
     }
 
   protected:
-    SourceChannelOwner() = default;
-
     void do_set_channel(edge::EdgeChannel<T>& edge_channel)
     {
         // Create 2 edges, one for reading and writing. On connection, persist the other to allow the node to still use
diff --git a/cpp/mrc/include/mrc/node/source_properties.hpp b/cpp/mrc/include/mrc/node/source_properties.hpp
index 12166eb43..44ef87b11 100644
--- a/cpp/mrc/include/mrc/node/source_properties.hpp
+++ b/cpp/mrc/include/mrc/node/source_properties.hpp
@@ -23,6 +23,7 @@
 #include "mrc/edge/edge_writable.hpp"
 #include "mrc/node/forward.hpp"
 #include "mrc/type_traits.hpp"
+#include "mrc/types.hpp"  // for Mutex
 #include "mrc/utils/type_utils.hpp"
 
 #include <memory>
@@ -111,7 +112,7 @@ class SourceProperties : public edge::EdgeHolder<T>, public SourcePropertiesBase
     }
 
   protected:
-    SourceProperties()
+    SourceProperties(std::string name = std::string()) : edge::EdgeHolder<T>(std::move(name))
     {
         // Set the default edge to be a null one in case no connection is made
         this->init_connected_edge(std::make_shared<NullWritableEdge<T>>());
@@ -208,37 +209,68 @@ template <typename T>
 class ForwardingEgressProvider : public ReadableProvider<T>
 {
   protected:
+    struct State
+    {
+        Mutex m_mutex;
+        bool m_is_destroyed{false};
+    };
+
     class ForwardingEdge : public edge::IEdgeReadable<T>
     {
       public:
-        ForwardingEdge(ForwardingEgressProvider<T>& parent) : m_parent(parent) {}
+        ForwardingEdge(ForwardingEgressProvider<T>& parent, std::shared_ptr<State> state) :
+          m_parent(parent),
+          m_state(std::move(state))
+        {}
 
         ~ForwardingEdge() = default;
 
         channel::Status await_read(T& t) override
         {
-            return m_parent.get_next(t);
+            std::lock_guard<decltype(m_state->m_mutex)> lock(m_state->m_mutex);
+            if (!(m_state->m_is_destroyed))
+            {
+                return m_parent.get_next(t);
+            }
+
+            return channel::Status::closed;
         }
 
       private:
         ForwardingEgressProvider<T>& m_parent;
+        std::shared_ptr<State> m_state;
     };
 
-    ForwardingEgressProvider()
+    ForwardingEgressProvider() : m_state(std::make_shared<State>())
     {
-        auto inner_edge = std::make_shared<ForwardingEdge>(*this);
+        auto inner_edge = std::make_shared<ForwardingEdge>(*this, m_state);
 
-        inner_edge->add_disconnector([this]() {
-            // Only call the on_complete if we have been connected
-            this->on_complete();
+        inner_edge->add_disconnector([this, state = m_state]() {
+            std::lock_guard<decltype(state->m_mutex)> lock(state->m_mutex);
+            if (!(state->m_is_destroyed))
+            {
+                // Only call the on_complete if we have been connected and `this` is still alive
+                this->on_complete();
+            }
         });
 
         ReadableProvider<T>::init_owned_edge(inner_edge);
     }
 
+    ~ForwardingEgressProvider()
+    {
+        SourceProperties<T>::disconnect();
+        {
+            std::lock_guard<decltype(m_state->m_mutex)> lock(m_state->m_mutex);
+            m_state->m_is_destroyed = true;
+        }
+    }
+
     virtual channel::Status get_next(T& t) = 0;
 
     virtual void on_complete() {}
+
+    std::shared_ptr<State> m_state;
 };
 
 }  // namespace mrc::node
diff --git a/cpp/mrc/include/mrc/segment/builder.hpp b/cpp/mrc/include/mrc/segment/builder.hpp
index a35f571c9..46bd756ab 100644
--- a/cpp/mrc/include/mrc/segment/builder.hpp
+++ b/cpp/mrc/include/mrc/segment/builder.hpp
@@ -364,7 +364,7 @@ class IBuilder
 template <typename ObjectT, typename... ArgsT>
 std::shared_ptr<Object<ObjectT>> IBuilder::construct_object(std::string name, ArgsT&&... args)
 {
-    auto uptr = std::make_unique<ObjectT>(std::forward<ArgsT>(args)...);
+    auto uptr = std::make_unique<ObjectT>(name, std::forward<ArgsT>(args)...);
 
     return make_object(std::move(name), std::move(uptr));
 }
@@ -376,13 +376,13 @@ std::shared_ptr<Object<ObjectT>> IBuilder::make_object(std::string name, std::un
 
     if constexpr (std::is_base_of_v<runnable::Runnable, ObjectT>)
     {
-        segment_object = std::make_shared<Runnable<ObjectT>>(std::move(node));
-        this->add_object(name, segment_object);
+        segment_object = std::make_shared<Runnable<ObjectT>>(name, std::move(node));
+        this->add_object(std::move(name), segment_object);
     }
     else
     {
-        segment_object = std::make_shared<Component<ObjectT>>(std::move(node));
-        this->add_object(name, segment_object);
+        segment_object = std::make_shared<Component<ObjectT>>(name, std::move(node));
+        this->add_object(std::move(name), segment_object);
     }
 
     CHECK(segment_object);
diff --git a/cpp/mrc/include/mrc/segment/component.hpp b/cpp/mrc/include/mrc/segment/component.hpp
index 3e25f9b63..adef7b7dc 100644
--- a/cpp/mrc/include/mrc/segment/component.hpp
+++ b/cpp/mrc/include/mrc/segment/component.hpp
@@ -23,6 +23,7 @@
 
 #include <memory>
 #include <ostream>
+#include <string>
 #include <utility>
 
 namespace mrc::segment {
@@ -32,8 +33,17 @@ class Component final : public Object<ResourceT>
 {
   public:
     Component(std::unique_ptr<ResourceT> resource) : m_resource(std::move(resource)) {}
+    Component(std::string name, std::unique_ptr<ResourceT> resource) :
+      m_resource(std::move(resource)),
+      m_name(std::move(name))
+    {}
     ~Component() final = default;
 
+    void destroy() final
+    {
+        m_resource.reset();
+    }
+
   private:
     ResourceT* get_object() const final
     {
@@ -42,6 +52,7 @@ class Component final : public Object<ResourceT>
     }
 
     std::unique_ptr<ResourceT> m_resource;
+    std::string m_name;
 };
 
 }  // namespace mrc::segment
diff --git a/cpp/mrc/include/mrc/segment/ingress_port.hpp b/cpp/mrc/include/mrc/segment/ingress_port.hpp
index fec6d469e..50b796ef8 100644
--- a/cpp/mrc/include/mrc/segment/ingress_port.hpp
+++ b/cpp/mrc/include/mrc/segment/ingress_port.hpp
@@ -85,6 +85,13 @@ class IngressPort : public Object<node::RxSourceBase<T>>, public IngressPortBase
         manifold->add_output(m_segment_address, m_source.get());
     }
 
+    void destroy() final
+    {
+        DVLOG(10) << "Destroying ingress port " << this->type_name();
+        m_source->on_shutdown_critical_section();
+        m_source.reset();
+    }
+
     SegmentAddress m_segment_address;
     PortName m_port_name;
     std::unique_ptr<node::RxNode<T>> m_source;
diff --git a/cpp/mrc/include/mrc/segment/object.hpp b/cpp/mrc/include/mrc/segment/object.hpp
index 2ccc80094..6e6652797 100644
--- a/cpp/mrc/include/mrc/segment/object.hpp
+++ b/cpp/mrc/include/mrc/segment/object.hpp
@@ -74,6 +74,8 @@ struct ObjectProperties
 
     virtual runnable::LaunchOptions& launch_options()             = 0;
     virtual const runnable::LaunchOptions& launch_options() const = 0;
+
+    virtual void destroy(){};
 };
 
 inline ObjectProperties::~ObjectProperties() = default;
@@ -152,6 +154,8 @@ template <typename ObjectT>
 class Object : public virtual ObjectProperties
 {
   public:
+    Object(std::string name = std::string()) : m_name(std::move(name)){};
+
     ObjectT& object();
 
     std::string name() const final;
diff --git a/cpp/mrc/include/mrc/segment/runnable.hpp b/cpp/mrc/include/mrc/segment/runnable.hpp
index ab5b590ca..e582e649e 100644
--- a/cpp/mrc/include/mrc/segment/runnable.hpp
+++ b/cpp/mrc/include/mrc/segment/runnable.hpp
@@ -41,11 +41,28 @@ class Runnable : public Object<NodeT>, public runnable::Launchable
     Runnable(ArgsT&&... args) : m_node(std::make_unique<NodeT>(std::forward<ArgsT>(args)...))
     {}
 
+    template <typename... ArgsT>
+    Runnable(std::string name, ArgsT&&... args) :
+      Object<NodeT>(std::move(name)),
+      m_node(std::make_unique<NodeT>(std::forward<ArgsT>(args)...))
+    {}
+
     Runnable(std::unique_ptr<NodeT> node) : m_node(std::move(node))
     {
         CHECK(m_node);
     }
 
+    void destroy() final
+    {
+        DVLOG(10) << "Destroying runnable " << this->type_name();
+        m_node.reset();
+    }
+
+    Runnable(std::string name, std::unique_ptr<NodeT> node) : Object<NodeT>(std::move(name)), m_node(std::move(node))
+    {
+        CHECK(m_node);
+    }
+
   private:
     NodeT* get_object() const final;
     std::unique_ptr<runnable::Launcher> prepare_launcher(runnable::LaunchControl& launch_control) final;
diff --git a/cpp/mrc/src/internal/pipeline/pipeline_instance.cpp b/cpp/mrc/src/internal/pipeline/pipeline_instance.cpp
index dddd73a3c..e88d4dac5 100644
--- a/cpp/mrc/src/internal/pipeline/pipeline_instance.cpp
+++ b/cpp/mrc/src/internal/pipeline/pipeline_instance.cpp
@@ -37,6 +37,7 @@
 
 #include <exception>
 #include <memory>
+#include <mutex>  // for lock_guard
 #include <ostream>
 #include <string>
 #include <utility>
@@ -89,7 +90,7 @@ void PipelineInstance::join_segment(const SegmentAddress& address)
     search->second->service_await_join();
 }
 
-void PipelineInstance::stop_segment(const SegmentAddress& address)
+void PipelineInstance::stop_segment(const SegmentAddress& address, bool kill)
 {
     auto search = m_segments.find(address);
     CHECK(search != m_segments.end());
@@ -97,10 +98,9 @@ void PipelineInstance::stop_segment(const SegmentAddress& address)
     auto [id, rank]    = segment_address_decode(address);
     const auto& segdef = m_definition->find_segment(id);
 
-    for (const auto& name : segdef->ingress_port_names())
+    if (kill)
     {
-        DVLOG(3) << "Dropping IngressPort for " << ::mrc::segment::info(address) << " on manifold " << name;
-        // manifold(name).drop_output(address);
+        search->second->shutdown();
     }
 
     search->second->service_stop();
@@ -111,6 +111,7 @@ void PipelineInstance::create_segment(const SegmentAddress& address, std::uint32
     // perform our allocations on the numa domain of the intended target
     // CHECK_LT(partition_id, m_resources->host_resources().size());
     CHECK_LT(partition_id, resources().partition_count());
+    DVLOG(10) << "Enqueing Creation of segment " << ::mrc::segment::info(address);
     resources()
         .partition(partition_id)
         .runnable()
@@ -121,7 +122,8 @@ void PipelineInstance::create_segment(const SegmentAddress& address, std::uint32
 
             auto [id, rank] = segment_address_decode(address);
             auto definition = m_definition->find_segment(id);
-            auto segment    = std::make_unique<segment::SegmentInstance>(definition, rank, *this, partition_id);
+            DVLOG(10) << "Creating segment " << definition->name() << " - " << ::mrc::segment::info(address);
+            auto segment = std::make_unique<segment::SegmentInstance>(definition, rank, *this, partition_id);
 
             for (const auto& name : definition->egress_port_names())
             {
@@ -149,6 +151,7 @@ void PipelineInstance::create_segment(const SegmentAddress& address, std::uint32
                 segment->attach_manifold(manifold);
             }
 
+            DVLOG(10) << "Created segment " << definition->name() << " - " << ::mrc::segment::info(address);
             m_segments[address] = std::move(segment);
         })
         .get();
@@ -200,12 +203,25 @@ void PipelineInstance::do_service_stop()
 
 void PipelineInstance::do_service_kill()
 {
+    std::lock_guard<decltype(m_kill_mux)> guard(m_kill_mux);
+    DVLOG(10) << "pipeline::PipelineInstance - killing " << m_manifolds.size() << " manifolds - " << m_segments.size()
+              << " segments";
     mark_joinable();
+    for (const auto& [name, manifold] : m_manifolds)
+    {
+        DVLOG(10) << "pipeline::PipelineInstance - killing manifold " << name;
+        manifold->shutdown();
+    }
+
+    m_manifolds.clear();
+
     for (auto& [id, segment] : m_segments)
     {
-        stop_segment(id);
+        stop_segment(id, true);
         segment->service_kill();
     }
+
+    m_segments.clear();
 }
 
 void PipelineInstance::do_service_await_join()
@@ -228,6 +244,7 @@ void PipelineInstance::do_service_await_join()
     if (first_exception)
     {
         LOG(ERROR) << "pipeline::PipelineInstance - an exception was caught while awaiting on segments - rethrowing";
+        do_service_kill();
         std::rethrow_exception(std::move(first_exception));
     }
 }
diff --git a/cpp/mrc/src/internal/pipeline/pipeline_instance.hpp b/cpp/mrc/src/internal/pipeline/pipeline_instance.hpp
index 7dc51e38e..d17b88a9b 100644
--- a/cpp/mrc/src/internal/pipeline/pipeline_instance.hpp
+++ b/cpp/mrc/src/internal/pipeline/pipeline_instance.hpp
@@ -51,7 +51,7 @@ class PipelineInstance final : public Service, public PipelineResources
     // we need to stage those object that are created into some struct/container so we can mass start them after all
     // object have been created
     void create_segment(const SegmentAddress& address, std::uint32_t partition_id);
-    void stop_segment(const SegmentAddress& address);
+    void stop_segment(const SegmentAddress& address, bool kill = false);
     void join_segment(const SegmentAddress& address);
     void remove_segment(const SegmentAddress& address);
 
@@ -85,6 +85,7 @@ class PipelineInstance final : public Service, public PipelineResources
     bool m_joinable{false};
     Promise<void> m_joinable_promise;
     SharedFuture<void> m_joinable_future;
+    Mutex m_kill_mux;
 };
 
 }  // namespace mrc::pipeline
diff --git a/cpp/mrc/src/internal/segment/builder_definition.cpp b/cpp/mrc/src/internal/segment/builder_definition.cpp
index b11614328..74dca81fe 100644
--- a/cpp/mrc/src/internal/segment/builder_definition.cpp
+++ b/cpp/mrc/src/internal/segment/builder_definition.cpp
@@ -28,7 +28,7 @@
 #include "mrc/modules/properties/persistent.hpp"  // IWYU pragma: keep
 #include "mrc/modules/segment_modules.hpp"
 #include "mrc/node/port_registry.hpp"
-#include "mrc/runnable/launchable.hpp"
+#include "mrc/runnable/launchable.hpp"   // for Launchable
 #include "mrc/segment/egress_port.hpp"   // IWYU pragma: keep
 #include "mrc/segment/ingress_port.hpp"  // IWYU pragma: keep
 #include "mrc/segment/object.hpp"
@@ -284,11 +284,32 @@ void BuilderDefinition::initialize()
                    << ", Segment Rank: " << m_rank << ". Exception message:\n"
                    << e.what();
 
-        // Rethrow after logging
+        shutdown();
+        //  Rethrow after logging
         std::rethrow_exception(std::current_exception());
     }
 }
 
+void BuilderDefinition::shutdown()
+{
+    DVLOG(10) << "Shutting down segment: " << m_definition->name();
+    for (auto& [name, obj_prop] : m_objects)
+    {
+        if (obj_prop->is_source() && !obj_prop->is_sink())
+        {
+            DVLOG(10) << "Destroying: " << name;
+            obj_prop->destroy();
+        }
+    }
+
+    m_ingress_ports.clear();
+    m_egress_ports.clear();
+    m_nodes.clear();
+    m_objects.clear();
+
+    DVLOG(10) << "Shutting down segment: " << m_definition->name() << " - done";
+}
+
 const std::map<std::string, std::shared_ptr<mrc::runnable::Launchable>>& BuilderDefinition::nodes() const
 {
     return m_nodes;
diff --git a/cpp/mrc/src/internal/segment/builder_definition.hpp b/cpp/mrc/src/internal/segment/builder_definition.hpp
index aa0c96140..ff2adc8b4 100644
--- a/cpp/mrc/src/internal/segment/builder_definition.hpp
+++ b/cpp/mrc/src/internal/segment/builder_definition.hpp
@@ -119,6 +119,7 @@ class BuilderDefinition : public IBuilder
     const SegmentDefinition& definition() const;
 
     void initialize();
+    void shutdown();
 
     const std::map<std::string, std::shared_ptr<runnable::Launchable>>& nodes() const;
     const std::map<std::string, std::shared_ptr<EgressPortBase>>& egress_ports() const;
diff --git a/cpp/mrc/src/internal/segment/segment_instance.cpp b/cpp/mrc/src/internal/segment/segment_instance.cpp
index 53f66b804..cbbdb0acb 100644
--- a/cpp/mrc/src/internal/segment/segment_instance.cpp
+++ b/cpp/mrc/src/internal/segment/segment_instance.cpp
@@ -331,4 +331,12 @@ std::shared_ptr<manifold::Interface> SegmentInstance::create_manifold(const Port
     return nullptr;
 }
 
+void SegmentInstance::shutdown()
+{
+    std::lock_guard<decltype(m_mutex)> lock(m_mutex);
+    DVLOG(10) << m_name << " - " << info() << " - shutting down segment";
+    do_service_kill();
+    m_builder->shutdown();
+}
+
 }  // namespace mrc::segment
diff --git a/cpp/mrc/src/internal/segment/segment_instance.hpp b/cpp/mrc/src/internal/segment/segment_instance.hpp
index addd38dd1..7d18fdd4e 100644
--- a/cpp/mrc/src/internal/segment/segment_instance.hpp
+++ b/cpp/mrc/src/internal/segment/segment_instance.hpp
@@ -56,6 +56,7 @@ class SegmentInstance final : public Service
 
     std::shared_ptr<manifold::Interface> create_manifold(const PortName& name);
     void attach_manifold(std::shared_ptr<manifold::Interface> manifold);
+    void shutdown();
 
   protected:
     const std::string& info() const;
diff --git a/cpp/mrc/src/public/manifold/manifold.cpp b/cpp/mrc/src/public/manifold/manifold.cpp
index a1a3cca25..3b4aba3eb 100644
--- a/cpp/mrc/src/public/manifold/manifold.cpp
+++ b/cpp/mrc/src/public/manifold/manifold.cpp
@@ -50,6 +50,8 @@ const std::string& Manifold::info() const
     return m_info;
 }
 
+void Manifold::shutdown() {}
+
 void Manifold::add_input(const SegmentAddress& address, edge::IWritableAcceptorBase* input_source)
 {
     DVLOG(3) << "manifold " << this->port_name() << ": connecting to upstream segment " << segment::info(address);
diff --git a/cpp/mrc/src/tests/test_pipeline.cpp b/cpp/mrc/src/tests/test_pipeline.cpp
index 0f23a6fa2..3785b7444 100644
--- a/cpp/mrc/src/tests/test_pipeline.cpp
+++ b/cpp/mrc/src/tests/test_pipeline.cpp
@@ -199,7 +199,7 @@ TEST_F(TestPipeline, Queue)
 
     auto segment = pipeline->make_segment("seg_1", [](segment::IBuilder& s) {
         auto source = s.make_object("source", test::nodes::infinite_int_rx_source());
-        auto queue  = s.make_object("queue", std::make_unique<node::Queue<int>>());
+        auto queue  = s.make_object("queue", std::make_unique<node::Queue<int>>("queue"));
         auto sink   = s.make_object("sink", test::nodes::int_sink());
         s.make_edge(source, queue);
         s.make_edge(queue, sink);
diff --git a/cpp/mrc/tests/test_edges.cpp b/cpp/mrc/tests/test_edges.cpp
index 91c6d4e09..cf68141ce 100644
--- a/cpp/mrc/tests/test_edges.cpp
+++ b/cpp/mrc/tests/test_edges.cpp
@@ -54,7 +54,7 @@
 // IWYU thinks we need vector for make_segment
 // IWYU pragma: no_include <vector>
 
-using namespace std::chrono_literals;
+using namespace std::literals;
 
 TEST_CLASS(Edges);
 
@@ -1002,6 +1002,7 @@ template <typename T>
 class TestEdgeHolder : public edge::EdgeHolder<T>
 {
   public:
+    TestEdgeHolder(std::string name) : edge::EdgeHolder<T>(std::move(name)) {}
     bool has_active_connection() const
     {
         return this->check_active_connection(false);
@@ -1016,11 +1017,26 @@ class TestEdgeHolder : public edge::EdgeHolder<T>
     {
         this->init_owned_edge(std::move(edge));
     }
+
+    void call_init_connected_edge(std::shared_ptr<edge::Edge<T>> edge)
+    {
+        this->init_connected_edge(std::move(edge));
+    }
+};
+
+template <typename T>
+class TestEdge : public edge::Edge<T>
+{
+  public:
+    void call_connect()
+    {
+        this->connect();
+    }
 };
 
 TEST_F(TestEdges, EdgeHolderIsConnected)
 {
-    TestEdgeHolder<int> edge_holder;
+    TestEdgeHolder<int> edge_holder("test_holder"s);
     auto edge = std::make_shared<edge::Edge<int>>();
     EXPECT_FALSE(edge_holder.has_active_connection());
 
@@ -1030,4 +1046,36 @@ TEST_F(TestEdges, EdgeHolderIsConnected)
     edge_holder.call_release_edge_connection();
     EXPECT_FALSE(edge_holder.has_active_connection());
 }
+
+TEST_F(TestEdges, EdgeHolderName)
+{
+    TestEdgeHolder<int> edge_holder("test_holder"s);
+    EXPECT_EQ(edge_holder.name(), "test_holder"s);
+}
+
+TEST_F(TestEdges, EdgeHolderConnectRelase)
+{
+    TestEdgeHolder<int> edge_holder("test_holder"s);
+    auto edge = std::make_shared<TestEdge<int>>();
+    EXPECT_FALSE(edge_holder.has_active_connection());
+
+    edge_holder.call_init_connected_edge(std::make_shared<node::NullWritableEdge<int>>());
+    EXPECT_FALSE(edge_holder.has_active_connection());
+
+    edge_holder.call_init_owned_edge(edge);
+    EXPECT_FALSE(edge_holder.has_active_connection());
+
+    edge->call_connect();
+    EXPECT_TRUE(edge_holder.has_active_connection());
+
+    edge_holder.call_release_edge_connection();
+
+    // EdgeHolder is disconnected, but someone is still holding a reference to the edge
+    EXPECT_TRUE(edge_holder.has_active_connection());
+
+    edge.reset();
+
+    EXPECT_FALSE(edge_holder.has_active_connection());
+}
+
 }  // namespace mrc
diff --git a/cpp/mrc/tests/test_pipeline.cpp b/cpp/mrc/tests/test_pipeline.cpp
index 6d1bc4499..9ded8d65a 100644
--- a/cpp/mrc/tests/test_pipeline.cpp
+++ b/cpp/mrc/tests/test_pipeline.cpp
@@ -16,9 +16,9 @@
  */
 
 #include "mrc/node/rx_sink.hpp"
-#include "mrc/node/rx_sink_base.hpp"
+#include "mrc/node/rx_sink_base.hpp"  // for RxSinkBase
 #include "mrc/node/rx_source.hpp"
-#include "mrc/node/rx_source_base.hpp"
+#include "mrc/node/rx_source_base.hpp"  // for RxSourceBase
 #include "mrc/options/options.hpp"
 #include "mrc/options/topology.hpp"
 #include "mrc/pipeline/executor.hpp"
@@ -37,6 +37,7 @@
 #include <cstdint>
 #include <memory>
 #include <ostream>
+#include <stdexcept>
 #include <utility>
 
 namespace mrc {
@@ -104,8 +105,6 @@ TEST_F(TestPipeline, DuplicateSegments)
 
 TEST_F(TestPipeline, TwoSegment)
 {
-    GTEST_SKIP() << "#185";
-
     std::atomic<int> next_count     = 0;
     std::atomic<int> complete_count = 0;
 
@@ -161,6 +160,131 @@ TEST_F(TestPipeline, TwoSegment)
     LOG(INFO) << "Done" << std::endl;
 }
 
+TEST_F(TestPipeline, SegmentInitErrorHandling)
+{
+    // Test to reproduce issue #360
+    auto pipeline = mrc::make_pipeline();
+
+    auto seg = pipeline->make_segment("seg_1", [](segment::IBuilder& seg) {
+        auto rx_source = seg.make_source<float>("rx_source", [](rxcpp::subscriber<float> s) {
+            FAIL() << "This should not be called";
+        });
+
+        auto rx_sink = seg.make_sink<float>("rx_sink",
+                                            rxcpp::make_observer_dynamic<float>(
+                                                [&](float x) {
+                                                    FAIL() << "This should not be "
+                                                              "called";
+                                                },
+                                                [&]() {
+                                                    FAIL() << "This should not be "
+                                                              "called";
+                                                }));
+
+        seg.make_edge(rx_source, rx_sink);
+
+        throw std::runtime_error("Error in initializer");
+    });
+
+    Executor exec(std::move(m_options));
+
+    exec.register_pipeline(std::move(pipeline));
+
+    exec.start();
+
+    EXPECT_THROW(exec.join(), std::runtime_error);
+}
+
+TEST_F(TestPipeline, SegmentInitErrorHandlingFirstSeg)
+{
+    // Test to reproduce issue #360
+    auto pipeline = mrc::make_pipeline();
+
+    auto seg_1 =
+        pipeline->make_segment("seg_1", segment::EgressPorts<float>({"float_port"}), [](segment::IBuilder& seg) {
+            auto rx_source = seg.make_source<float>("rx_source", [](rxcpp::subscriber<float> s) {
+                FAIL() << "This should not be called";
+            });
+
+            auto my_float_egress = seg.get_egress<float>("float_port");
+
+            seg.make_edge(rx_source, my_float_egress);
+            throw std::runtime_error("Error in initializer");
+        });
+
+    auto seg_2 = pipeline->make_segment("seg_2",
+                                        segment::IngressPorts<float>({"float_port"}),
+                                        [&](segment::IBuilder& seg) {
+                                            auto my_float_ingress = seg.get_ingress<float>("float_port");
+
+                                            auto rx_sink = seg.make_sink<float>("rx_sink",
+                                                                                rxcpp::make_observer_dynamic<float>(
+                                                                                    [&](float x) {
+                                                                                        FAIL() << "This should not be "
+                                                                                                  "called";
+                                                                                    },
+                                                                                    [&]() {
+                                                                                        FAIL() << "This should not be "
+                                                                                                  "called";
+                                                                                    }));
+
+                                            seg.make_edge(my_float_ingress, rx_sink);
+                                        });
+
+    Executor exec(std::move(m_options));
+
+    exec.register_pipeline(std::move(pipeline));
+
+    exec.start();
+
+    EXPECT_THROW(exec.join(), std::runtime_error);
+}
+
+TEST_F(TestPipeline, SegmentInitErrorHandlingSecondSeg)
+{
+    // Test to reproduce issue #360
+    auto pipeline = mrc::make_pipeline();
+
+    auto seg_1 =
+        pipeline->make_segment("seg_1", segment::EgressPorts<float>({"float_port"}), [](segment::IBuilder& seg) {
+            auto rx_source = seg.make_source<float>("rx_source", [](rxcpp::subscriber<float> s) {
+                FAIL() << "This should not be called";
+            });
+
+            auto my_float_egress = seg.get_egress<float>("float_port");
+
+            seg.make_edge(rx_source, my_float_egress);
+        });
+
+    auto seg_2 = pipeline->make_segment("seg_2",
+                                        segment::IngressPorts<float>({"float_port"}),
+                                        [&](segment::IBuilder& seg) {
+                                            auto my_float_ingress = seg.get_ingress<float>("float_port");
+
+                                            auto rx_sink = seg.make_sink<float>("rx_sink",
+                                                                                rxcpp::make_observer_dynamic<float>(
+                                                                                    [&](float x) {
+                                                                                        FAIL() << "This should not be "
+                                                                                                  "called";
+                                                                                    },
+                                                                                    [&]() {
+                                                                                        FAIL() << "This should not be "
+                                                                                                  "called";
+                                                                                    }));
+
+                                            seg.make_edge(my_float_ingress, rx_sink);
+                                            throw std::runtime_error("Error in initializer");
+                                        });
+
+    Executor exec(std::move(m_options));
+
+    exec.register_pipeline(std::move(pipeline));
+
+    exec.start();
+
+    EXPECT_THROW(exec.join(), std::runtime_error);
+}
+
 /*
 TEST_F(TestPipeline, TwoSegmentManualTag)
 {
diff --git a/docs/quickstart/cpp/common/include/nodes.hpp b/docs/quickstart/cpp/common/include/nodes.hpp
index d2e5eaf89..64706146e 100644
--- a/docs/quickstart/cpp/common/include/nodes.hpp
+++ b/docs/quickstart/cpp/common/include/nodes.hpp
@@ -19,12 +19,14 @@
 
 #include <mrc/node/rx_source.hpp>
 
+#include <string>
+
 namespace mrc::quickstart::cpp::common {
 
 class IntSource : public mrc::node::RxSource<int>
 {
   public:
-    IntSource();
+    IntSource(std::string name = std::string());
 };
 
 }  // namespace mrc::quickstart::cpp::common
diff --git a/docs/quickstart/cpp/common/src/nodes.cpp b/docs/quickstart/cpp/common/src/nodes.cpp
index bf19a26dc..cb9a73c10 100644
--- a/docs/quickstart/cpp/common/src/nodes.cpp
+++ b/docs/quickstart/cpp/common/src/nodes.cpp
@@ -21,13 +21,13 @@
 
 namespace mrc::quickstart::cpp::common {
 
-IntSource::IntSource() :
-  mrc::node::RxSource<int>(rxcpp::observable<>::create<int>([](rxcpp::subscriber<int> s) {
-      s.on_next(1);
-      s.on_next(2);
-      s.on_next(3);
-      s.on_completed();
-  }))
+IntSource::IntSource(std::string name) :
+  mrc::node::RxSource<int>(std::move(name), rxcpp::observable<>::create<int>([](rxcpp::subscriber<int> s) {
+                               s.on_next(1);
+                               s.on_next(2);
+                               s.on_next(3);
+                               s.on_completed();
+                           }))
 {}
 
 }  // namespace mrc::quickstart::cpp::common
diff --git a/docs/quickstart/hybrid/mrc_qs_hybrid/common/nodes.cpp b/docs/quickstart/hybrid/mrc_qs_hybrid/common/nodes.cpp
index 10b060ae6..ab5f655a7 100644
--- a/docs/quickstart/hybrid/mrc_qs_hybrid/common/nodes.cpp
+++ b/docs/quickstart/hybrid/mrc_qs_hybrid/common/nodes.cpp
@@ -39,7 +39,7 @@ class DataObjectSource : public mrc::pymrc::PythonSource<std::shared_ptr<common:
     // using typename base_t::subscriber_fn_t;
 
   public:
-    DataObjectSource(size_t count) : PythonSource(build()), m_count(count) {}
+    DataObjectSource(std::string name, size_t count) : PythonSource(std::move(name), build()), m_count(count) {}
 
   private:
     subscriber_fn_t build()
@@ -66,7 +66,9 @@ class DataObjectNode
     using base_t = mrc::pymrc::PythonNode<std::shared_ptr<common::DataObject>, std::shared_ptr<common::DataObject>>;
 
   public:
-    DataObjectNode() : PythonNode(base_t::op_factory_from_sub_fn(build())) {}
+    DataObjectNode(std::string name = std::string()) :
+      PythonNode(std::move(name), base_t::op_factory_from_sub_fn(build()))
+    {}
 
   private:
     subscribe_fn_t build()
@@ -96,7 +98,8 @@ class DataObjectSink : public mrc::pymrc::PythonSink<std::shared_ptr<common::Dat
     // using typename base_t::sink_type_t;
 
   public:
-    DataObjectSink() : PythonSink(build_on_next(), build_on_complete()) {}
+    DataObjectSink(std::string name = std::string()) : PythonSink(std::move(name), build_on_next(), build_on_complete())
+    {}
 
   private:
     on_next_fn_t build_on_next()
diff --git a/docs/quickstart/hybrid/mrc_qs_hybrid/ex01_wrap_nodes/nodes.cpp b/docs/quickstart/hybrid/mrc_qs_hybrid/ex01_wrap_nodes/nodes.cpp
index ea30b6b48..195e924ac 100644
--- a/docs/quickstart/hybrid/mrc_qs_hybrid/ex01_wrap_nodes/nodes.cpp
+++ b/docs/quickstart/hybrid/mrc_qs_hybrid/ex01_wrap_nodes/nodes.cpp
@@ -35,7 +35,7 @@ namespace py = pybind11;
 class MyDataObjectSource : public mrc::pymrc::PythonSource<std::shared_ptr<common::DataObject>>
 {
   public:
-    MyDataObjectSource(size_t count) : PythonSource(build()), m_count(count) {}
+    MyDataObjectSource(std::string name, size_t count) : PythonSource(std::move(name), build()), m_count(count) {}
 
   private:
     subscriber_fn_t build()
@@ -62,7 +62,9 @@ class MyDataObjectNode
     using base_t = mrc::pymrc::PythonNode<std::shared_ptr<common::DataObject>, std::shared_ptr<common::DataObject>>;
 
   public:
-    MyDataObjectNode() : PythonNode(base_t::op_factory_from_sub_fn(build())) {}
+    MyDataObjectNode(std::string name = std::string()) :
+      PythonNode(std::move(name), base_t::op_factory_from_sub_fn(build()))
+    {}
 
   private:
     subscribe_fn_t build()
@@ -89,7 +91,9 @@ class MyDataObjectNode
 class MyDataObjectSink : public mrc::pymrc::PythonSink<std::shared_ptr<common::DataObject>>
 {
   public:
-    MyDataObjectSink() : PythonSink(build_on_next(), build_on_complete()) {}
+    MyDataObjectSink(std::string name = std::string()) :
+      PythonSink(std::move(name), build_on_next(), build_on_complete())
+    {}
 
   private:
     on_next_fn_t build_on_next()
diff --git a/python/mrc/_pymrc/include/pymrc/node.hpp b/python/mrc/_pymrc/include/pymrc/node.hpp
index f5d72e7c3..91346b706 100644
--- a/python/mrc/_pymrc/include/pymrc/node.hpp
+++ b/python/mrc/_pymrc/include/pymrc/node.hpp
@@ -32,6 +32,8 @@
 #include "mrc/node/rx_node.hpp"
 #include "mrc/node/rx_sink.hpp"
 #include "mrc/node/rx_source.hpp"
+#include "mrc/node/sink_properties.hpp"
+#include "mrc/node/source_properties.hpp"
 #include "mrc/runnable/context.hpp"
 
 #include <pybind11/cast.h>
@@ -295,6 +297,8 @@ class PythonSink : public node::RxSink<InputT, ContextT>,
     using typename base_t::observer_t;
 
     using base_t::base_t;
+
+    PythonSink(std::string name = std::string()) : node::SinkProperties<InputT>(name), base_t(name) {}
 };
 
 template <typename InputT>
@@ -325,6 +329,12 @@ class PythonNode : public node::RxNode<InputT, OutputT, ContextT>,
 
     using base_t::base_t;
 
+    PythonNode(std::string name = std::string()) :
+      node::SinkProperties<InputT>(name),
+      node::SourceProperties<OutputT>(name),
+      base_t(name)
+    {}
+
   protected:
     static auto op_factory_from_sub_fn(subscribe_fn_t sub_fn)
     {
@@ -380,11 +390,12 @@ class PythonSource : public node::RxSource<OutputT, ContextT>,
 
     using base_t::base_t;
 
-    PythonSource(const subscriber_fn_t& f) :
-      base_t(rxcpp::observable<>::create<OutputT>([f](rxcpp::subscriber<OutputT>& s) {
-          // Call the wrapped subscriber function
-          f(s);
-      }))
+    PythonSource(std::string name, const subscriber_fn_t& f) :
+      node::SourceProperties<OutputT>(name),
+      base_t(name, rxcpp::observable<>::create<OutputT>([f](rxcpp::subscriber<OutputT>& s) {
+                 // Call the wrapped subscriber function
+                 f(s);
+             }))
     {}
 };
 
@@ -396,7 +407,10 @@ class PythonSourceComponent : public node::LambdaSourceComponent<OutputT>,
     using base_t = node::LambdaSourceComponent<OutputT>;
 
   public:
-    using base_t::base_t;
+    PythonSourceComponent(std::string name, const typename base_t::get_data_fn_t& f) :
+      node::SourceProperties<OutputT>(name),
+      base_t(name, f)
+    {}
 };
 
 class SegmentObjectProxy
diff --git a/python/mrc/tests/test_edges.cpp b/python/mrc/tests/test_edges.cpp
index ccac5a2d7..731b4b0fe 100644
--- a/python/mrc/tests/test_edges.cpp
+++ b/python/mrc/tests/test_edges.cpp
@@ -150,7 +150,7 @@ class TestSource : public pymrc::PythonSource<std::shared_ptr<T>>, public TestSo
     using base_t = pymrc::PythonSource<std::shared_ptr<T>>;
 
     TestSource(std::string name, pymrc::PyHolder counter, size_t msg_count = 5) :
-      base_t(),
+      base_t(name),
       TestSourceImpl<T>(std::move(name), std::move(counter), msg_count)
     {
         this->set_observable(this->build());
@@ -164,7 +164,7 @@ class TestSourceComponent : public pymrc::PythonSourceComponent<std::shared_ptr<
     using base_t = pymrc::PythonSourceComponent<std::shared_ptr<T>>;
 
     TestSourceComponent(std::string name, pymrc::PyHolder counter, size_t msg_count = 5) :
-      base_t(build()),
+      base_t(name, build()),
       TestSourceImpl<T>(std::move(name), std::move(counter), msg_count)
     {}
 
@@ -227,6 +227,7 @@ class TestNode : public pymrc::PythonNode<std::shared_ptr<T>, std::shared_ptr<T>
 
   public:
     TestNode(std::string name, pymrc::PyHolder counter, size_t msg_count = 5) :
+      base_t(name),
       TestNodeImpl<T>(std::move(name), std::move(counter))
     {
         this->make_stream(this->build_operator());
@@ -241,6 +242,7 @@ class TestNodeComponent : public pymrc::PythonNodeComponent<std::shared_ptr<T>,
 
   public:
     TestNodeComponent(std::string name, pymrc::PyHolder counter, size_t msg_count = 5) :
+      base_t(name),
       TestNodeImpl<T>(std::move(name), std::move(counter))
     {
         this->make_stream(this->build_operator());
@@ -278,8 +280,11 @@ class TestSinkImpl : public PythonTestNodeMixin
 template <typename T>
 class TestSink : public pymrc::PythonSink<std::shared_ptr<T>>, public TestSinkImpl<T>
 {
+    using base_t = pymrc::PythonSink<std::shared_ptr<T>>;
+
   public:
     TestSink(std::string name, pymrc::PyHolder counter, size_t msg_count = 5) :
+      base_t(name),
       TestSinkImpl<T>(std::move(name), std::move(counter))
     {
         this->set_observer(this->build());
@@ -289,8 +294,11 @@ class TestSink : public pymrc::PythonSink<std::shared_ptr<T>>, public TestSinkIm
 template <typename T>
 class TestSinkComponent : public pymrc::PythonSinkComponent<std::shared_ptr<T>>, public TestSinkImpl<T>
 {
+    using base_t = pymrc::PythonSinkComponent<std::shared_ptr<T>>;
+
   public:
     TestSinkComponent(std::string name, pymrc::PyHolder counter, size_t msg_count = 5) :
+      base_t(name),
       TestSinkImpl<T>(std::move(name), std::move(counter))
     {
         this->set_observer(this->build());
@@ -310,7 +318,7 @@ GENERATE_NODE_TYPES(TestSinkComponent, SinkComponent);
                std::shared_ptr<segment::Object<class_name>>>(py_mod, #class_name)                                  \
         .def(py::init<>(                                                                                           \
                  [](mrc::segment::IBuilder& parent, const std::string& name, py::dict counter, size_t msg_count) { \
-                     auto stage = parent.construct_object<class_name>(name, name, std::move(counter), msg_count);  \
+                     auto stage = parent.construct_object<class_name>(name, std::move(counter), msg_count);        \
                      return stage;                                                                                 \
                  }),                                                                                               \
              py::arg("parent"),                                                                                    \
diff --git a/python/tests/test_edges.py b/python/tests/test_edges.py
index 98ed11d0e..a9c359518 100644
--- a/python/tests/test_edges.py
+++ b/python/tests/test_edges.py
@@ -14,6 +14,7 @@
 # limitations under the License.
 
 import itertools
+import logging
 import typing
 
 import pytest
@@ -23,9 +24,12 @@
 import mrc.core.operators as ops
 import mrc.tests.test_edges_cpp as m
 
+mrc.logging.init_logging("test_edges")
+mrc.logging.set_level(logging.INFO)
 
-@pytest.fixture
-def ex_runner():
+
+@pytest.fixture(params=[mrc.core.options.EngineType.Thread, mrc.core.options.EngineType.Fiber], ids=["thread", "fiber"])
+def ex_runner(request: pytest.FixtureRequest):
 
     def run_exec(segment_init):
         pipeline = mrc.Pipeline()
@@ -36,6 +40,7 @@ def run_exec(segment_init):
 
         # Set to 1 thread
         options.topology.user_cpuset = "0-0"
+        options.engine_factories.default_engine_type = request.param
 
         executor = mrc.Executor(options)
 
@@ -431,14 +436,15 @@ def fail_if_more_derived_type(combo: typing.Tuple):
 @pytest.mark.parametrize("source_cpp", [True, False], ids=["source_cpp", "source_py"])
 @pytest.mark.parametrize("sink1_cpp", [True, False], ids=["sink1_cpp", "sink2_py"])
 @pytest.mark.parametrize("sink2_cpp", [True, False], ids=["sink2_cpp", "sink2_py"])
-@pytest.mark.parametrize("source_type,sink1_type,sink2_type",
-                         gen_parameters("source",
-                                        "sink1",
-                                        "sink2",
-                                        is_fail_fn=fail_if_more_derived_type,
-                                        values={
-                                            "base": m.Base, "derived": m.DerivedA
-                                        }))
+@pytest.mark.parametrize(
+    "source_type,sink1_type,sink2_type",
+    gen_parameters("source",
+                   "sink1",
+                   "sink2",
+                   is_fail_fn=fail_if_more_derived_type,
+                   values={
+                       "base": m.Base, "derived": m.DerivedA
+                   }))
 def test_source_to_broadcast_to_sinks(run_segment,
                                       sink1_component: bool,
                                       sink2_component: bool,
@@ -504,12 +510,10 @@ def segment_init(seg: mrc.Builder):
 
 
 @pytest.mark.parametrize("source_cpp", [True, False], ids=["source_cpp", "source_py"])
-@pytest.mark.parametrize("source_type",
-                         gen_parameters("source",
-                                        is_fail_fn=lambda _: False,
-                                        values={
-                                            "base": m.Base, "derived": m.DerivedA
-                                        }))
+@pytest.mark.parametrize(
+    "source_type", gen_parameters("source", is_fail_fn=lambda _: False, values={
+        "base": m.Base, "derived": m.DerivedA
+    }))
 def test_source_to_null(run_segment, source_cpp: bool, source_type: type):
 
     def segment_init(seg: mrc.Builder):
@@ -522,24 +526,24 @@ def segment_init(seg: mrc.Builder):
     assert results == expected_node_counts
 
 
-@pytest.mark.parametrize("source_cpp,node_cpp",
-                         gen_parameters("source", "node", is_fail_fn=lambda _: False, values={
-                             "cpp": True, "py": False
-                         }))
-@pytest.mark.parametrize("source_type,node_type",
-                         gen_parameters("source",
-                                        "node",
-                                        is_fail_fn=fail_if_more_derived_type,
-                                        values={
-                                            "base": m.Base, "derived": m.DerivedA
-                                        }))
-@pytest.mark.parametrize("source_component,node_component",
-                         gen_parameters("source",
-                                        "node",
-                                        is_fail_fn=lambda x: x[0] and x[1],
-                                        values={
-                                            "run": False, "com": True
-                                        }))
+@pytest.mark.parametrize(
+    "source_cpp,node_cpp",
+    gen_parameters("source", "node", is_fail_fn=lambda _: False, values={
+        "cpp": True, "py": False
+    }))
+@pytest.mark.parametrize(
+    "source_type,node_type",
+    gen_parameters("source",
+                   "node",
+                   is_fail_fn=fail_if_more_derived_type,
+                   values={
+                       "base": m.Base, "derived": m.DerivedA
+                   }))
+@pytest.mark.parametrize(
+    "source_component,node_component",
+    gen_parameters("source", "node", is_fail_fn=lambda x: x[0] and x[1], values={
+        "run": False, "com": True
+    }))
 def test_source_to_node_to_null(run_segment,
                                 source_cpp: bool,
                                 node_cpp: bool,
diff --git a/python/tests/test_pipeline.py b/python/tests/test_pipeline.py
index 7f70068e6..49b5a891f 100644
--- a/python/tests/test_pipeline.py
+++ b/python/tests/test_pipeline.py
@@ -19,7 +19,8 @@
 # from functools import partial
 
 # import numpy as np
-# import pytest
+
+import pytest
 
 import mrc
 import mrc.tests.test_edges_cpp as m
@@ -445,9 +446,49 @@ def on_complete():
     executor.join()
 
 
+def test_segment_init_error():
+    """
+    Test for issue #360
+    """
+
+    def gen_data():
+        yield 1
+
+    def init1(builder: mrc.Builder):
+        source = builder.make_source("source", gen_data)
+        egress = builder.get_egress("b")
+        builder.make_edge(source, egress)
+        raise RuntimeError("Test for #360")
+
+    def init2(builder: mrc.Builder):
+
+        def on_next(input):
+            pass
+
+        ingress = builder.get_ingress("b")
+        sink = builder.make_sink("sink", on_next)
+
+        builder.make_edge(ingress, sink)
+
+    pipe = mrc.Pipeline()
+
+    pipe.make_segment("TestSegment1", [], [("b", int, False)], init1)
+    pipe.make_segment("TestSegment2", [("b", int, False)], [], init2)
+
+    options = mrc.Options()
+
+    executor = mrc.Executor(options)
+    executor.register_pipeline(pipe)
+
+    with pytest.raises(RuntimeError):
+        executor.start()
+        executor.join()
+
+
 if (__name__ in ("__main__", )):
     test_dynamic_port_creation_good()
     test_dynamic_port_creation_bad()
     test_ingress_egress_custom_type_construction()
     test_dynamic_port_get_ingress_egress()
     test_dynamic_port_with_type_get_ingress_egress()
+    test_segment_init_error()