Merge bitcoin/bitcoin#31397: p2p: track and use all potential peers f…

…or orphan resolution

86d7135 [p2p] only attempt 1p1c when both txns provided by the same peer (glozow)
f7658d9 [cleanup] remove p2p_inv from AddTxAnnouncement (glozow)
063c132 [functional test] getorphantxs reflects multiple announcers (glozow)
0da693f [functional test] orphan handling with multiple announcers (glozow)
b6ea4a9 [p2p] try multiple peers for orphan resolution (glozow)
1d2e1d7 [refactor] move creation of unique_parents to helper function (glozow)
c6893b0 [txdownload] remove unique_parents that we already have (glozow)
163aaf2 [fuzz] orphanage multiple announcer functions (glozow)
22b023b [unit test] multiple orphan announcers (glozow)
96c1a82 [unit test] TxOrphanage EraseForBlock (glozow)
04448ce [txorphanage] add GetTx so that orphan vin can be read (glozow)
e810842 [txorphanage] support multiple announcers (glozow)
62a9ff1 [refactor] change type of unique_parents to Txid (glozow)
6951ddc [txrequest] GetCandidatePeers (glozow)

Pull request description:

  Part of #27463.

  (Transaction) **orphan resolution** is a process that kicks off when we are missing UTXOs to validate an unconfirmed transaction. We currently request missing parents by txid; BIP 331 also defines a way to [explicitly request ancestors](https://github.com/bitcoin/bips/blob/master/bip-0331.mediawiki#handle-orphans-better).

  Currently, when we find that a transaction is an orphan, we only try to resolve it with the peer who provided the `tx`. If this doesn't work out (e.g. they send a `notfound` or don't respond), we do not try again. We actually can't, because we've already forgotten who else could resolve this orphan (i.e. all the other peers who announced the transaction).

  What is wrong with this? It makes transaction download less reliable, particularly for 1p1c packages which must go through orphan resolution in order to be downloaded.

  Can we fix this with BIP 331 / is this "duct tape" before the real solution?
  BIP 331 (receiver-initiated ancestor package relay) is also based on the idea that there is an orphan that needs resolution, but it's just a new way of communicating information. It's not inherently more honest; you can request ancestor package information and get a `notfound`. So ancestor package relay still requires some kind of procedure for retrying when an orphan resolution attempt fails. See the #27742 implementation which builds on this orphan resolution tracker to keep track of what packages to download (it just isn't rebased on this exact branch). The difference when using BIP 331 is that we request `ancpkginfo` and then `pkgtxns` instead of the parent txids.

  Zooming out, we'd like orphan handling to be:
  - Bandwidth-efficient: don't have too many requests out at once. As already implemented today, transaction requests for orphan parents and regular download both go through the `TxRequestTracker` so that we don't have duplicate requests out.
  - Not vulnerable to censorship: don't give up too easily, use all candidate peers. See e.g. https://bitcoincore.org/en/2024/07/03/disclose_already_asked_for/
  - Load-balance between peers: don't overload peers; use all peers available. This is also useful for when we introduce per-peer orphan protection, since each peer will have limited slots.

  The approach taken in this PR is to think of each peer who announces an orphan as a potential "orphan resolution candidate." These candidates include:
  - the peer who sent us the orphan tx
  - any peers who announced the orphan prior to us downloading it
  - any peers who subsequently announce the orphan after we have started trying to resolve it
  For each orphan resolution candidate, we treat them as having "announced" all of the missing parents to us at the time of receipt of this orphan transaction (or at the time they announced the tx if they do so after we've already started tracking it as an orphan). We add the missing parents as entries to `m_txrequest`, incorporating the logic of typical txrequest processing, which means we prefer outbounds, try not to have duplicate requests in flight, don't overload peers, etc.

ACKs for top commit:
  marcofleon:
    Code review ACK 86d7135
  instagibbs:
    reACK 86d7135
  dergoegge:
    Code review ACK 86d7135
  mzumsande:
    ACK 86d7135

Tree-SHA512: 618d523b86e60c3ea039e88326d50db4e55e8e18309c6a20e8f2b10ed9e076f1de0315c335fd3b8abdabcc8b53cbceb66fb59147d05470ea25b83a2b4bd9c877

src/net_processing.cpp

@@ -2978,7 +2978,7 @@ std::optional<node::PackageToValidate> PeerManagerImpl::ProcessInvalidTx(NodeId
     if (add_extra_compact_tx && RecursiveDynamicUsage(*ptx) < 100000) {
         AddToCompactExtraTransactions(ptx);
     }
-    for (const uint256& parent_txid : unique_parents) {
+    for (const Txid& parent_txid : unique_parents) {
         if (peer) AddKnownTx(*peer, parent_txid);
     }
 
@@ -3934,7 +3934,7 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
                 AddKnownTx(*peer, inv.hash);
 
                 if (!m_chainman.IsInitialBlockDownload()) {
-                    const bool fAlreadyHave{m_txdownloadman.AddTxAnnouncement(pfrom.GetId(), gtxid, current_time, /*p2p_inv=*/true)};
+                    const bool fAlreadyHave{m_txdownloadman.AddTxAnnouncement(pfrom.GetId(), gtxid, current_time)};
                     LogDebug(BCLog::NET, "got inv: %s  %s peer=%d\n", inv.ToString(), fAlreadyHave ? "have" : "new", pfrom.GetId());
                 }
             } else {

src/node/txdownloadman.h

@@ -92,7 +92,7 @@ struct PackageToValidate {
 struct RejectedTxTodo
 {
     bool m_should_add_extra_compact_tx;
-    std::vector<uint256> m_unique_parents;
+    std::vector<Txid> m_unique_parents;
     std::optional<PackageToValidate> m_package_to_validate;
 };
 
@@ -136,9 +136,8 @@ class TxDownloadManager {
 
     /** Consider adding this tx hash to txrequest. Should be called whenever a new inv has been received.
      * Also called internally when a transaction is missing parents so that we can request them.
-     * @param[in] p2p_inv     When true, only add this announcement if we don't already have the tx.
      * Returns true if this was a dropped inv (p2p_inv=true and we already have the tx), false otherwise. */
-    bool AddTxAnnouncement(NodeId peer, const GenTxid& gtxid, std::chrono::microseconds now, bool p2p_inv);
+    bool AddTxAnnouncement(NodeId peer, const GenTxid& gtxid, std::chrono::microseconds now);
 
     /** Get getdata requests to send. */
     std::vector<GenTxid> GetRequestsToSend(NodeId nodeid, std::chrono::microseconds current_time);

src/node/txdownloadman_impl.cpp

@@ -39,9 +39,9 @@ void TxDownloadManager::DisconnectedPeer(NodeId nodeid)
 {
     m_impl->DisconnectedPeer(nodeid);
 }
-bool TxDownloadManager::AddTxAnnouncement(NodeId peer, const GenTxid& gtxid, std::chrono::microseconds now, bool p2p_inv)
+bool TxDownloadManager::AddTxAnnouncement(NodeId peer, const GenTxid& gtxid, std::chrono::microseconds now)
 {
-    return m_impl->AddTxAnnouncement(peer, gtxid, now, p2p_inv);
+    return m_impl->AddTxAnnouncement(peer, gtxid, now);
 }
 std::vector<GenTxid> TxDownloadManager::GetRequestsToSend(NodeId nodeid, std::chrono::microseconds current_time)
 {
@@ -172,12 +172,39 @@ void TxDownloadManagerImpl::DisconnectedPeer(NodeId nodeid)
 
 }
 
-bool TxDownloadManagerImpl::AddTxAnnouncement(NodeId peer, const GenTxid& gtxid, std::chrono::microseconds now, bool p2p_inv)
+bool TxDownloadManagerImpl::AddTxAnnouncement(NodeId peer, const GenTxid& gtxid, std::chrono::microseconds now)
 {
+    // If this is an orphan we are trying to resolve, consider this peer as a orphan resolution candidate instead.
+    // - is wtxid matching something in orphanage
+    // - exists in orphanage
+    // - peer can be an orphan resolution candidate
+    if (gtxid.IsWtxid()) {
+        if (auto orphan_tx{m_orphanage.GetTx(Wtxid::FromUint256(gtxid.GetHash()))}) {
+            auto unique_parents{GetUniqueParents(*orphan_tx)};
+            std::erase_if(unique_parents, [&](const auto& txid){
+                return AlreadyHaveTx(GenTxid::Txid(txid), /*include_reconsiderable=*/false);
+            });
+
+            if (unique_parents.empty()) return true;
+
+            if (auto delay{OrphanResolutionCandidate(peer, Wtxid::FromUint256(gtxid.GetHash()), unique_parents.size())}) {
+                m_orphanage.AddAnnouncer(Wtxid::FromUint256(gtxid.GetHash()), peer);
+
+                const auto& info = m_peer_info.at(peer).m_connection_info;
+                for (const auto& parent_txid : unique_parents) {
+                    m_txrequest.ReceivedInv(peer, GenTxid::Txid(parent_txid), info.m_preferred, now + *delay);
+                }
+
+                LogDebug(BCLog::TXPACKAGES, "added peer=%d as a candidate for resolving orphan %s\n", peer, gtxid.GetHash().ToString());
+            }
+
+            // Return even if the peer isn't an orphan resolution candidate. This would be caught by AlreadyHaveTx.
+            return true;
+        }
+    }
+
     // If this is an inv received from a peer and we already have it, we can drop it.
-    // If this is a request for the parent of an orphan, we don't drop transactions that we already have. In particular,
-    // we *do* want to request parents that are in m_lazy_recent_rejects_reconsiderable, since they can be CPFP'd.
-    if (p2p_inv && AlreadyHaveTx(gtxid, /*include_reconsiderable=*/true)) return true;
+    if (AlreadyHaveTx(gtxid, /*include_reconsiderable=*/true)) return true;
 
     auto it = m_peer_info.find(peer);
     if (it == m_peer_info.end()) return false;
@@ -204,6 +231,36 @@ bool TxDownloadManagerImpl::AddTxAnnouncement(NodeId peer, const GenTxid& gtxid,
     return false;
 }
 
+std::optional<std::chrono::seconds> TxDownloadManagerImpl::OrphanResolutionCandidate(NodeId nodeid, const Wtxid& orphan_wtxid, size_t num_parents)
+{
+    if (m_peer_info.count(nodeid) == 0) return std::nullopt;
+    if (m_orphanage.HaveTxFromPeer(orphan_wtxid, nodeid)) return std::nullopt;
+
+    const auto& peer_entry = m_peer_info.at(nodeid);
+    const auto& info = peer_entry.m_connection_info;
+    // TODO: add delays and limits based on the amount of orphan resolution we are already doing
+    // with this peer, how much they are using the orphanage, etc.
+    if (!info.m_relay_permissions) {
+        // This mirrors the delaying and dropping behavior in AddTxAnnouncement in order to preserve
+        // existing behavior: drop if we are tracking too many invs for this peer already. Each
+        // orphan resolution involves at least 1 transaction request which may or may not be
+        // currently tracked in m_txrequest, so we include that in the count.
+        if (m_txrequest.Count(nodeid) + num_parents > MAX_PEER_TX_ANNOUNCEMENTS) return std::nullopt;
+    }
+
+    std::chrono::seconds delay{0s};
+    if (!info.m_preferred) delay += NONPREF_PEER_TX_DELAY;
+    // The orphan wtxid is used, but resolution entails requesting the parents by txid. Sometimes
+    // parent and child are announced and thus requested around the same time, and we happen to
+    // receive child sooner. Waiting a few seconds may allow us to cancel the orphan resolution
+    // request if the parent arrives in that time.
+    if (m_num_wtxid_peers > 0) delay += TXID_RELAY_DELAY;
+    const bool overloaded = !info.m_relay_permissions && m_txrequest.CountInFlight(nodeid) >= MAX_PEER_TX_REQUEST_IN_FLIGHT;
+    if (overloaded) delay += OVERLOADED_PEER_TX_DELAY;
+
+    return delay;
+}
+
 std::vector<GenTxid> TxDownloadManagerImpl::GetRequestsToSend(NodeId nodeid, std::chrono::microseconds current_time)
 {
     std::vector<GenTxid> requests;
@@ -243,9 +300,11 @@ std::optional<PackageToValidate> TxDownloadManagerImpl::Find1P1CPackage(const CT
 
     Assume(RecentRejectsReconsiderableFilter().contains(parent_wtxid.ToUint256()));
 
-    // Prefer children from this peer. This helps prevent censorship attempts in which an attacker
+    // Only consider children from this peer. This helps prevent censorship attempts in which an attacker
     // sends lots of fake children for the parent, and we (unluckily) keep selecting the fake
-    // children instead of the real one provided by the honest peer.
+    // children instead of the real one provided by the honest peer. Since we track all announcers
+    // of an orphan, this does not exclude parent + orphan pairs that we happened to request from
+    // different peers.
     const auto cpfp_candidates_same_peer{m_orphanage.GetChildrenFromSamePeer(ptx, nodeid)};
 
     // These children should be sorted from newest to oldest. In the (probably uncommon) case
@@ -258,34 +317,6 @@ std::optional<PackageToValidate> TxDownloadManagerImpl::Find1P1CPackage(const CT
             return PackageToValidate{ptx, child, nodeid, nodeid};
         }
     }
-
-    // If no suitable candidate from the same peer is found, also try children that were provided by
-    // a different peer. This is useful because sometimes multiple peers announce both transactions
-    // to us, and we happen to download them from different peers (we wouldn't have known that these
-    // 2 transactions are related). We still want to find 1p1c packages then.
-    //
-    // If we start tracking all announcers of orphans, we can restrict this logic to parent + child
-    // pairs in which both were provided by the same peer, i.e. delete this step.
-    const auto cpfp_candidates_different_peer{m_orphanage.GetChildrenFromDifferentPeer(ptx, nodeid)};
-
-    // Find the first 1p1c that hasn't already been rejected. We randomize the order to not
-    // create a bias that attackers can use to delay package acceptance.
-    //
-    // Create a random permutation of the indices.
-    std::vector<size_t> tx_indices(cpfp_candidates_different_peer.size());
-    std::iota(tx_indices.begin(), tx_indices.end(), 0);
-    std::shuffle(tx_indices.begin(), tx_indices.end(), m_opts.m_rng);
-
-    for (const auto index : tx_indices) {
-        // If we already tried a package and failed for any reason, the combined hash was
-        // cached in m_lazy_recent_rejects_reconsiderable.
-        const auto [child_tx, child_sender] = cpfp_candidates_different_peer.at(index);
-        Package maybe_cpfp_package{ptx, child_tx};
-        if (!RecentRejectsReconsiderableFilter().contains(GetPackageHash(maybe_cpfp_package)) &&
-            !RecentRejectsFilter().contains(child_tx->GetHash().ToUint256())) {
-            return PackageToValidate{ptx, child_tx, nodeid, child_sender};
-        }
-    }
     return std::nullopt;
 }
 
@@ -301,14 +332,29 @@ void TxDownloadManagerImpl::MempoolAcceptedTx(const CTransactionRef& tx)
     m_orphanage.EraseTx(tx->GetWitnessHash());
 }
 
+std::vector<Txid> TxDownloadManagerImpl::GetUniqueParents(const CTransaction& tx)
+{
+    std::vector<Txid> unique_parents;
+    unique_parents.reserve(tx.vin.size());
+    for (const CTxIn& txin : tx.vin) {
+        // We start with all parents, and then remove duplicates below.
+        unique_parents.push_back(txin.prevout.hash);
+    }
+
+    std::sort(unique_parents.begin(), unique_parents.end());
+    unique_parents.erase(std::unique(unique_parents.begin(), unique_parents.end()), unique_parents.end());
+
+    return unique_parents;
+}
+
 node::RejectedTxTodo TxDownloadManagerImpl::MempoolRejectedTx(const CTransactionRef& ptx, const TxValidationState& state, NodeId nodeid, bool first_time_failure)
 {
     const CTransaction& tx{*ptx};
     // Results returned to caller
     // Whether we should call AddToCompactExtraTransactions at the end
     bool add_extra_compact_tx{first_time_failure};
     // Hashes to pass to AddKnownTx later
-    std::vector<uint256> unique_parents;
+    std::vector<Txid> unique_parents;
     // Populated if failure is reconsiderable and eligible package is found.
     std::optional<node::PackageToValidate> package_to_validate;
 
@@ -320,13 +366,7 @@ node::RejectedTxTodo TxDownloadManagerImpl::MempoolRejectedTx(const CTransaction
 
             // Deduplicate parent txids, so that we don't have to loop over
             // the same parent txid more than once down below.
-            unique_parents.reserve(tx.vin.size());
-            for (const CTxIn& txin : tx.vin) {
-                // We start with all parents, and then remove duplicates below.
-                unique_parents.push_back(txin.prevout.hash);
-            }
-            std::sort(unique_parents.begin(), unique_parents.end());
-            unique_parents.erase(std::unique(unique_parents.begin(), unique_parents.end()), unique_parents.end());
+            unique_parents = GetUniqueParents(tx);
 
             // Distinguish between parents in m_lazy_recent_rejects and m_lazy_recent_rejects_reconsiderable.
             // We can tolerate having up to 1 parent in m_lazy_recent_rejects_reconsiderable since we
@@ -348,30 +388,48 @@ node::RejectedTxTodo TxDownloadManagerImpl::MempoolRejectedTx(const CTransaction
                 }
             }
             if (!fRejectedParents) {
-                const auto current_time{GetTime<std::chrono::microseconds>()};
-
-                for (const uint256& parent_txid : unique_parents) {
-                    // Here, we only have the txid (and not wtxid) of the
-                    // inputs, so we only request in txid mode, even for
-                    // wtxidrelay peers.
-                    // Eventually we should replace this with an improved
-                    // protocol for getting all unconfirmed parents.
-                    const auto gtxid{GenTxid::Txid(parent_txid)};
-                    // Exclude m_lazy_recent_rejects_reconsiderable: the missing parent may have been
-                    // previously rejected for being too low feerate. This orphan might CPFP it.
-                    if (!AlreadyHaveTx(gtxid, /*include_reconsiderable=*/false)) {
-                        AddTxAnnouncement(nodeid, gtxid, current_time, /*p2p_inv=*/false);
+                // Filter parents that we already have.
+                // Exclude m_lazy_recent_rejects_reconsiderable: the missing parent may have been
+                // previously rejected for being too low feerate. This orphan might CPFP it.
+                std::erase_if(unique_parents, [&](const auto& txid){
+                    return AlreadyHaveTx(GenTxid::Txid(txid), /*include_reconsiderable=*/false);
+                });
+                const auto now{GetTime<std::chrono::microseconds>()};
+                const auto& wtxid = ptx->GetWitnessHash();
+                // Potentially flip add_extra_compact_tx to false if tx is already in orphanage, which
+                // means it was already added to vExtraTxnForCompact.
+                add_extra_compact_tx &= !m_orphanage.HaveTx(wtxid);
+
+                auto add_orphan_reso_candidate = [&](const CTransactionRef& orphan_tx, const std::vector<Txid>& unique_parents, NodeId nodeid, std::chrono::microseconds now) {
+                    const auto& wtxid = orphan_tx->GetWitnessHash();
+                    if (auto delay{OrphanResolutionCandidate(nodeid, wtxid, unique_parents.size())}) {
+                        const auto& info = m_peer_info.at(nodeid).m_connection_info;
+                        m_orphanage.AddTx(orphan_tx, nodeid);
+
+                        // Treat finding orphan resolution candidate as equivalent to the peer announcing all missing parents
+                        // In the future, orphan resolution may include more explicit steps
+                        for (const auto& parent_txid : unique_parents) {
+                            m_txrequest.ReceivedInv(nodeid, GenTxid::Txid(parent_txid), info.m_preferred, now + *delay);
+                        }
+                        LogDebug(BCLog::TXPACKAGES, "added peer=%d as a candidate for resolving orphan %s\n", nodeid, wtxid.ToString());
                     }
+                };
+
+                // If there is no candidate for orphan resolution, AddTx will not be called. This means
+                // that if a peer is overloading us with invs and orphans, they will eventually not be
+                // able to add any more transactions to the orphanage.
+                add_orphan_reso_candidate(ptx, unique_parents, nodeid, now);
+                for (const auto& candidate : m_txrequest.GetCandidatePeers(ptx)) {
+                    add_orphan_reso_candidate(ptx, unique_parents, candidate, now);
                 }
 
-                // Potentially flip add_extra_compact_tx to false if AddTx returns false because the tx was already there
-                add_extra_compact_tx &= m_orphanage.AddTx(ptx, nodeid);
-
                 // Once added to the orphan pool, a tx is considered AlreadyHave, and we shouldn't request it anymore.
                 m_txrequest.ForgetTxHash(tx.GetHash());
                 m_txrequest.ForgetTxHash(tx.GetWitnessHash());
 
                 // DoS prevention: do not allow m_orphanage to grow unbounded (see CVE-2012-3789)
+                // Note that, if the orphanage reaches capacity, it's possible that we immediately evict
+                // the transaction we just added.
                 m_orphanage.LimitOrphans(m_opts.m_max_orphan_txs, m_opts.m_rng);
             } else {
                 unique_parents.clear();

src/node/txdownloadman_impl.h

@@ -163,7 +163,7 @@ class TxDownloadManagerImpl {
     /** Consider adding this tx hash to txrequest. Should be called whenever a new inv has been received.
      * Also called internally when a transaction is missing parents so that we can request them.
      */
-    bool AddTxAnnouncement(NodeId peer, const GenTxid& gtxid, std::chrono::microseconds now, bool p2p_inv);
+    bool AddTxAnnouncement(NodeId peer, const GenTxid& gtxid, std::chrono::microseconds now);
 
     /** Get getdata requests to send. */
     std::vector<GenTxid> GetRequestsToSend(NodeId nodeid, std::chrono::microseconds current_time);
@@ -189,6 +189,16 @@ class TxDownloadManagerImpl {
     void CheckIsEmpty(NodeId nodeid);
 
     std::vector<TxOrphanage::OrphanTxBase> GetOrphanTransactions() const;
+
+protected:
+    /** Helper for getting deduplicated vector of Txids in vin. */
+    std::vector<Txid> GetUniqueParents(const CTransaction& tx);
+
+    /** Determine candidacy (and delay) for potential orphan resolution candidate.
+     * @returns delay for orphan resolution if this peer is a good candidate for orphan resolution,
+     * std::nullopt if this peer cannot be added because it has reached download/orphanage limits.
+     * */
+    std::optional<std::chrono::seconds> OrphanResolutionCandidate(NodeId nodeid, const Wtxid& orphan_wtxid, size_t num_parents);
 };
 } // namespace node
 #endif // BITCOIN_NODE_TXDOWNLOADMAN_IMPL_H

src/rpc/mempool.cpp

@@ -845,7 +845,9 @@ static UniValue OrphanToJSON(const TxOrphanage::OrphanTxBase& orphan)
     o.pushKV("entry", int64_t{TicksSinceEpoch<std::chrono::seconds>(orphan.nTimeExpire - ORPHAN_TX_EXPIRE_TIME)});
     o.pushKV("expiration", int64_t{TicksSinceEpoch<std::chrono::seconds>(orphan.nTimeExpire)});
     UniValue from(UniValue::VARR);
-    from.push_back(orphan.fromPeer); // only one fromPeer for now
+    for (const auto fromPeer: orphan.announcers) {
+        from.push_back(fromPeer);
+    }
     o.pushKV("from", from);
     return o;
 }

src/test/fuzz/txdownloadman.cpp

@@ -230,7 +230,7 @@ FUZZ_TARGET(txdownloadman, .init = initialize)
                 GenTxid gtxid = fuzzed_data_provider.ConsumeBool() ?
                                 GenTxid::Txid(rand_tx->GetHash()) :
                                 GenTxid::Wtxid(rand_tx->GetWitnessHash());
-                txdownloadman.AddTxAnnouncement(rand_peer, gtxid, time, /*p2p_inv=*/fuzzed_data_provider.ConsumeBool());
+                txdownloadman.AddTxAnnouncement(rand_peer, gtxid, time);
             },
             [&] {
                 txdownloadman.GetRequestsToSend(rand_peer, time);
@@ -375,7 +375,7 @@ FUZZ_TARGET(txdownloadman_impl, .init = initialize)
                 GenTxid gtxid = fuzzed_data_provider.ConsumeBool() ?
                                 GenTxid::Txid(rand_tx->GetHash()) :
                                 GenTxid::Wtxid(rand_tx->GetWitnessHash());
-                txdownload_impl.AddTxAnnouncement(rand_peer, gtxid, time, /*p2p_inv=*/fuzzed_data_provider.ConsumeBool());
+                txdownload_impl.AddTxAnnouncement(rand_peer, gtxid, time);
             },
             [&] {
                 const auto getdata_requests = txdownload_impl.GetRequestsToSend(rand_peer, time);