diff --git a/p2p/src/peer_manager/tests/addr_list_response_caching.rs b/p2p/src/peer_manager/tests/addr_list_response_caching.rs
index a92eb6dd26..fd1d698854 100644
--- a/p2p/src/peer_manager/tests/addr_list_response_caching.rs
+++ b/p2p/src/peer_manager/tests/addr_list_response_caching.rs
@@ -247,6 +247,7 @@ fn make_p2p_config() -> P2pConfig {
             msg_max_locator_count: Default::default(),
             max_message_size: Default::default(),
             max_peer_tx_announcements: Default::default(),
+            ..Default::default()
         },
 
         bind_addresses: Default::default(),
diff --git a/p2p/src/protocol.rs b/p2p/src/protocol.rs
index 34a167f055..d7d06e1d9d 100644
--- a/p2p/src/protocol.rs
+++ b/p2p/src/protocol.rs
@@ -88,6 +88,12 @@ make_config_setting!(MaxMessageSize, usize, 10 * 1024 * 1024);
 make_config_setting!(MaxPeerTxAnnouncements, usize, 5000);
 make_config_setting!(MaxUnconnectedHeaders, usize, 10);
 make_config_setting!(MaxAddrListResponseAddressCount, usize, 1000);
+make_config_setting!(ForkDownloadLimit, usize, 2000);
+make_config_setting!(
+    ForkDownloadRefillInterval,
+    std::time::Duration,
+    std::time::Duration::from_secs(600)
+);
 
 /// Protocol configuration. These values are supposed to be modified in tests only.
 ///
@@ -115,4 +121,10 @@ pub struct ProtocolConfig {
     pub max_message_size: MaxMessageSize,
     /// The maximum number of announcements (hashes) for which we haven't received transactions.
     pub max_peer_tx_announcements: MaxPeerTxAnnouncements,
+    /// The maximum number of blocks that a peer can make us download for the header lists
+    /// it announces, before further announced header lists are deferred until the budget
+    /// is refilled.
+    pub max_fork_downloads_per_peer: ForkDownloadLimit,
+    /// How often the per-peer fork download budget is refilled.
+    pub fork_download_refill_interval: ForkDownloadRefillInterval,
 }
diff --git a/p2p/src/sync/peer/block_manager.rs b/p2p/src/sync/peer/block_manager.rs
index e20d5107af..5031f22a64 100644
--- a/p2p/src/sync/peer/block_manager.rs
+++ b/p2p/src/sync/peer/block_manager.rs
@@ -16,6 +16,7 @@
 use std::{
     collections::{BTreeSet, VecDeque},
     mem,
+    time::Duration,
 };
 
 use itertools::Itertools;
@@ -63,6 +64,8 @@ use crate::{
     utils::oneshot_nofail,
 };
 
+use super::fork_download_budget::ForkDownloadBudget;
+
 #[derive(Debug, Clone)]
 pub enum PeerBlockSyncManagerLocalEvent {
     /// Chainstate got new tip.
@@ -94,8 +97,26 @@ pub struct PeerBlockSyncManager<T: NetworkingService> {
     /// of headers less than the maximum. This is the signal to the peer that we have no more
     /// headers, so it may not ask us for more of them in the future.
     have_sent_all_headers: bool,
+    /// A per-peer budget limiting the total amount of fork (non-tip) block downloads that
+    /// this peer can cause us to perform. See `ForkDownloadBudget` for the rationale.
+    fork_download_budget: ForkDownloadBudget,
+    /// The number of consecutive header lists that contained no new headers while claiming
+    /// (by their size) that the peer may have more of them. Used to stop serving peers that
+    /// keep sending already-known header lists, which would otherwise make us issue header
+    /// requests forever, at no cost to them.
+    consecutive_known_full_header_lists: u32,
+    /// If set, a header list has been deferred due to the fork download budget and we
+    /// should ask the peer for its headers again once the given time is reached. Without
+    /// this, a deferred list would never be fetched unless the peer re-announces, which
+    /// cannot be relied upon (e.g. a quiet upstream).
+    fork_budget_retry_at: Option<Time>,
 }
 
+/// How often the block sync manager checks whether a fork download budget retry is due.
+/// Only the due-ness is checked against the (possibly mocked) clock; the interval itself
+/// is fixed so that the check cannot be postponed by other main loop activity.
+const FORK_BUDGET_RETRY_POLL_INTERVAL: Duration = Duration::from_secs(1);
+
 struct IncomingDataState {
     /// A list of headers received via the `HeaderList` message that we haven't yet
     /// requested the blocks for.
@@ -136,6 +157,11 @@ where
         local_event_receiver: UnboundedReceiver<PeerBlockSyncManagerLocalEvent>,
         time_getter: TimeGetter,
     ) -> Self {
+        let fork_download_budget = ForkDownloadBudget::new(
+            *p2p_config.protocol_config.max_fork_downloads_per_peer,
+            *p2p_config.protocol_config.fork_download_refill_interval,
+            time_getter.get_time(),
+        );
         Self {
             id: id.into(),
             chain_config,
@@ -159,6 +185,9 @@ where
             },
             peer_activity: PeerActivity::new(),
             have_sent_all_headers: false,
+            fork_download_budget,
+            consecutive_known_full_header_lists: 0,
+            fork_budget_retry_at: None,
         }
     }
 
