Adding Dandelion++ support to public networks:

  - New flag in NOTIFY_NEW_TRANSACTION to indicate stem mode
  - Stem loops detected in tx_pool.cpp
  - Embargo timeout for a blackhole attack during stem phase

7 files changed, 216 insertions, 69 deletions

diff --git a/src/cryptonote_protocol/cryptonote_protocol_defs.h b/src/cryptonote_protocol/cryptonote_protocol_defs.h
index f809bff74..76b57afd3 100644
--- a/src/cryptonote_protocol/cryptonote_protocol_defs.h
+++ b/src/cryptonote_protocol/cryptonote_protocol_defs.h
@@ -197,10 +197,12 @@ namespace cryptonote
     {
       std::vector<blobdata>   txs;
       std::string _; // padding
+      bool dandelionpp_fluff; //zero initialization defaults to stem mode
 
       BEGIN_KV_SERIALIZE_MAP()
         KV_SERIALIZE(txs)
         KV_SERIALIZE(_)
+        KV_SERIALIZE_OPT(dandelionpp_fluff, true) // backwards compatible mode is fluff
       END_KV_SERIALIZE_MAP()
     };
     typedef epee::misc_utils::struct_init<request_t> request;
diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.h b/src/cryptonote_protocol/cryptonote_protocol_handler.h
index 2664716a8..e6479b941 100644
--- a/src/cryptonote_protocol/cryptonote_protocol_handler.h
+++ b/src/cryptonote_protocol/cryptonote_protocol_handler.h
@@ -129,7 +129,7 @@ namespace cryptonote
 		
     //----------------- i_bc_protocol_layout ---------------------------------------
     virtual bool relay_block(NOTIFY_NEW_BLOCK::request& arg, cryptonote_connection_context& exclude_context);
-    virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone);
+    virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone, relay_method tx_relay);
     //----------------------------------------------------------------------------------
     //bool get_payload_sync_data(HANDSHAKE_DATA::request& hshd, cryptonote_connection_context& context);
     bool should_drop_connection(cryptonote_connection_context& context, uint32_t next_stripe);
diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.inl b/src/cryptonote_protocol/cryptonote_protocol_handler.inl
index 3aacce421..ca06fe51c 100644
--- a/src/cryptonote_protocol/cryptonote_protocol_handler.inl
+++ b/src/cryptonote_protocol/cryptonote_protocol_handler.inl
@@ -926,29 +926,60 @@ namespace cryptonote
       return 1;
     }
 
-    std::vector<cryptonote::blobdata> newtxs;
-    newtxs.reserve(arg.txs.size());
-    for (size_t i = 0; i < arg.txs.size(); ++i)
+    relay_method tx_relay;
+    std::vector<blobdata> stem_txs{};
+    std::vector<blobdata> fluff_txs{};
+    if (arg.dandelionpp_fluff)
     {
-      cryptonote::tx_verification_context tvc{};
-      m_core.handle_incoming_tx({arg.txs[i], crypto::null_hash}, tvc, relay_method::fluff, true);
-      if(tvc.m_verifivation_failed)
+      tx_relay = relay_method::fluff;
+      fluff_txs.reserve(arg.txs.size());
+    }
+    else
+    {
+      tx_relay = relay_method::stem;
+      stem_txs.reserve(arg.txs.size());
+    }
+
+    for (auto& tx : arg.txs)
+    {
+      tx_verification_context tvc{};
+      if (!m_core.handle_incoming_tx({tx, crypto::null_hash}, tvc, tx_relay, true))
       {
         LOG_PRINT_CCONTEXT_L1("Tx verification failed, dropping connection");
         drop_connection(context, false, false);
         return 1;
       }
-      if(tvc.m_should_be_relayed)
-        newtxs.push_back(std::move(arg.txs[i]));
+
+      switch (tvc.m_relay)
+      {
+        case relay_method::local:
+        case relay_method::stem:
+          stem_txs.push_back(std::move(tx));
+          break;
+        case relay_method::block:
+        case relay_method::fluff:
+          fluff_txs.push_back(std::move(tx));
+          break;
+        default:
+        case relay_method::none:
+          break;
+      }
     }
-    arg.txs = std::move(newtxs);
 