@@ -209,6 +238,27 @@ where
 
                 _ = tokio::time::sleep(stalling_timeout),
                     if self.peer_activity.earliest_expected_activity_time().is_some() => {}
+
+                // An unconditional tick: it guarantees the loop wakes up at least once per
+                // interval even when there is no other activity, so that the fork budget
+                // retry deadline check below cannot be postponed indefinitely by idle
+                // waiting. When the loop is busy, the deadline check runs on every
+                // iteration anyway.
+                _ = tokio::time::sleep(FORK_BUDGET_RETRY_POLL_INTERVAL) => {}
+            }
+
+            // A header list deferred due to the fork download budget is retried once the
+            // (possibly mocked) clock passes the scheduled retry time. The check lives in
+            // the loop body rather than in a select! branch, so that a busy message flow
+            // cannot starve it.
+            let is_fork_budget_retry_due = match self.fork_budget_retry_at {
+                Some(retry_at) => self.time_getter.get_time() >= retry_at,
+                None => false,
+            };
+            if is_fork_budget_retry_due && self.common_services.has_service(Service::Blocks) {
+                self.fork_budget_retry_at = None;
+                log::debug!("Retrying a header request after a fork download budget deferral");
+                self.request_headers().await?;
             }
 
             self.handle_sync_status_change(&last_sync_status)?;
@@ -663,6 +713,53 @@ where
             .expect("Headers shouldn't be empty")
             .prev_block_id();
 
+        // Downloading and validating the blocks behind announced headers is expensive.
+        // To prevent a peer from making us perform an unbounded amount of such work over
+        // time (e.g. by repeatedly announcing distinct header chains that don't extend our
+        // tip), every header list costs tokens from the peer's download budget, and a list
+        // is only processed while the budget can pay for it. Everything sent during the
+        // initial block download is exempt, so that the budget cannot stall node bootstrap.
+        // Note that this must apply to tip-anchored lists as well: a malicious miner can
+        // produce an unlimited number of distinct valid children of our tip, so exempting
+        // them would leave the aforementioned attack open. Honest tip announcements are
+        // tiny (a few headers per new block) compared to the budget capacity, so they are
+        // unaffected in practice.
+        if !self.chainstate_handle.call(|c| Ok(c.is_initial_block_download())).await? {
+            let headers_count = headers.len();
+            if !self.fork_download_budget.try_take(headers_count, self.time_getter.get_time()) {
+                log::info!(
+                    "Deferring the processing of a header list of {headers_count} headers:\
+                     the peer's fork download budget is exhausted"
+                );
+                // No error is returned and no ban score is issued: the peer hasn't violated
+                // the protocol, it has merely exhausted a resource quota. Also note that
+                // `expecting_headers_since` has already been reset at the beginning of this
+                // function, so this deferral cannot be mistaken for peer stalling.
+                //
+                // Schedule a header request retry for when the budget is likely to have
+                // recovered. This is what makes the deferral safe for liveness: without it,
+                // the deferred data would only be fetched if the peer re-announced it. The
+                // retry is a regular header request, so it's rate-limited by the refill
+                // interval and is subject to the budget itself.
+                //
+                // Schedule the retry no later than any retry that was already scheduled: a
+                // fresh deferral must not postpone a retry the budget may already have
+                // recovered for. This also keeps the schedule sane when the clock doesn't
+                // advance (as with a mocked time getter in tests).
+                let retry_at_candidate = self.time_getter.get_time().saturating_duration_add(
+                    *self.p2p_config.protocol_config.fork_download_refill_interval,
+                );
+                self.fork_budget_retry_at = Some(match self.fork_budget_retry_at {
+                    Some(existing) => existing.min(retry_at_candidate),
+                    None => retry_at_candidate,
+                });
+                // Note: the retry is deliberately left armed here. This charge may be for a
+                // list different from the one the pending retry was scheduled for, and a
+                // deferred list can only be recovered through that retry.
+                return Ok(());
+            }
+        }
+
         // Note: we require a peer to send headers starting from a block that we already have
         // in our chainstate. I.e. we don't allow:
         // 1) Basing new headers on a previously sent header, because this would give a malicious