-    if(arg.txs.size())
+    if (!stem_txs.empty())
     {
       //TODO: add announce usage here
-      relay_transactions(arg, context.m_connection_id, context.m_remote_address.get_zone());
+      arg.dandelionpp_fluff = false;
+      arg.txs = std::move(stem_txs);
+      relay_transactions(arg, context.m_connection_id, context.m_remote_address.get_zone(), relay_method::stem);
+    }
+    if (!fluff_txs.empty())
+    {
+      //TODO: add announce usage here
+      arg.dandelionpp_fluff = true;
+      arg.txs = std::move(fluff_txs);
+      relay_transactions(arg, context.m_connection_id, context.m_remote_address.get_zone(), relay_method::fluff);
     }
-
     return 1;
   }
   //------------------------------------------------------------------------------------------------------------------------
@@ -2387,14 +2418,14 @@ skip:
   }
   //------------------------------------------------------------------------------------------------------------------------
   template<class t_core>
-  bool t_cryptonote_protocol_handler<t_core>::relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone)
+  bool t_cryptonote_protocol_handler<t_core>::relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone, relay_method tx_relay)
   {
     /* Push all outgoing transactions to this function. The behavior needs to
        identify how the transaction is going to be relayed, and then update the
        local mempool before doing the relay. The code was already updating the
        DB twice on received transactions - it is difficult to workaround this
        due to the internal design. */
-    return m_p2p->send_txs(std::move(arg.txs), zone, source, m_core) != epee::net_utils::zone::invalid;
+    return m_p2p->send_txs(std::move(arg.txs), zone, source, m_core, tx_relay) != epee::net_utils::zone::invalid;
   }
   //------------------------------------------------------------------------------------------------------------------------
   template<class t_core>
diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler_common.h b/src/cryptonote_protocol/cryptonote_protocol_handler_common.h
index 978a9ebf3..11184299d 100644
--- a/src/cryptonote_protocol/cryptonote_protocol_handler_common.h
+++ b/src/cryptonote_protocol/cryptonote_protocol_handler_common.h
@@ -41,7 +41,7 @@ namespace cryptonote
   struct i_cryptonote_protocol
   {
     virtual bool relay_block(NOTIFY_NEW_BLOCK::request& arg, cryptonote_connection_context& exclude_context)=0;
-    virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone)=0;
+    virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone, relay_method tx_relay)=0;
     //virtual bool request_objects(NOTIFY_REQUEST_GET_OBJECTS::request& arg, cryptonote_connection_context& context)=0;
   };
 
@@ -54,7 +54,7 @@ namespace cryptonote
     {
       return false;
     }
-    virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone)
+    virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone, relay_method tx_relay)
     {
       return false;
     }
diff --git a/src/cryptonote_protocol/enums.h b/src/cryptonote_protocol/enums.h
index 2ec622d94..a3a52b83f 100644
--- a/src/cryptonote_protocol/enums.h
+++ b/src/cryptonote_protocol/enums.h
@@ -37,7 +37,8 @@ namespace cryptonote
   {
     none = 0, //!< Received via RPC with `do_not_relay` set
     local,    //!< Received via RPC; trying to send over i2p/tor, etc.
-    block,    //!< Received in block, takes precedence over others
-    fluff     //!< Received/sent over public networks
+    stem,     //!< Received/send over network using Dandelion++ stem
+    fluff,    //!< Received/sent over network using Dandelion++ fluff
+    block     //!< Received in block, takes precedence over others
   };
 }
diff --git a/src/cryptonote_protocol/levin_notify.cpp b/src/cryptonote_protocol/levin_notify.cpp
index 428b739bc..127801092 100644
--- a/src/cryptonote_protocol/levin_notify.cpp
+++ b/src/cryptonote_protocol/levin_notify.cpp
@@ -30,6 +30,7 @@
 
 #include <boost/asio/steady_timer.hpp>
 #include <boost/system/system_error.hpp>
+#include <boost/uuid/uuid_io.hpp>
 #include <chrono>
 #include <deque>
 #include <stdexcept>
@@ -38,8 +39,10 @@
 #include "common/expect.h"
 #include "common/varint.h"
 #include "cryptonote_config.h"
-#include "crypto/random.h"
+#include "crypto/crypto.h"
+#include "crypto/duration.h"
 #include "cryptonote_basic/connection_context.h"
+#include "cryptonote_core/i_core_events.h"
 #include "cryptonote_protocol/cryptonote_protocol_defs.h"
 #include "net/dandelionpp.h"
 #include "p2p/net_node.h"