@@ -741,11 +838,32 @@ where
 
         if new_block_headers.is_empty() {
             if peer_may_have_more_headers {
-                self.request_headers().await?;
+                // A peer that keeps sending us already-known header lists while claiming that
+                // it may have more of them would otherwise make us issue header requests
+                // forever, at no cost to itself. After a few such lists in a row, stop
+                // requesting until the peer sends us something actually new. This is not a
+                // protocol violation, so the peer isn't punished; any new headers reset the
+                // counter.
+                const MAX_CONSECUTIVE_KNOWN_FULL_HEADER_LISTS: u32 = 3;
+                self.consecutive_known_full_header_lists =
+                    self.consecutive_known_full_header_lists.saturating_add(1);
+                if self.consecutive_known_full_header_lists
+                    <= MAX_CONSECUTIVE_KNOWN_FULL_HEADER_LISTS
+                {
+                    self.request_headers().await?;
+                } else {
+                    log::info!(
+                        "Not requesting more headers from the peer:\
+                         it keeps sending already-known header lists"
+                    );
+                }
+            } else {
+                self.consecutive_known_full_header_lists = 0;
             }
             return Ok(());
         }
 
+        self.consecutive_known_full_header_lists = 0;
         self.request_blocks(new_block_headers)
     }
 
diff --git a/p2p/src/sync/peer/fork_download_budget.rs b/p2p/src/sync/peer/fork_download_budget.rs
new file mode 100644
index 0000000000..1d497ac30f
--- /dev/null
+++ b/p2p/src/sync/peer/fork_download_budget.rs
@@ -0,0 +1,191 @@
+// Copyright (c) 2023 RBB S.r.l
+// opensource@mintlayer.org
+// SPDX-License-Identifier: MIT
+// Licensed under the MIT License;
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! A per-peer budget limiting the amount of block downloads that announced headers can
+//! trigger.
+//!
+//! Downloading and validating block bodies is expensive. The peer synchronization manager
+//! already limits how many blocks can be requested at once, but that does not bound the
+//! total amount of work a peer can cause over time: a peer that keeps announcing header
+//! lists can, in principle, make us download and process an unlimited number of blocks,
+//! one bounded request at a time.
+//!
+//! This budget bounds that cumulative work: every header list a peer sends us consumes
+//! tokens from the peer's bucket; when the bucket is empty, further header lists are
+//! deferred (with no error and no ban score) until tokens are refilled. The budget is not
+//! spent during the initial block download, so that node bootstrap cannot be stalled.
+//! Honest announcement traffic is tiny compared to the budget capacity (a few headers per
+//! newly produced block), so regular block propagation is unaffected.
+
+use std::time::Duration;
+
+use common::primitives::time::Time;
+
+#[derive(Debug, Clone)]
+pub struct ForkDownloadBudget {
+    /// Total number of tokens (block downloads) the bucket can hold.
+    capacity: u64,
+    /// How often the full capacity is refilled.
+    refill_interval: Duration,
+    /// Currently available tokens.
+    tokens: u64,
+    /// Time at which the last refill (partial or full) happened.
+    last_refill: Time,
+}
+
+impl ForkDownloadBudget {
+    pub fn new(capacity: usize, refill_interval: Duration, now: Time) -> Self {
+        let capacity = capacity as u64;
+        Self {
+            capacity,
+            refill_interval,
+            tokens: capacity,
+            last_refill: now,
+        }
+    }
+
+    fn refill(&mut self, now: Time) {
+        if self.tokens >= self.capacity {
+            self.last_refill = now;
+            return;
+        }
+
+        let interval_secs = self.refill_interval.as_secs();
+        if interval_secs == 0 {
+            // Degenerate configuration: refill instantly.
+            self.tokens = self.capacity;
+            self.last_refill = now;
+            return;
+        }
+
+        let elapsed = now.saturating_sub(self.last_refill);
+        // Tokens gained since the last refill. Use u128 to avoid overflow for very long
+        // elapsed times.
+        let gained = (elapsed.as_secs() as u128)
+            .checked_mul(self.capacity as u128)
+            .and_then(|v| v.checked_div(interval_secs as u128))
+            .unwrap_or(u128::MAX);
+
+        if gained == 0 {
+            return;
+        }
+
+        // Advance the refill time by the amount of time that corresponds to the granted
+        // tokens, so that remainders are not lost across calls.
+        let granted = gained.min(u64::MAX as u128) as u64;
+        self.tokens = self.tokens.saturating_add(granted).min(self.capacity);
+        let consumed_secs = (granted as u128)
+            .checked_mul(interval_secs as u128)
+            .and_then(|v| v.checked_div(self.capacity as u128))
+            .unwrap_or(0);
+        self.last_refill = self
+            .last_refill
+            .saturating_duration_add(Duration::from_secs(consumed_secs as u64));
+    }
+
+    /// Try to consume `amount` tokens at time `now`. Returns `false` if the budget is
+    /// exhausted, in which case nothing is consumed.
+    pub fn try_take(&mut self, amount: usize, now: Time) -> bool {
+        self.refill(now);
+
+        if self.tokens >= amount as u64 {
+            self.tokens -= amount as u64;
+            true
+        } else {
+            false
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use common::primitives::time::Time;
+
+    use super::*;
+
+    fn t(secs: u64) -> Time {
+        Time::from_secs_since_epoch(secs)
+    }
+
+    #[test]
+    fn starts_full() {
+        let mut budget = ForkDownloadBudget::new(10, Duration::from_secs(100), t(0));
+        assert!(budget.try_take(10, t(0)));
+        assert!(!budget.try_take(1, t(0)));
+    }
+
+    #[test]
+    fn exhausts_and_refills_over_time() {
+        // Rate: 10 tokens per 100 seconds.
+        let mut budget = ForkDownloadBudget::new(10, Duration::from_secs(100), t(0));
+        assert!(budget.try_take(10, t(0)));
+        // No time passed - nothing to refill.
+        assert!(!budget.try_take(1, t(0)));
+        // Half of the interval passed - half of the capacity refilled.
+        assert!(budget.try_take(5, t(50)));
+        assert!(!budget.try_take(1, t(50)));
+        // Another half of the interval passed - another half of the capacity refilled.
+        assert!(budget.try_take(5, t(100)));
+        assert!(!budget.try_take(6, t(100)));
+        // At t(150) the bucket holds another 5 tokens, i.e. it has been refilled at a
+        // constant rate all along.
+        assert!(budget.try_take(5, t(150)));
+        assert!(!budget.try_take(1, t(150)));
+    }
+
+    #[test]
+    fn remainders_are_not_lost() {
+        // 4 tokens per 8 seconds => 0.5 tokens per second. Frequent small calls must
+        // accumulate fractional refills instead of discarding them.
+        let mut budget = ForkDownloadBudget::new(4, Duration::from_secs(8), t(0));
+        assert!(budget.try_take(4, t(0)));
+        // After 1 second only 0.5 tokens have accumulated - not enough.
+        assert!(!budget.try_take(1, t(1)));
+        // After 2 seconds a whole token has accumulated.
+        assert!(budget.try_take(1, t(2)));
+        assert!(!budget.try_take(1, t(3)));
+        // After 4 seconds, another token has accumulated.
+        assert!(budget.try_take(1, t(4)));
+        assert!(!budget.try_take(1, t(5)));
+        // And so on at a constant 0.5 tokens per second.
+        assert!(budget.try_take(1, t(6)));
+        assert!(!budget.try_take(1, t(7)));
+    }
+
+    #[test]
+    fn overdraw_is_rejected_atomically() {
+        let mut budget = ForkDownloadBudget::new(10, Duration::from_secs(100), t(0));
+        assert!(!budget.try_take(11, t(0)));
+        // The rejected request must not consume anything.
+        assert!(budget.try_take(10, t(0)));
+    }
+
+    #[test]
+    fn zero_interval_refills_instantly() {
+        let mut budget = ForkDownloadBudget::new(10, Duration::from_secs(0), t(0));
+        assert!(budget.try_take(10, t(0)));
+        assert!(budget.try_take(10, t(0)));
+    }
+
+    #[test]
+    fn time_going_backwards_is_safe() {
+        let mut budget = ForkDownloadBudget::new(10, Duration::from_secs(100), t(1000));
+        assert!(budget.try_take(10, t(1000)));
+        // Refill time in the future: saturating arithmetic must not panic nor over-refill.
+        assert!(!budget.try_take(1, t(0)));
+    }
+}
diff --git a/p2p/src/sync/peer/mod.rs b/p2p/src/sync/peer/mod.rs
index 9f2882dcb9..fade50b232 100644
--- a/p2p/src/sync/peer/mod.rs
+++ b/p2p/src/sync/peer/mod.rs
@@ -14,5 +14,6 @@
 // limitations under the License.
 
 pub mod block_manager;
+pub mod fork_download_budget;
 pub mod requested_transactions;
 pub mod transaction_manager;
diff --git a/p2p/src/sync/tests/block_announcement.rs b/p2p/src/sync/tests/block_announcement.rs
index 638ebfce67..bfc9efc7d4 100644
--- a/p2p/src/sync/tests/block_announcement.rs
+++ b/p2p/src/sync/tests/block_announcement.rs
@@ -532,6 +532,7 @@ async fn send_headers_connected_to_previously_sent_headers(#[case] seed: Seed) {
                 msg_max_locator_count: Default::default(),
                 max_message_size: Default::default(),
                 max_peer_tx_announcements: Default::default(),
+                ..Default::default()
             },
 
             bind_addresses: Default::default(),
@@ -635,6 +636,7 @@ async fn send_headers_connected_to_block_which_is_being_downloaded(#[case] seed:
                 msg_max_locator_count: Default::default(),
                 max_message_size: Default::default(),
                 max_peer_tx_announcements: Default::default(),
+                ..Default::default()
             },
 
             bind_addresses: Default::default(),
@@ -735,6 +737,7 @@ async fn correct_pending_headers_update(#[case] seed: Seed) {
                 msg_max_locator_count: Default::default(),
                 max_message_size: Default::default(),
                 max_peer_tx_announcements: Default::default(),
+                ..Default::default()
             },
 
             bind_addresses: Default::default(),
diff --git a/p2p/src/sync/tests/fork_download_budget.rs b/p2p/src/sync/tests/fork_download_budget.rs
new file mode 100644
index 0000000000..e10f4ad0d2
--- /dev/null
+++ b/p2p/src/sync/tests/fork_download_budget.rs
@@ -0,0 +1,365 @@
+// Copyright (c) 2022 RBB S.r.l
+// opensource@mintlayer.org
+// SPDX-License-Identifier: MIT
+// Licensed under the MIT License;
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Tests for the per-peer fork download budget (see `sync::peer::fork_download_budget`).
+//!
+//! The budget limits the total number of headers a peer can make us process per refill
+//! interval. These tests verify that:
+//! * exceeding the budget defers the processing of header lists without any punishment of
+//!   the peer (no ban/discouragement events),
+//! * peers cannot bypass the budget with announcements that are anchored at our tip,
+//! * deferred data is fetched once the budget is refilled (liveness is preserved),
+//! * the budget is not enforced during the initial block download.
+
+use std::{sync::Arc, time::Duration};
+
+use chainstate::ChainstateHandle;
+use common::{
+    chain::{
+        GenBlock,
+        block::{Block, signed_block_header::SignedBlockHeader},
+        config::create_unit_test_config,
+    },
+    primitives::{Id, Idable},
+};
+use test_utils::{
+    BasicTestTimeGetter,
+    random::{Seed, make_seedable_rng},
+};
+
+use crate::{
+    config::P2pConfig,
+    message::{BlockSyncMessage, HeaderList},
+    protocol::{ForkDownloadLimit, ForkDownloadRefillInterval, ProtocolConfig},
+    sync::tests::helpers::{
+        TestNode, make_new_block, make_new_top_blocks_return_headers,
+        test_node_group::{BlockSyncMessageWithNodeIdx, TestNodeGroup},
+    },
+    test_helpers::{for_each_protocol_version, test_p2p_config_with_protocol_config},
+};
+
+// The retry timer for deferred header lists runs on the real tokio clock and sleeps for
+// this interval, while the budget refill itself is driven by the mocked time getter.
+// Keeping the interval short bounds the real-time wait each exhaustion test incurs and
+// limits how far a firing timer can lag behind the mocked clock under load.
+const REFILL_INTERVAL: Duration = Duration::from_secs(2);
+
+fn make_p2p_config(max_fork_downloads_per_peer: usize) -> P2pConfig {
+    test_p2p_config_with_protocol_config(ProtocolConfig {
+        max_fork_downloads_per_peer: ForkDownloadLimit::new(max_fork_downloads_per_peer),
+        fork_download_refill_interval: ForkDownloadRefillInterval::new(REFILL_INTERVAL),
+        ..Default::default()
+    })
+}
+
+async fn best_block_id(chainstate: &ChainstateHandle) -> Id<GenBlock> {
+    chainstate
+        .call(|c| -> Result<_, chainstate::ChainstateError> { c.get_best_block_id() })
+        .await
+        .unwrap()
+        .unwrap()
+}
+
+async fn has_block(chainstate: &ChainstateHandle, block_id: Id<Block>) -> bool {
+    chainstate
+        .call(move |c| -> Result<_, chainstate::ChainstateError> {
+            c.get_block(&block_id).map(|b| b.is_some())
+        })
+        .await
+        .unwrap()
+        .unwrap()
+}
+
+fn known_list_msg(headers: &[SignedBlockHeader]) -> BlockSyncMessageWithNodeIdx {
+    BlockSyncMessageWithNodeIdx {
+        message: BlockSyncMessage::HeaderList(HeaderList::new(headers.to_vec())),
+        sender_node_idx: 1,
+        receiver_node_idx: 0,
+    }
+}
+
+/// Extend the chain with `count` new blocks and return only the newly created blocks.
+fn make_chain(
+    chain_config: &Arc<common::chain::ChainConfig>,
+    prev_blocks: &[Block],
+    count: usize,
+    time_getter: &BasicTestTimeGetter,
+    rng: &mut impl randomness::Rng,
+) -> Vec<Block> {
+    let mut last_block = prev_blocks.last();
+    let mut new_blocks = Vec::with_capacity(count);
+    for _ in 0..count {
+        let block = make_new_block(
+            chain_config,
+            last_block,
+            &time_getter.get_time_getter(),
+            rng,
+        );
+        time_getter.advance_time(Duration::from_secs(60));
+        new_blocks.push(block);
+        last_block = new_blocks.last();
+    }
+    new_blocks
+}
+
+/// Exhausting the fork download budget must defer the processing of header lists without
+/// any punishment of the peer, and the deferred data must be fetched once the budget is
+/// refilled.
+#[tracing::instrument(skip(seed))]
+#[rstest::rstest]
+#[trace]
+#[case(Seed::from_entropy())]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn fork_download_budget_exhaustion_defers_without_punishment_and_recovers_after_refill(
+    #[case] seed: Seed,
+) {
+    for_each_protocol_version(|protocol_version| async move {
+        let mut rng = make_seedable_rng(seed);
+        let chain_config = Arc::new(create_unit_test_config());
+        let time_getter = BasicTestTimeGetter::new();
+
+        let blocks = make_chain(&chain_config, &[], 10, &time_getter, &mut rng);
+        let mainchain_tip_id = blocks.last().unwrap().get_id();
+        let mainchain_headers = blocks.iter().map(|b| b.header().clone()).collect::<Vec<_>>();
+
+        // `node1` is the node under test; it has a budget of 100 headers.
+        let node1 = TestNode::builder(protocol_version)
+            .with_chain_config(Arc::clone(&chain_config))
+            .with_time_getter(time_getter.get_time_getter())
+            .with_blocks(blocks.clone())
+            .with_p2p_config(Arc::new(make_p2p_config(100)))
+            .build()
+            .await;
+        let chainstate1 = node1.chainstate().clone();
+
+        let node2 = TestNode::builder(protocol_version)
+            .with_chain_config(Arc::clone(&chain_config))
+            .with_time_getter(time_getter.get_time_getter())
+            .with_blocks(blocks)
+            .build()
+            .await;
+        let chainstate2 = node2.chainstate().clone();
+
+        let mut nodes = TestNodeGroup::new(vec![node1, node2]);
+        nodes.set_assert_no_peer_manager_events(true);
+
+        nodes.sync_all(&mainchain_tip_id.into(), &mut rng).await;
+
+        // Exhaust the budget of `node1` by sending it an already-known header list 12 times.
+        // Each list charges 10 tokens and the capacity is 100, so the budget is guaranteed
+        // to be exhausted regardless of what was charged during the bootstrap.
+        for _ in 0..12 {
+            nodes.send_sync_message(known_list_msg(&mainchain_headers)).await;
+        }
+        nodes.exchange_block_sync_messages(&mut rng).await;
+
+        // The budget is exhausted, so a newly produced header list must be deferred: no
+        // block download must happen and the peer must not be punished for the
+        // announcement (any ban/discouragement event fails the test).
+        let fork_headers = make_new_top_blocks_return_headers(
+            &chainstate2,
+            time_getter.get_time_getter(),
+            &mut rng,
+            0,
+            1,
+        )
+        .await;
+        let fork_tip_id = fork_headers.last().unwrap().block_id();
+        nodes.exchange_block_sync_messages(&mut rng).await;
+
+        assert_eq!(
+            best_block_id(&chainstate1).await,
+            Id::<GenBlock>::from(mainchain_tip_id)
+        );
+        assert!(!has_block(&chainstate1, fork_tip_id).await);
+
+        // Once the budget is refilled, the deferred headers must be fetched and the nodes
+        // must converge.
+        time_getter.advance_time(REFILL_INTERVAL * 2);
+        nodes.sync_all(&fork_tip_id.into(), &mut rng).await;
+
+        assert_eq!(
+            best_block_id(&chainstate1).await,
+            Id::<GenBlock>::from(fork_tip_id)
+        );
+        assert!(has_block(&chainstate1, fork_tip_id).await);
+
+        nodes.join_subsystem_managers().await;
+    })
+    .await;
+}
+
+/// The budget must also apply to announcements that are anchored at our tip. Otherwise an
+/// adversary could produce an unlimited number of valid blocks that share the same parent
+/// (each of them being a tip-anchored single-header announcement) and make us download and
+/// process all of them for free.
+#[tracing::instrument(skip(seed))]
+#[rstest::rstest]
+#[trace]
+#[case(Seed::from_entropy())]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn tip_anchored_fork_variants_are_budget_limited(#[case] seed: Seed) {
+    for_each_protocol_version(|protocol_version| async move {
+        let mut rng = make_seedable_rng(seed);
+        let chain_config = Arc::new(create_unit_test_config());
+        let time_getter = BasicTestTimeGetter::new();
+
+        let blocks = make_chain(&chain_config, &[], 10, &time_getter, &mut rng);
+        let mainchain_tip_id = blocks.last().unwrap().get_id();
+        let mainchain_headers = blocks.iter().map(|b| b.header().clone()).collect::<Vec<_>>();
+
+        let node1 = TestNode::builder(protocol_version)
+            .with_chain_config(Arc::clone(&chain_config))
+            .with_time_getter(time_getter.get_time_getter())
+            .with_blocks(blocks.clone())
+            .with_p2p_config(Arc::new(make_p2p_config(100)))
+            .build()
+            .await;
+        let chainstate1 = node1.chainstate().clone();
+
+        let node2 = TestNode::builder(protocol_version)
+            .with_chain_config(Arc::clone(&chain_config))
+            .with_time_getter(time_getter.get_time_getter())
+            .with_blocks(blocks)
+            .build()
+            .await;
+        let chainstate2 = node2.chainstate().clone();
+
+        let mut nodes = TestNodeGroup::new(vec![node1, node2]);
+        nodes.set_assert_no_peer_manager_events(true);
+
+        nodes.sync_all(&mainchain_tip_id.into(), &mut rng).await;
+
+        // Exhaust the budget of `node1`.
+        for _ in 0..12 {
+            nodes.send_sync_message(known_list_msg(&mainchain_headers)).await;
+        }
+        nodes.exchange_block_sync_messages(&mut rng).await;
+
+        // Produce 5 consecutive 2-block forks. `make_new_top_blocks_return_headers`
+        // anchors each call at node2's current best block minus one, and each fork's
+        // child outcompetes the previous tip, so fork i >= 2 extends the previous fork's
+        // first block, which `node1` hasn't accepted (its announcement was deferred).
+        // Only the first list is thus genuinely tip-anchored with respect to `node1`;
+        // the later ones would fail the anchor check if they ever reached it. They
+        // can't: the budget is charged before the anchoring check, so every list here
+        // is deferred (and nothing is punished), which is exactly the ordering this
+        // test pins down.
+        let mut variant_ids = Vec::new();
+        for _ in 0..5 {
+            let headers = make_new_top_blocks_return_headers(
+                &chainstate2,
+                time_getter.get_time_getter(),
+                &mut rng,
+                1,
+                2,
+            )
+            .await;
+            variant_ids.extend(headers.iter().map(|h| h.block_id()));
+        }
+        let best_variant_id = *variant_ids.last().unwrap();
+        nodes.exchange_block_sync_messages(&mut rng).await;
+
+        assert_eq!(
+            best_block_id(&chainstate1).await,
+            Id::<GenBlock>::from(mainchain_tip_id)
+        );
+        for variant_id in &variant_ids {
+            assert!(!has_block(&chainstate1, *variant_id).await);
+        }
+
+        // After the budget is refilled, the nodes must converge on the peer's best chain
+        // without any punishment of the peer.
+        time_getter.advance_time(REFILL_INTERVAL * 2);
+        nodes.sync_all(&best_variant_id.into(), &mut rng).await;
+
+        assert_eq!(
+            best_block_id(&chainstate1).await,
+            Id::<GenBlock>::from(best_variant_id)
+        );
+        assert!(has_block(&chainstate1, best_variant_id).await);
+
+        nodes.join_subsystem_managers().await;
+    })
+    .await;
+}
+
+/// The fork download budget must not be enforced during the initial block download,
+/// otherwise a node that is catching up with the network could be stalled by the budget.
+#[tracing::instrument(skip(seed))]
+#[rstest::rstest]
+#[trace]
+#[case(Seed::from_entropy())]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn fork_download_budget_is_not_enforced_during_initial_block_download(#[case] seed: Seed) {
+    for_each_protocol_version(|protocol_version| async move {
+        let mut rng = make_seedable_rng(seed);
+        let chain_config = Arc::new(create_unit_test_config());
+
+        // Both nodes have a 10-block chain whose tip is 2 days old, which is older than the
+        // default max tip age (24 hours), so `node2` considers itself to be in the initial
+        // block download.
+        let old_time_getter = BasicTestTimeGetter::with_secs_since_epoch(
+            std::time::SystemTime::now()
+                .duration_since(std::time::SystemTime::UNIX_EPOCH)
+                .unwrap()
+                .as_secs()
+                .saturating_sub(2 * 24 * 60 * 60),
+        );
+
+        let blocks = make_chain(&chain_config, &[], 10, &old_time_getter, &mut rng);
+
+        // `node1` also has a 30-block fork that `node2` must download in a single
+        // 30-header list, which exceeds the budget of 10 headers.
+        let fork_blocks = make_chain(&chain_config, &blocks, 30, &old_time_getter, &mut rng);
+        let fork_tip_id = fork_blocks.last().unwrap().get_id();
+
+        // `node1` follows the fork, so its tip is "fresh" relative to its own (old) clock
+        // and it is not in the initial block download.
+        let mut node1_blocks = blocks.clone();
+        node1_blocks.extend(fork_blocks);
+
+        let node1 = TestNode::builder(protocol_version)
+            .with_chain_config(Arc::clone(&chain_config))
+            .with_time_getter(old_time_getter.get_time_getter())
+            .with_blocks(node1_blocks)
+            .build()
+            .await;
+
+        let node2 = TestNode::builder(protocol_version)
+            .with_chain_config(Arc::clone(&chain_config))
+            .with_time_getter(BasicTestTimeGetter::new().get_time_getter())
+            .with_blocks(blocks)
+            .with_p2p_config(Arc::new(make_p2p_config(10)))
+            .build()
+            .await;
+        let chainstate2 = node2.chainstate().clone();
+
+        let mut nodes = TestNodeGroup::new(vec![node1, node2]);
+        nodes.set_assert_no_peer_manager_events(true);
+
+        // The 30-header list sent in response to the header list request of `node2` must
+        // not be deferred despite exceeding the budget.
+        nodes.sync_all(&fork_tip_id.into(), &mut rng).await;
+
+        assert_eq!(
+            best_block_id(&chainstate2).await,
+            Id::<GenBlock>::from(fork_tip_id)
+        );
+
+        nodes.join_subsystem_managers().await;
+    })
+    .await;
+}
diff --git a/p2p/src/sync/tests/header_list_request.rs b/p2p/src/sync/tests/header_list_request.rs
index 65d3b8a647..24e78c0554 100644
--- a/p2p/src/sync/tests/header_list_request.rs
+++ b/p2p/src/sync/tests/header_list_request.rs
@@ -235,6 +235,7 @@ async fn locator_must_be_from_peers_known_best_block(#[case] seed: Seed) {
             msg_max_locator_count: Default::default(),
             max_message_size: Default::default(),
             max_peer_tx_announcements: Default::default(),
+            ..Default::default()
         }));
 
         let node_blocks = make_new_blocks(
diff --git a/p2p/src/sync/tests/mod.rs b/p2p/src/sync/tests/mod.rs
index e69621ddb5..68b8e41b38 100644
--- a/p2p/src/sync/tests/mod.rs
+++ b/p2p/src/sync/tests/mod.rs
@@ -17,6 +17,7 @@ mod ban_scores;
 mod block_announcement;
 mod block_list_request;
 mod block_response;
+mod fork_download_budget;
 mod header_list_request;
 mod header_list_response;
 pub mod helpers;
diff --git a/p2p/src/sync/tests/network_sync.rs b/p2p/src/sync/tests/network_sync.rs
index 910c0e9e85..10fa53ef7b 100644
--- a/p2p/src/sync/tests/network_sync.rs
+++ b/p2p/src/sync/tests/network_sync.rs
@@ -68,6 +68,7 @@ async fn basic(#[case] seed: Seed) {
                 msg_max_locator_count: Default::default(),
                 max_message_size: Default::default(),
                 max_peer_tx_announcements: Default::default(),
+                ..Default::default()
             },
 
             bind_addresses: Default::default(),
@@ -310,6 +311,7 @@ async fn block_announcement_disconnected_headers(#[case] seed: Seed) {
                 msg_max_locator_count: Default::default(),
                 max_message_size: Default::default(),
                 max_peer_tx_announcements: Default::default(),
+                ..Default::default()
             },
 
             bind_addresses: Default::default(),
@@ -745,6 +747,7 @@ async fn process_block_interference2(#[case] seed: Seed) {
                 msg_max_locator_count: Default::default(),
                 max_message_size: Default::default(),
                 max_peer_tx_announcements: Default::default(),
+                ..Default::default()
             }));
             let mut node = TestNode::builder(protocol_version)
                 .with_chain_config(chain_config)