@@ -61,11 +64,14 @@ namespace levin
 {
   namespace
   {
-    constexpr std::size_t connection_id_reserve_size = 100;
+    constexpr const std::size_t connection_id_reserve_size = 100;
 
     constexpr const std::chrono::minutes noise_min_epoch{CRYPTONOTE_NOISE_MIN_EPOCH};
     constexpr const std::chrono::seconds noise_epoch_range{CRYPTONOTE_NOISE_EPOCH_RANGE};
 
+    constexpr const std::chrono::minutes dandelionpp_min_epoch{CRYPTONOTE_DANDELIONPP_MIN_EPOCH};
+    constexpr const std::chrono::seconds dandelionpp_epoch_range{CRYPTONOTE_DANDELIONPP_EPOCH_RANGE};
+
     constexpr const std::chrono::seconds noise_min_delay{CRYPTONOTE_NOISE_MIN_DELAY};
     constexpr const std::chrono::seconds noise_delay_range{CRYPTONOTE_NOISE_DELAY_RANGE};
 
@@ -83,22 +89,8 @@ namespace levin
         connections (Dandelion++ makes similar assumptions in its stem
         algorithm). The randomization yields 95% values between 1s-4s in
 	1/4s increments. */
-    constexpr const fluff_stepsize fluff_average_out{fluff_stepsize{fluff_average_in} / 2};
-
-    class random_poisson
-    {
-      std::poisson_distribution<fluff_stepsize::rep> dist;
-    public:
-      explicit random_poisson(fluff_stepsize average)
-        : dist(average.count() < 0 ? 0 : average.count())
-      {}
-
-      fluff_stepsize operator()()
-      {
-        crypto::random_device rand{};
-        return fluff_stepsize{dist(rand)};
-      }
-    };
+    using fluff_duration = crypto::random_poisson_subseconds::result_type;
+    constexpr const fluff_duration fluff_average_out{fluff_duration{fluff_average_in} / 2};
 
     /*! Select a randomized duration from 0 to `range`. The precision will be to
         the systems `steady_clock`. As an example, supplying 3 seconds to this
@@ -132,10 +124,11 @@ namespace levin
       return outs;
     }
 
-    std::string make_tx_payload(std::vector<blobdata>&& txs, const bool pad)
+    std::string make_tx_payload(std::vector<blobdata>&& txs, const bool pad, const bool fluff)
     {
       NOTIFY_NEW_TRANSACTIONS::request request{};
       request.txs = std::move(txs);
+      request.dandelionpp_fluff = fluff;
 
       if (pad)
       {
@@ -172,9 +165,9 @@ namespace levin
       return fullBlob;
     }
 
-    bool make_payload_send_txs(connections& p2p, std::vector<blobdata>&& txs, const boost::uuids::uuid& destination, const bool pad)
+    bool make_payload_send_txs(connections& p2p, std::vector<blobdata>&& txs, const boost::uuids::uuid& destination, const bool pad, const bool fluff)
     {
-      const cryptonote::blobdata blob = make_tx_payload(std::move(txs), pad);
+      const cryptonote::blobdata blob = make_tx_payload(std::move(txs), pad, fluff);
       p2p.for_connection(destination, [&blob](detail::p2p_context& context) {
         on_levin_traffic(context, true, true, false, blob.size(), get_command_from_message(blob));
         return true;
@@ -251,7 +244,8 @@ namespace levin
           flush_time(std::chrono::steady_clock::time_point::max()),
           connection_count(0),
           is_public(is_public),
-          pad_txs(pad_txs)
+          pad_txs(pad_txs),
+          fluffing(false)
       {
         for (std::size_t count = 0; !noise.empty() && count < CRYPTONOTE_NOISE_CHANNELS; ++count)
           channels.emplace_back(io_service);
@@ -268,6 +262,7 @@ namespace levin
       std::atomic<std::size_t> connection_count; //!< Only update in strand, can be read at any time
       const bool is_public;                      //!< Zone is public ipv4/ipv6 connections
       const bool pad_txs;                        //!< Pad txs to the next boundary for privacy
+      bool fluffing;                             //!< Zone is in Dandelion++ fluff epoch
     };
   } // detail
 
@@ -362,10 +357,11 @@ namespace levin
           return true;
         });
 
+        // Always send txs in stem mode over i2p/tor, see comments in `send_txs` below.
         for (auto& connection : connections)
         {
           std::sort(connection.first.begin(), connection.first.end()); // don't leak receive order
-          make_payload_send_txs(*zone_->p2p, std::move(connection.first), connection.second, zone_->pad_txs);
+          make_payload_send_txs(*zone_->p2p, std::move(connection.first), connection.second, zone_->pad_txs, zone_->is_public);
         }
 
         if (next_flush != std::chrono::steady_clock::time_point::max())
@@ -387,29 +383,38 @@ namespace levin
 
       void operator()()
       {
-        if (!zone_ || !zone_->p2p || txs_.empty())
+        run(std::move(zone_), epee::to_span(txs_), source_);
+      }
+
+      static void run(std::shared_ptr<detail::zone> zone, epee::span<const blobdata> txs, const boost::uuids::uuid& source)
+      {
+        if (!zone || !zone->p2p || txs.empty())
           return;
 
-        assert(zone_->strand.running_in_this_thread());
+        assert(zone->strand.running_in_this_thread());
 
         const auto now = std::chrono::steady_clock::now();
         auto next_flush = std::chrono::steady_clock::time_point::max();
 
-        random_poisson in_duration(fluff_average_in);
-        random_poisson out_duration(fluff_average_out);
+        crypto::random_poisson_subseconds in_duration(fluff_average_in);
+        crypto::random_poisson_subseconds out_duration(fluff_average_out);
+
+
+        MDEBUG("Queueing " << txs.size() << " transaction(s) for Dandelion++ fluffing");
 
         bool available = false;
-        zone_->p2p->foreach_connection([this, now, &in_duration, &out_duration, &next_flush, &available] (detail::p2p_context& context)
+        zone->p2p->foreach_connection([txs, now, &zone, &source, &in_duration, &out_duration, &next_flush, &available] (detail::p2p_context& context)
         {
-          if (this->source_ != context.m_connection_id && (this->zone_->is_public || !context.m_is_income))
+          // When i2p/tor, only fluff to outbound connections
+          if (source != context.m_connection_id && (zone->is_public || !context.m_is_income))
           {
             available = true;
             if (context.fluff_txs.empty())
               context.flush_time = now + (context.m_is_income ? in_duration() : out_duration());
 
             next_flush = std::min(next_flush, context.flush_time);
-            context.fluff_txs.reserve(context.fluff_txs.size() + this->txs_.size());
-            for (const blobdata& tx : this->txs_)
+            context.fluff_txs.reserve(context.fluff_txs.size() + txs.size());
+            for (const blobdata& tx : txs)
               context.fluff_txs.push_back(tx); // must copy instead of move (multiple conns)
           }
           return true;
@@ -418,8 +423,8 @@ namespace levin
         if (!available)
           MWARNING("Unable to send transaction(s), no available connections");
 
-        if (next_flush < zone_->flush_time)
-          fluff_flush::queue(std::move(zone_), next_flush);
+        if (next_flush < zone->flush_time)
+          fluff_flush::queue(std::move(zone), next_flush);
       }
     };
 
@@ -471,6 +476,11 @@ namespace levin
         assert(zone->strand.running_in_this_thread());
 
         zone->connection_count = zone->map.size();
+
+        // only noise uses the "noise channels", only update when enabled
+        if (zone->noise.empty())
+          return;
+
         for (auto id = zone->map.begin(); id != zone->map.end(); ++id)
         {
           const std::size_t i = id - zone->map.begin();
@@ -479,26 +489,75 @@ namespace levin
       }
 
       //! \pre Called within `zone_->strand`.
+      static void run(std::shared_ptr<detail::zone> zone, std::vector<boost::uuids::uuid> out_connections)
+      {
+        if (!zone)
+          return;
+
+        assert(zone->strand.running_in_this_thread());
+        if (zone->map.update(std::move(out_connections)))
+          post(std::move(zone));
+      }
+
+      //! \pre Called within `zone_->strand`.
       void operator()()
       {
-        if (!zone_)
+        run(std::move(zone_), std::move(out_connections_));
+      }
+    };
+
+    //! Checks fluff status for this node, and then does stem or fluff for txes
+    struct dandelionpp_notify
+    {
+      std::shared_ptr<detail::zone> zone_;
+      i_core_events* core_;
+      std::vector<blobdata> txs_;
+      boost::uuids::uuid source_;
+
+      //! \pre Called in `zone_->strand`
+      void operator()()
+      {
+        if (!zone_ || !core_ || txs_.empty())
           return;
 
-        assert(zone_->strand.running_in_this_thread());
-        if (zone_->map.update(std::move(out_connections_)))
-          post(std::move(zone_));
+        if (zone_->fluffing)
+        {
+          core_->on_transactions_relayed(epee::to_span(txs_), relay_method::fluff);
+          fluff_notify::run(std::move(zone_), epee::to_span(txs_), source_);
+        }
+        else // forward tx in stem
+        {
+          core_->on_transactions_relayed(epee::to_span(txs_), relay_method::stem);
+          for (int tries = 2; 0 < tries; tries--)
+          {
+            const boost::uuids::uuid destination = zone_->map.get_stem(source_);
+            if (!destination.is_nil() && make_payload_send_txs(*zone_->p2p, std::vector<blobdata>{txs_}, destination, zone_->pad_txs, false))
+            {
+              /* Source is intentionally omitted in debug log for privacy - a
+                 nil uuid indicates source is that node. */
+              MDEBUG("Sent " << txs_.size() << " transaction(s) to " << destination << " using Dandelion++ stem");
+              return;
+            }
+
+            // connection list may be outdated, try again
+            update_channels::run(zone_, get_out_connections(*zone_->p2p));
+          }
+
+          MERROR("Unable to send transaction(s) via Dandelion++ stem");
+        }
       }
     };
 