@@ -917,6 +920,7 @@ async fn no_infinite_stalling_when_first_locator_cant_locate(#[case] seed: Seed)
             msg_max_locator_count: Default::default(),
             max_message_size: Default::default(),
             max_peer_tx_announcements: Default::default(),
+            ..Default::default()
         }));
 
         let common_blocks = make_new_blocks(
diff --git a/p2p/src/sync/tests/tx_announcement.rs b/p2p/src/sync/tests/tx_announcement.rs
index 81ef9d57cf..9546114aa6 100644
--- a/p2p/src/sync/tests/tx_announcement.rs
+++ b/p2p/src/sync/tests/tx_announcement.rs
@@ -246,6 +246,7 @@ async fn too_many_announcements(#[case] seed: Seed) {
                 max_addr_list_response_address_count: Default::default(),
                 msg_max_locator_count: Default::default(),
                 max_message_size: Default::default(),
+                ..Default::default()
             },
 
             bind_addresses: Default::default(),
diff --git a/p2p/src/tests/unsupported_message.rs b/p2p/src/tests/unsupported_message.rs
index fdde0ae0d1..e62adb9061 100644
--- a/p2p/src/tests/unsupported_message.rs
+++ b/p2p/src/tests/unsupported_message.rs
@@ -72,6 +72,7 @@ async fn unsupported_message_impl(seed: Seed, make_msg_too_big: bool) {
         max_addr_list_response_address_count: Default::default(),
         msg_max_locator_count: Default::default(),
         max_peer_tx_announcements: Default::default(),
+        ..Default::default()
     }));
 
     let mut test_node = TestNode::<Transport>::start(