-    //! Swaps out noise channels entirely; new epoch start.
+    //! Swaps out noise/dandelionpp channels entirely; new epoch start.
     class change_channels
     {
       std::shared_ptr<detail::zone> zone_;
       net::dandelionpp::connection_map map_; // Requires manual copy constructor
+      bool fluffing_;
 
     public:
-      explicit change_channels(std::shared_ptr<detail::zone> zone, net::dandelionpp::connection_map map)
-        : zone_(std::move(zone)), map_(std::move(map))
+      explicit change_channels(std::shared_ptr<detail::zone> zone, net::dandelionpp::connection_map map, const bool fluffing)
+        : zone_(std::move(zone)), map_(std::move(map)), fluffing_(fluffing)
       {}
 
       change_channels(change_channels&&) = default;
@@ -510,11 +569,15 @@ namespace levin
       void operator()()
       {
         if (!zone_)
-          return
+          return;
 
         assert(zone_->strand.running_in_this_thread());
 
+        if (zone_->is_public)
+          MDEBUG("Starting new Dandelion++ epoch: " << (fluffing_ ? "fluff" : "stem"));
+
         zone_->map = std::move(map_);
+        zone_->fluffing = fluffing_;
         update_channels::post(std::move(zone_));
       }
     };
@@ -608,9 +671,10 @@ namespace levin
         if (error && error != boost::system::errc::operation_canceled)
           throw boost::system::system_error{error, "start_epoch timer failed"};
 
+        const bool fluffing = crypto::rand_idx(unsigned(100)) < CRYPTONOTE_DANDELIONPP_FLUFF_PROBABILITY;
         const auto start = std::chrono::steady_clock::now();
         zone_->strand.dispatch(
-          change_channels{zone_, net::dandelionpp::connection_map{get_out_connections(*(zone_->p2p)), count_}}
+          change_channels{zone_, net::dandelionpp::connection_map{get_out_connections(*(zone_->p2p)), count_}, fluffing}
         );
 
         detail::zone& alias = *zone_;
@@ -626,10 +690,16 @@ namespace levin
     if (!zone_->p2p)
       throw std::logic_error{"cryptonote::levin::notify cannot have nullptr p2p argument"};
 
-    if (!zone_->noise.empty())
+    const bool noise_enabled = !zone_->noise.empty();
+    if (noise_enabled || is_public)
     {
       const auto now = std::chrono::steady_clock::now();
-      start_epoch{zone_, noise_min_epoch, noise_epoch_range, CRYPTONOTE_NOISE_CHANNELS}();
+      const auto min_epoch = noise_enabled ? noise_min_epoch : dandelionpp_min_epoch;
+      const auto epoch_range = noise_enabled ? noise_epoch_range : dandelionpp_epoch_range;
+      const std::size_t out_count = noise_enabled ? CRYPTONOTE_NOISE_CHANNELS : CRYPTONOTE_DANDELIONPP_STEMS;
+
+      start_epoch{zone_, min_epoch, epoch_range, out_count}();
+
       for (std::size_t channel = 0; channel < zone_->channels.size(); ++channel)
         send_noise::wait(now, zone_, channel);
     }
@@ -679,7 +749,7 @@ namespace levin
     zone_->flush_txs.cancel();
   }
 
-  bool notify::send_txs(std::vector<blobdata> txs, const boost::uuids::uuid& source)
+  bool notify::send_txs(std::vector<blobdata> txs, const boost::uuids::uuid& source, i_core_events& core, relay_method tx_relay)
   {
     if (txs.empty())
       return true;
@@ -687,6 +757,17 @@ namespace levin
     if (!zone_)
       return false;
 
+    /* If noise is enabled in a zone, it always takes precedence. The technique
+       provides good protection against ISP adversaries, but not sybil
+       adversaries. Noise is currently only enabled over I2P/Tor - those
+       networks provide protection against sybil attacks (we only send to
+       outgoing connections).
+
+       If noise is disabled, Dandelion++ is used for public networks only.
+       Dandelion++ over I2P/Tor should be an interesting case to investigate,
+       but the mempool/stempool needs to know the zone a tx originated from to
+       work properly. */
+
     if (!zone_->noise.empty() && !zone_->channels.empty())
     {
       // covert send in "noise" channel
@@ -694,8 +775,17 @@ namespace levin
         CRYPTONOTE_MAX_FRAGMENTS * CRYPTONOTE_NOISE_BYTES <= LEVIN_DEFAULT_MAX_PACKET_SIZE, "most nodes will reject this fragment setting"
       );
 
-      // padding is not useful when using noise mode
-      const std::string payload = make_tx_payload(std::move(txs), false);
+      if (tx_relay == relay_method::stem)
+      {
+        MWARNING("Dandelion++ stem not supported over noise networks");
+        tx_relay = relay_method::local; // do not put into stempool embargo (hopefully not there already!).
+      }
+
+      core.on_transactions_relayed(epee::to_span(txs), tx_relay);
+
+      // Padding is not useful when using noise mode. Send as stem so receiver
+      // forwards in Dandelion++ mode.
+      const std::string payload = make_tx_payload(std::move(txs), false, false);
       epee::byte_slice message = epee::levin::make_fragmented_notify(
         zone_->noise, NOTIFY_NEW_TRANSACTIONS::ID, epee::strspan<std::uint8_t>(payload)
       );
@@ -714,9 +804,31 @@ namespace levin
     }
     else
     {
-      zone_->strand.dispatch(fluff_notify{zone_, std::move(txs), source});
+      switch (tx_relay)
+      {
+        default:
+        case relay_method::none:
+        case relay_method::block:
+          return false;
+        case relay_method::stem:
+	  tx_relay = relay_method::fluff; // don't set stempool embargo when skipping to fluff
+	  /* fallthrough */
+        case relay_method::local:
+          if (zone_->is_public)
+          {
+	    // this will change a local tx to stem or fluff ...
+            zone_->strand.dispatch(
+              dandelionpp_notify{zone_, std::addressof(core), std::move(txs), source}
+            );
+            break;
+          }
+          /* fallthrough */
+        case relay_method::fluff:
+          core.on_transactions_relayed(epee::to_span(txs), tx_relay);
+          zone_->strand.dispatch(fluff_notify{zone_, std::move(txs), source});
+          break;
+      }
     }
-
     return true;
   }
 } // levin
diff --git a/src/cryptonote_protocol/levin_notify.h b/src/cryptonote_protocol/levin_notify.h
index ce652d933..641f1f956 100644
--- a/src/cryptonote_protocol/levin_notify.h
+++ b/src/cryptonote_protocol/levin_notify.h
@@ -35,6 +35,7 @@
 
 #include "byte_slice.h"
 #include "cryptonote_basic/blobdatatype.h"
+#include "cryptonote_protocol/enums.h"
 #include "cryptonote_protocol/fwd.h"
 #include "net/enums.h"
 #include "span.h"
@@ -122,7 +123,7 @@ namespace levin
           particular stem.
 
       \return True iff the notification is queued for sending. */
-    bool send_txs(std::vector<blobdata> txs, const boost::uuids::uuid& source);
+    bool send_txs(std::vector<blobdata> txs, const boost::uuids::uuid& source, i_core_events& core, relay_method tx_relay);
   };
 } // levin
 } // net