hopr_protocol_pix/reconstructor/

mod.rs

mod utils;

use std::collections::{HashMap, HashSet};

use hopr_types::{
    crypto::{
        crypto_traits::elliptic_curve::Field,
        prelude::{HalfKey, HalfKeyChallenge, OffchainPublicKey},
    },
    internal::prelude::Acknowledgement,
};
use utils::{AddShareOutcome, RecoveredSsaTail, RecoveredTailCredit, SsaCommitmentBuilder, SsaCycle};
use validator::Validate;

use crate::{
    CoefficientIndex, CompletedShare, ExitAcknowledgementShareProcessor, Group, MAX_POLYS_PER_SSA, PixGroup,
    PixGroupRepr, PixParams, PixScalar, PixSpec, PolynomialIndex, RecoveredSsa, ShareResolution, SsaCommitmentProof,
    SsaCommitmentState, SsaIndex, SsaPolynomialId, SsaRecoveryProgress, TaggedEncryptedPartialSsaShare,
    errors::PixError, types::SsaId,
};

type ReadyResolutionNotifier = Box<dyn Fn() + Send + Sync + 'static>;

/// Compact retirement state for one Session.
///
/// `highest_retired` alone is not enough: a later member of a batch may retire while an earlier
/// member is still live. The exception sets retain only indices that still own a registration or
/// were released before going live and may be retried. Live registrations are bounded by the
/// Session layer's generation limit; a failed pre-live request closes a production Session, whose
/// scope releases its retriable entries. The number of completed cycles is represented by one
/// watermark rather than one allocation each.
#[derive(Default)]
struct RetirementFrontier {
    highest_retired: Option<SsaIndex>,
    registered: HashSet<SsaIndex>,
    retriable: HashSet<SsaIndex>,
    /// The most recently recovered cycle's bounded liveness tail.
    ///
    /// One slot is sufficient and security-critical: the Entry emits cycles in index order, so
    /// only the immediate recovered predecessor can legitimately remain in the Exit's FIFO. A
    /// newer recovered cycle replaces it, preventing old shares from buying service indefinitely
    /// while keeping retirement memory constant over the Session's lifetime.
    recovered_tail: Option<RecoveredSsaTail>,
}

impl RetirementFrontier {
    fn is_retired(&self, index: SsaIndex) -> bool {
        self.highest_retired.is_some_and(|highest| index <= highest)
            && !self.registered.contains(&index)
            && !self.retriable.contains(&index)
    }

    fn register(&mut self, index: SsaIndex) -> bool {
        if self.is_retired(index) {
            return false;
        }
        self.retriable.remove(&index);
        self.registered.insert(index);
        true
    }

    /// Releases a registration that never went live while leaving its index reusable.
    fn abandon(&mut self, index: SsaIndex) {
        if self.registered.remove(&index) {
            self.retriable.insert(index);
        }
    }

    /// Retires a known registration and raises the compact watermark.
    ///
    /// Returning `false` for an unknown index is load-bearing: allowing an arbitrary `retire_ssa`
    /// call to raise the watermark would condemn legitimate lower indices that are registered
    /// later.
    fn retire(&mut self, index: SsaIndex) -> bool {
        let was_registered = self.registered.remove(&index);
        let was_retriable = self.retriable.remove(&index);
        if !was_registered && !was_retriable {
            return false;
        }

        self.highest_retired = Some(match self.highest_retired {
            Some(highest) if highest >= index => highest,
            _ => index,
        });
        true
    }

    fn retire_recovered(&mut self, index: SsaIndex, tail: RecoveredSsaTail) -> bool {
        if !self.retire(index) {
            return false;
        }
        debug_assert_eq!(index, tail.ssa_index());
        self.recovered_tail = Some(tail);
        true
    }
}

type SharedRetirementFrontier = std::sync::Arc<parking_lot::Mutex<RetirementFrontier>>;
type RegisteredExitCommitment<S> = (PixGroup<S>, std::sync::Arc<RegisteredCommitment<S>>);

/// A commitment registration paired with the Session retirement state that governs it.
struct RegisteredCommitment<S: PixSpec> {
    builder: parking_lot::Mutex<SsaCommitmentBuilder<S>>,
    retirement: SharedRetirementFrontier,
}

impl<S: PixSpec> std::ops::Deref for RegisteredCommitment<S> {
    type Target = parking_lot::Mutex<SsaCommitmentBuilder<S>>;

    fn deref(&self) -> &Self::Target {
        &self.builder
    }
}

/// A published cycle paired with the same retirement state as its commitment registration.
struct RegisteredCycle<S: PixSpec> {
    cycle: SsaCycle<S>,
    registration: std::sync::Arc<RegisteredCommitment<S>>,
}

impl<S: PixSpec> std::ops::Deref for RegisteredCycle<S> {
    type Target = SsaCycle<S>;

    fn deref(&self) -> &Self::Target {
        &self.cycle
    }
}

/// Owns one Session's retirement-frontier registration in an [`SsaReconstructor`].
///
/// The Session action driver holds this for its whole lifetime. Dropping it removes the pseudonym
/// from the reconstructor immediately; commitment guards and in-flight workers retain their own
/// `Arc` to the frontier until the publication/retirement race has settled.
#[must_use = "dropping the scope stops tracking retirement state for its Session"]
pub struct SsaRetirementScope<S: PixSpec> {
    reconstructor: std::sync::Weak<SsaReconstructor<S>>,
    pseudonym: S::Pseudonym,
    retirement: SharedRetirementFrontier,
}

/// Rejects an early-recovery fraction that is not a finite value in `0.0..=1.0`.
///
/// A `range` validator does not cover this. Every IEEE comparison against `NaN` is false, so `NaN`
/// satisfies `min = 0.0, max = 1.0` untouched — and it does not stay inert downstream:
/// [`SsaBuilder::check_early_threshold`](utils::SsaBuilder::check_early_threshold) computes
/// `(threshold * num_polys).ceil() as usize`,
/// and casting a `NaN` float to an integer saturates to zero in Rust. The early-recovery signal then
/// fires on the very first reconstructed polynomial rather than at 85 % of them, which is the
/// earliest instant the Exit can possibly ask for its next batch — and the Entry's successor gate,
/// which exists to refuse exactly that, drops the request and leaves the Session to die on its
/// commitment deadline.
fn validate_early_recovery_threshold(threshold: f64) -> Result<(), validator::ValidationError> {
    if threshold.is_finite() && (0.0..=1.0).contains(&threshold) {
        Ok(())
    } else {
        Err(validator::ValidationError::new(
            "early_recovery_threshold must be a finite fraction between 0.0 and 1.0",
        ))
    }
}

/// Configuration for the SSA reconstructor.
#[derive(Debug, Clone, Copy, PartialEq, smart_default::SmartDefault, validator::Validate)]
pub struct SsaReconstructorConfig {
    /// Time until the complete commitment to an SSA must be received.
    ///
    /// Default is 2 minutes.
    #[default(Self::DEFAULT_INCOMPLETE_COMMITMENT_LIFETIME)]
    pub incomplete_commitment_lifetime: std::time::Duration,
    /// Maximum time an SSA cycle can go without progress before it is discarded.
    ///
    /// Measured from the last acknowledged share *anywhere* in the cycle, not per polynomial — see
    /// `SsaCycle` for why that distinction is load-bearing. A cycle that is still being served
    /// therefore never expires, whatever the line rate.
    ///
    /// Default is 30 minutes.
    #[default(Self::DEFAULT_UNUSED_VERIFIER_LIFETIME)]
    pub unused_verifier_lifetime: std::time::Duration,
    /// Maximum number of peers that can be tracked simultaneously with unacknowledged shares.
    ///
    /// Default is 2000, minimum is 10.
    ///
    /// This is a per-*peer* fan-out bound, and it guards the opposite concentration to
    /// [`max_awaiting_acks`](Self::max_awaiting_acks): traffic spread thinly across many
    /// first-relayers. The two cannot both be saturated at once, which is why their product is not
    /// the reconstructor's memory bound — see `PixReconstructorConfig` in `hopr-transport` for the
    /// bound that is.
    #[validate(range(min = 10))]
    #[default(Self::DEFAULT_MAX_TRACKED_PEERS)]
    pub max_tracked_peers: usize,
    /// Maximum number of awaited acknowledgements to extract a single share.
    ///
    /// This corresponds to the maximum number of unacknowledged HOPR packets awaiting acknowledgement.
    ///
    /// Default is 1 000 000, must be at least 10 000.
    ///
    /// Sizes one inner cache **per peer**, so it must cover the concentrated case: every Session on
    /// the node returning through a single first-relayer. At the operating point
    /// `tests/memory_profile.rs` models that is ~542 000 entries, which is what makes a cap of this
    /// order the right one rather than an oversight.
    #[default(Self::DEFAULT_MAX_AWAITING_ACKS)]
    #[validate(range(min = 10000))]
    pub max_awaiting_acks: usize,
    /// Maximum time an acknowledgement can be awaited before it is discarded.
    ///
    /// Default is 30 seconds.
    ///
    /// Multiplies the whole awaiting-ack buffer: the reachable state is the Exit's share-emission
    /// rate times this window, so it — not either cap above — is the dial that actually sizes the
    /// buffer.
    #[default(Self::DEFAULT_MAX_ACK_AWAIT_TIME)]
    pub max_ack_await_time: std::time::Duration,
    /// Indicates whether to use batch verification algorithm for acknowledgements.
    ///
    /// Default is false.
    ///
    /// Batching only covers the acknowledgement *signature* check. While each share also cost a
    /// `threshold`-term multi-scalar multiplication, that MSM dominated and the choice was
    /// immaterial — measured 2.46 MiB/s batched against 2.50 MiB/s unbatched. Committing to the
    /// constant term alone removed the MSM, and the batching overhead is no longer hidden by it:
    /// the same benchmark then measures 50.2 MiB/s batched against 92.8 MiB/s unbatched, so
    /// batching costs 46 % of the sustained rate.
    ///
    /// **Settled against the concurrent pipeline shape**, which is what the Exit actually runs and
    /// what the sequential figures above could not speak to. `concurrent_quota_rate` on 48 cores at
    /// production width, aggregate MiB/s of Session quota:
    ///
    /// | callers | unbatched | batched  |                              |
    /// | ------- | --------- | -------- | ---------------------------- |
    /// | 1       | **90.9**  | 46.4     | unbatched 1.96×              |
    /// | 10      | 130.2     | 126.2    | tie — confidence intervals overlap |
    /// | 48      | 139.8     | **152.6**| batched 1.09×                |
    ///
    /// So batching does eventually pay, but only *above* the concurrency the pipeline is
    /// configured for: `DEFAULT_ACK_INPUT_CONCURRENCY` is 10, which is precisely the row where the
    /// two are indistinguishable. `false` stays the default because it is far better at low
    /// concurrency and no worse at the configured one, making it the safer choice across the range
    /// an operator can set — not because batching is slower everywhere.
    ///
    /// Kept configurable for the operator who raises `ack_input_concurrency` well past its default,
    /// where the last row says the choice flips.
    ///
    /// **Headroom is narrower than it looks.** A deployed Exit serves 10–30 clients at 16–20 Mbps,
    /// so it absorbs 19–72 MiB/s; against the 130 MiB/s above that is **1.8× at the top of the
    /// range**, not the 7.3× an earlier reading of these figures claimed. That claim came from
    /// comparing against the benchmark suite's own model of production — 100 Sessions at 1.5 Mbps,
    /// 18.75 MiB/s — which understates real per-Session rate by 13×. The rates here are also
    /// measured at 4096 polynomials rather than the 512 used previously, which costs about 5 %.
    ///
    /// All of these were measured at 1038 B of quota per share. A share is now priced at the 1452 B
    /// Session MTU, so the same work is ≈ 1.4× more quota: 130.2 MiB/s becomes ≈ 182 MiB/s, and the
    /// headroom at the top of the range ≈ 2.5×.
    #[default(Self::DEFAULT_USE_BATCH_VERIFICATION)]
    pub use_batch_verification: bool,
    /// Fraction of reconstructed polynomials at which to emit an early recovery
    /// notification, triggering pipelined SSA request preparation.
    ///
    /// Range: 0.0..1.0. Default: 0.85.
    ///
    /// A reconstructor driving a real Session is additionally held to
    /// [`MIN_EARLY_RECOVERY_THRESHOLD`](crate::MIN_EARLY_RECOVERY_THRESHOLD), the floor an Entry
    /// assumes when it decides whether a successor `SsaRequest` has been earned. Anything below it is
    /// unusable rather than merely aggressive — the request goes out early and the Entry refuses it.
    /// The bound is not in this validator because it belongs to the pairing rather than to the
    /// reconstructor: `hopr-transport-session` checks it where both halves are known.
    #[default(Self::DEFAULT_EARLY_RECOVERY_THRESHOLD)]
    #[validate(custom(function = "validate_early_recovery_threshold"))]
    pub early_recovery_threshold: f64,
    /// Ceiling on the total live state held in the awaiting-acknowledgement buffer, across every
    /// peer, in bytes.
    ///
    /// This is the *global* bound; [`max_tracked_peers`](Self::max_tracked_peers) and
    /// [`max_awaiting_acks`](Self::max_awaiting_acks) are per-dimension backstops and their product
    /// is not one. See [`SsaReconstructor::insert_encrypted_share`] for why the product overstates
    /// by roughly three thousandfold and why bounding it instead would lose shares.
    ///
    /// Enforced at insertion time rather than by validating a workload model. A model has to assume
    /// a Session count and a packet rate; the node enforces neither — `maximum_managed_sessions`
    /// validates to 100 000 and `SessionCapability::NoRateControl` removes the rate limiter
    /// entirely — so a configuration can be perfectly valid and still exceed any modelled budget.
    /// Counting what is actually held is indifferent to all of that.
    ///
    /// Default is 1 GiB, which at [`AWAITING_ACK_ENTRY_BYTES`] is ~2.68 M shares in flight.
    ///
    /// The minimum is 25 600 B — 64 entries. That is a sanity floor, not a sizing recommendation:
    /// its only job is to stop the budget rounding down to a handful of shares. Whether a given
    /// value is *adequate* depends on the node's traffic, which is exactly the thing this design
    /// stopped trying to predict, so the floor deliberately does not pretend to encode it. A node
    /// configured near it will drop shares and say so.
    #[default(Self::DEFAULT_MAX_ACK_BUFFER_BYTES)]
    #[validate(range(min = 25_600))]
    pub max_ack_buffer_bytes: usize,
}

/// Live heap one entry in the awaiting-acknowledgement buffer costs, in bytes.
///
/// **Measured, not derived.** `size_of` accounts for only 145 B of it — a 33 B `HalfKeyChallenge`
/// key and a 112 B [`TaggedEncryptedPartialSsaShare`] value, both inline arrays — and moka's
/// per-entry bookkeeping (hash map entry, LRU node, TTL timer-wheel node, the `Arc` around the
/// value) is the other 244 B. Run `awaiting_ack_entry_cost` in `tests/memory_profile.rs` to
/// re-derive it; at the time of writing it reports 383 B/entry at 20 000 entries and 389 B/entry at
/// 100 000. Rounded up, because understating it would let
/// [`max_ack_buffer_bytes`](SsaReconstructorConfig::max_ack_buffer_bytes) be exceeded.
///
/// Every entry is this size — the payload is fixed-width inline arrays with no indirection — which
/// is what lets the runtime bound count entries rather than weigh each one.
pub const AWAITING_ACK_ENTRY_BYTES: usize = 400;

/// Per-polynomial live heap a cycle holds that `size_of` cannot account for, in bytes.
///
/// [`peak_cycle_bytes`] reads the structural per-polynomial terms — the part builder, the shared
/// surplus counter, and the commitment map entry retained behind them — straight from the types,
/// so they track the curve and the structs without anyone maintaining a number. This covers what
/// is left:
///
/// * `SsaBuilder`'s received-index set, sized at `num_polys` on construction;
/// * the `Box<[_]>` header the part builders live in, and allocator rounding on both collections;
/// * the acknowledgement-path residue a cycle drags along while it fills. `awaiting_acks` is bounded globally by
///   [`max_ack_buffer_bytes`](SsaReconstructorConfig::max_ack_buffer_bytes) rather than per cycle, but `moka` applies
///   removals on a later maintenance pass, so a cycle being fed at line rate carries a backlog of redeemed-but-not-yet
///   -reclaimed entries. Charging a slice of it twice is deliberate: it is the difference between a model that is a
///   ceiling and one that is merely a good estimate.
///
/// **Measured, not derived**, like [`AWAITING_ACK_ENTRY_BYTES`]:
/// `exit_reconstructor_worst_case_share_order` in `tests/memory_profile.rs` prints the modelled and
/// measured per-polynomial figures side by side with the share buffers subtracted, which is exactly
/// this number. Run it to re-derive every figure below.
///
/// The arithmetic, at the time of writing and at the deployed dimensions:
///
/// * measured per polynomial, share buffers excluded: **~880–905 B**. A range because it is a range: the
///   acknowledgement residue is a scheduling artefact, and two runs of the same test on the same machine gave 876 B and
///   904 B. That variance is the reason this constant is set with headroom rather than at the measurement;
/// * of which `size_of` accounts for **458 B**, leaving a **~420–450 B** residue the types cannot see;
/// * this constant is **704 B**, ~1.6× that residue, so the model charges 458 + 704 = **1162 B** per polynomial — ~30 %
///   above the measurement;
/// * across a whole cycle that dilutes to ~6 %, because the share buffers are modelled exactly and dominate: **41.1
///   MiB** modelled against **38.8 MiB** measured, of which 32.0 MiB is share buffers.
///
/// Both headroom figures are worth keeping straight, because they answer different questions: ~30 %
/// is the margin on the term this constant actually sets, and ~6 % is the margin on the number the
/// Session layer's live-cycle budget is denominated in.
///
/// Headroom rather than the measured figure for the variance above, and because understating this
/// would let that budget be exceeded.
pub const PART_BUILDER_OVERHEAD_BYTES: usize = 704;

/// The per-polynomial share buffer [`peak_cycle_bytes`] models, in bytes.
///
/// `Vec` doubles from a minimum of four elements, so a buffer holding `threshold - 1` shares has a
/// capacity of `threshold.next_power_of_two()` — up to twice what is stored. Modelling the
/// allocation rather than the occupancy is what makes [`peak_cycle_bytes`] a ceiling.
///
/// Public because it is the one term of that model a consumer has reason to subtract back out:
/// `tests/memory_profile.rs` reports the per-polynomial figure both with and without it, since the
/// buffers dominate at the deployed dimensions and the residue is what
/// [`PART_BUILDER_OVERHEAD_BYTES`] is set against. Restating the expression at the call site is what
/// this exists to prevent — an external caller cannot reproduce it even in principle, because
/// `CompletedShare` is crate-private and they would have to assume it is two scalars. True of the
/// shipped specs, and not what the model reads.
pub fn peak_share_buffer_bytes<S: PixSpec>(params: &PixParams) -> u64 {
    (params.shares_per_poly() as u64)
        .next_power_of_two()
        .max(4)
        .saturating_mul(size_of::<CompletedShare<S>>() as u64)
}

/// Worst-case live heap one Exit-side SSA cycle can hold, in bytes.
///
/// This is the figure the Session layer's live-cycle budget is denominated in, and it is
/// deliberately the **adversarial** peak rather than the conforming one. Nothing constrains the
/// order in which an Entry emits shares within a cycle: the shipped generator walks polynomials in
/// blocks of [`SHARE_EMISSION_WINDOW`](crate::SHARE_EMISSION_WINDOW), so only that window holds
/// share buffers at once and a conforming cycle peaks five times below this — 7.8 MiB against
/// 39.1 MiB measured at the deployed dimensions — but a peer running anything else can drive *every*
/// polynomial to one share short of its threshold and hold the lot.
/// `SsaPartBuilder::release_verification_state` frees a buffer when its polynomial reconstructs; a
/// polynomial one share short never reconstructs.
///
/// Not free to the peer, which is why this is a capacity bound and not a security one: a share only
/// reaches the Exit on a return SURB the Exit itself spends, so reaching this peak costs a full
/// quota deposit per cycle. What it bounds is an Exit selling more service than its memory can hold.
///
/// Excludes the awaiting-acknowledgement buffer, which is bounded separately and globally by
/// [`max_ack_buffer_bytes`](SsaReconstructorConfig::max_ack_buffer_bytes) — counting it here would
/// charge the same bytes twice.
pub fn peak_cycle_bytes<S: PixSpec>(params: &PixParams) -> u64 {
    let share_buffer = peak_share_buffer_bytes::<S>(params);

    // Read from the types rather than restated as a constant, so the model follows the curve this
    // node was built for and cannot drift when a field is added to either struct.
    //
    // The commitment map term is the one that surprises: `SsaCommitmentBuilder` is *drained* when
    // the part builders are handed out, but `drain` does not return the allocation and the builder
    // lives until `remove_cycle` — so a completed cycle still holds a bucket per polynomial. The
    // factor of two is hashbrown's power-of-two bucket count at its 87.5 % load factor, and the
    // extra byte is the control byte beside each one.
    let per_poly = size_of::<parking_lot::Mutex<utils::SsaPartBuilder<S>>>() as u64
        // This allocation is shared with the compact recovered tail. It used to be the
        // `surplus_seen` field inside each part builder; keeping it explicit here preserves the
        // admission bound when the counter moves behind an Arc to span recovery.
        + size_of::<std::sync::atomic::AtomicUsize>() as u64
        + 2 * (size_of::<(PolynomialIndex, PixGroup<S>)>() as u64 + 1)
        + PART_BUILDER_OVERHEAD_BYTES as u64;

    (params.polys_per_ssa() as u64).saturating_mul(per_poly + share_buffer)
}

/// The defaults, named so that a mirror can share them instead of restating them.
///
/// `hopr-transport`'s `PixReconstructorConfig` is the operator-facing shape of this struct, and a
/// second copy of these literals over there would be a second thing to keep true. Referencing them
/// from both `#[default(…)]` sites means the two cannot disagree by construction, which is a
/// stronger guarantee than any test comparing them after the fact.
impl SsaReconstructorConfig {
    /// Default [`early_recovery_threshold`](Self::early_recovery_threshold).
    pub const DEFAULT_EARLY_RECOVERY_THRESHOLD: f64 = 0.85;
    /// Default [`incomplete_commitment_lifetime`](Self::incomplete_commitment_lifetime).
    pub const DEFAULT_INCOMPLETE_COMMITMENT_LIFETIME: std::time::Duration = std::time::Duration::from_secs(120);
    /// Default [`max_ack_await_time`](Self::max_ack_await_time).
    pub const DEFAULT_MAX_ACK_AWAIT_TIME: std::time::Duration = std::time::Duration::from_secs(30);
    /// Default [`max_ack_buffer_bytes`](Self::max_ack_buffer_bytes).
    pub const DEFAULT_MAX_ACK_BUFFER_BYTES: usize = 1024 * 1024 * 1024;
    /// Default [`max_awaiting_acks`](Self::max_awaiting_acks).
    pub const DEFAULT_MAX_AWAITING_ACKS: usize = 1_000_000;
    /// Default [`max_tracked_peers`](Self::max_tracked_peers).
    pub const DEFAULT_MAX_TRACKED_PEERS: usize = 2000;
    /// Default [`unused_verifier_lifetime`](Self::unused_verifier_lifetime).
    pub const DEFAULT_UNUSED_VERIFIER_LIFETIME: std::time::Duration = std::time::Duration::from_secs(1800);
    /// Default [`use_batch_verification`](Self::use_batch_verification).
    pub const DEFAULT_USE_BATCH_VERIFICATION: bool = false;
}

type EncryptedShareCache<S> =
    moka::sync::Cache<HalfKeyChallenge, TaggedEncryptedPartialSsaShare<S, <S as PixSpec>::Pseudonym, PixScalar<S>>>;

/// An acknowledgement that arrived before its polynomial's verifier was installed.
///
/// The peer is carried per entry because a polynomial's shares are spread across return paths, and
/// therefore across first-relayers: one bucket can hold deferred acks from several peers.
type DeferredAck = (OffchainPublicKey, HalfKeyChallenge, HalfKey);

/// Deferred acknowledgements for one cycle, drained in one shot when its part builders install.
///
/// Plain `Vec`s under one mutex rather than nested caches: a bucket is only ever appended to and
/// then drained whole, so per-entry cache bookkeeping (and its ~200 B overhead per entry) buys
/// nothing. The mutex serialises deferrals *within one cycle* only, and deferral is O(1) work off
/// the steady-state path.
#[derive(Default)]
struct DeferredAcks {
    by_poly: std::collections::HashMap<PolynomialIndex, Vec<DeferredAck>>,
    /// Running sum of the `by_poly` lengths, maintained so the per-cycle cap is an O(1) check.
    ///
    /// Recomputing it would walk every sub-bucket inside the mutex on every deferral, and there
    /// can be one sub-bucket per entry: filling a bucket to
    /// [`MAX_DEFERRED_ACKS_PER_CYCLE`] would then cost ~33M map-entry visits, on the path every
    /// acknowledgement takes during the commitment window. The invariant
    /// `total == by_poly.values().map(Vec::len).sum()` has one increment site and one reset site,
    /// both under this mutex.
    total: usize,
    /// Set by the one drain this bucket will ever get, in the same critical section as the take.
    ///
    /// A bucket is reachable through two routes: the `pending_acks` key, and an `Arc` a
    /// [`defer_ack`](SsaReconstructor::defer_ack) already holds. The drain removes the first but
    /// cannot revoke the second, so an append that lands after it would sit in a bucket nothing
    /// will ever read again. This flag is how such an append notices — see
    /// [`Deferral::Orphaned`].
    drained: bool,
}

type DeferredAckBucket = std::sync::Arc<parking_lot::Mutex<DeferredAcks>>;

/// What became of an acknowledgement handed to [`defer_ack`](SsaReconstructor::defer_ack).
///
/// Fieldless — the caller still owns the acknowledgement, since [`DeferredAck`] is `Copy` — so this
/// stays a discriminant rather than carrying 264 bytes back out of the critical section.
#[derive(Clone, Copy)]
enum Deferral {
    /// Appended to a live bucket. The drain that installs the cycle will redeem it.
    Buffered,
    /// The bucket had already been drained, so no drain will come for this one. The caller must
    /// redeem it inline; parking it would be silent loss.
    Orphaned,
    /// Over one of the two caps. Already warned about, and deliberately discarded.
    Dropped,
}

/// Cap on deferred acknowledgements held for a single polynomial.
///
/// A conforming Entry emits `threshold + surplus` shares per polynomial — 96 at the default
/// dimensions — across all return paths combined, so at the defaults this cannot be reached without
/// the peer exceeding its own share budget. Anything above the cap is dropped rather than buffered.
///
/// It is *not* unreachable in general: both halves are a byte wide, so a conforming Entry may
/// legitimately announce up to `255 + 255` and have its excess deferrals silently discarded. Both
/// values now travel in [`PixParams`], so an Exit that cares *can* compare
/// `shares_per_poly + surplus_shares` against this cap when it accepts a Session.
///
/// Deliberately not refused on that comparison, though, and the distinction matters. Only shares
/// arriving *before* the cycle's commitments install are deferred at all — a prefix, not the whole
/// emission — so `shares_per_poly + surplus_shares <= MAX_DEFERRED_ACKS_PER_POLYNOMIAL` is a
/// sufficient condition for "no acknowledgement is ever dropped for want of room", not a necessary
/// one for the dimensions to work. Enforcing it would reject legitimate re-splits: `4096 x 128` holds
/// the profiled product exactly and sums to 160. The shipping defaults sum to 96 and are asserted to
/// stay under the cap in `hopr-transport`'s config tests; wider dimensions lose the guarantee and
/// fall back on the deferral window being short.
///
/// Public because it is observable behaviour, not an implementation detail: past the cap an
/// acknowledgement is discarded, so anything measuring or exercising the deferral path has to stay
/// underneath it or it silently measures the discard instead.
pub const MAX_DEFERRED_ACKS_PER_POLYNOMIAL: usize = 128;

/// Cap on deferred acknowledgements held across all polynomials of one cycle.
///
/// The per-polynomial cap alone leaves the cycle total at `num_polys × 128` — a million entries at
/// production dimensions, which is no bound at all. This one is derived rather than chosen: the
/// drain discards any acknowledgement whose share has already left `awaiting_acks`, and that cache
/// expires entries after
/// `max_ack_await_time` (30 s by default). An older deferral is therefore provably dead, so the
/// ceiling only has to cover the shares one cycle can receive inside that window: ~181 shares/s,
/// the per-Session rate this was sized at, so ~5 400. 8192 covers up to ~273 shares/s and costs at most
/// ~786 KB per cycle.
///
/// Public for the same reason as [`MAX_DEFERRED_ACKS_PER_POLYNOMIAL`].
pub const MAX_DEFERRED_ACKS_PER_CYCLE: usize = 8192;

/// Allows server-side reconstruction of SSAs.
///
/// There are 3 inputs that reconstructor is dependent on (in order):
/// 1. SSA commitments from the Client (delivered via
///    [`insert_coefficient_commitments`](ExitAcknowledgementShareProcessor::insert_coefficient_commitments))
/// 2. Extraction of pending encrypted shares (added via
///    [`insert_encrypted_share`](ExitAcknowledgementShareProcessor::insert_encrypted_share)
/// 3. Decryption of pending encrypted shares via [`Acknowledgement`]s (via
///    [`acknowledge_shares`](ExitAcknowledgementShareProcessor::acknowledge_shares))
///
/// It is able to track SSA for multiple different pseudonyms (Sessions).
pub struct SsaReconstructor<S: PixSpec> {
    commitment_builder: moka::sync::Cache<SsaId<S::Pseudonym>, std::sync::Arc<RegisteredCommitment<S>>>,
    /// Post-commitment state of every live cycle: the part accumulator and all part builders,
    /// published and reclaimed as one unit. See [`SsaCycle`].
    ssa_cycles: moka::sync::Cache<SsaId<S::Pseudonym>, std::sync::Arc<RegisteredCycle<S>>>,
    /// One compact retirement frontier per active Session pseudonym.
    ///
    /// A plain `HashMap` under a short synchronous lock is deliberate. Session scopes remove their
    /// entries immediately on teardown; a cache would defer physical removal and put the same
    /// turnover-rate assumption back into the memory bound this state replaces.
    retirement_frontiers: parking_lot::Mutex<HashMap<S::Pseudonym, SharedRetirementFrontier>>,
    awaiting_acks: moka::sync::Cache<OffchainPublicKey, EncryptedShareCache<S>>,
    /// Acknowledgements that arrived before their cycle's part builders were installed, bucketed by
    /// cycle and then by polynomial.
    ///
    /// ## Why bucketed at all
    ///
    /// The bucket key is exactly the thing whose arrival unblocks the entries inside it, so a
    /// bucket is drained once, by the installation of its own cycle, and never scanned
    /// speculatively. That is what keeps [`Self::acknowledge_shares`] free of retry work: it only
    /// ever *appends* to a bucket.
    ///
    /// The original per-peer stash had to be re-scanned in full on every `acknowledge_shares` call,
    /// because a per-peer key says nothing about which entries have become viable. That is
    /// quadratic in the number of acks received while a cycle's commitments are in flight, and the
    /// per-peer key aggregates across every Session sharing a first-relayer.
    ///
    /// ## Why keyed by cycle, sub-bucketed by polynomial
    ///
    /// Part builders are installed for a whole cycle at once, so a per-polynomial *key* no longer
    /// buys selective draining — the drain would just walk every polynomial of the cycle. Keying by
    /// cycle makes it one lookup. The per-polynomial sub-bucket is kept because its cap is what
    /// bounds a misbehaving peer (see [`MAX_DEFERRED_ACKS_PER_POLYNOMIAL`]).
    ///
    /// The capacity unit is cycles, not polynomials. Keyed per polynomial it was `2 *
    /// MAX_POLYS_PER_SSA` entries — which one cycle can exhaust on its own, so past roughly four
    /// concurrent cycles node-wide, moka began LRU-evicting buckets and silently dropping real
    /// shares. A size eviction here is share loss, so the headroom is deliberate and the
    /// `max_ack_await_time` TTL is the operative bound.
    pending_acks: moka::sync::Cache<SsaId<S::Pseudonym>, DeferredAckBucket>,
    /// Resolutions produced by draining deferred-ack buckets at verifier-installation time, waiting
    /// for either the transport notification stream or an acknowledgement batch to collect them.
    ///
    /// Draining happens on the commitment path (`insert_coefficient_commitments`), which is where
    /// the verifier that unblocks the acks is installed. That deliberately keeps the share
    /// verification off the acknowledgement hot path, but it also means the resolutions surface
    /// on a synchronous worker. The buffer bridges that worker to the transport's asynchronous
    /// resolution stream without blocking it or making recovered deposit keys depend on later
    /// acknowledgement traffic.
    ready_resolutions: parking_lot::Mutex<Vec<ShareResolution<S::Pseudonym, S::AddressPrivateKey>>>,
    /// Length of [`ready_resolutions`](Self::ready_resolutions), so the common case (nothing to pick
    /// up) costs one relaxed load instead of a mutex acquisition on every ack batch.
    ready_resolutions_len: std::sync::atomic::AtomicUsize,
    /// Non-blocking wake-up for the asynchronous consumer of [`ready_resolutions`](Self::ready_resolutions).
    ///
    /// The callback carries no data: resolutions remain owned by `ready_resolutions` until a
    /// consumer atomically takes them, so wake-ups may safely coalesce and channel backpressure
    /// cannot drop a recovered deposit key. `OnceLock` prevents two transport streams from racing
    /// as competing designated consumers; `acknowledge_shares` remains an opportunistic fallback.
    ready_resolution_notifier: std::sync::OnceLock<ReadyResolutionNotifier>,
    /// Running estimate of the entries live in [`awaiting_acks`](Self::awaiting_acks), summed over
    /// every peer, so the global budget costs one relaxed load per insertion.
    ///
    /// An *estimate*, and knowingly so — see [`Self::resync_ack_buffer`] for the one drift source
    /// that cannot be listened for and what keeps it from accumulating.
    ///
    /// Behind an `Arc` because each peer's inner cache decrements it from an eviction listener, and
    /// moka requires those to be `'static` — they cannot borrow the reconstructor that owns them.
    ack_buffer_entries: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    /// [`max_ack_buffer_bytes`](SsaReconstructorConfig::max_ack_buffer_bytes) in entries, divided
    /// once here rather than on every insertion.
    max_ack_buffer_entries: usize,
    /// When [`resync_ack_buffer`](Self::resync_ack_buffer) last ran, and the lock that keeps two
    /// from running at once.
    ///
    /// `None` until the first run, so a buffer that saturates immediately is not made to wait out
    /// an interval before its first ground-truth reading.
    ack_buffer_resync: parking_lot::Mutex<Option<std::time::Instant>>,
    cfg: SsaReconstructorConfig,
}

impl<S: PixSpec> Drop for SsaRetirementScope<S> {
    fn drop(&mut self) {
        let Some(reconstructor) = self.reconstructor.upgrade() else {
            return;
        };

        let mut frontiers = reconstructor.retirement_frontiers.lock();
        let owns_entry = frontiers
            .get(&self.pseudonym)
            .is_some_and(|current| std::sync::Arc::ptr_eq(current, &self.retirement));
        if owns_entry {
            frontiers.remove(&self.pseudonym);
        }
    }
}

/// Result of processing a single verified acknowledgement in the SSA reconstructor.
///
/// The counters behind [`SsaRecoveryProgress`] are updated by `process_verified_ack` itself, on the
/// cycle it already holds, so these variants only have to say *whether* a snapshot is worth emitting
/// — not what changed. That is why a duplicate, a surplus share and an unmatched acknowledgement all
/// collapse into [`NoProgress`](Self::NoProgress): none of them moves a counter, so none of them can
/// make a snapshot differ from the last one sent.
enum ProcessedAckResult<S: PixSpec> {
    /// Nothing to report: the acknowledgement matched no pending share, or the share was a duplicate
    /// or was absorbed by an already-failed polynomial.
    ///
    /// A *surplus* share is not in this set. It advances nothing, but it is evidence the Entry is
    /// still serving, and a consumer needs to be able to tell that from silence — see
    /// [`Progressed`](Self::Progressed).
    NoProgress,
    /// The share is valid but its polynomial's verifier is not installed yet, so it cannot be
    /// checked. Deferral, not failure: the ack is bucketed under this
    /// [`SsaPolynomialId`] and retried once the verifier arrives.
    VerifierNotReady(SsaPolynomialId<<S as PixSpec>::Pseudonym>),
    /// A share landed on this cycle without finishing it. The snapshot says whether it advanced
    /// recovery: `useful_shares` moves for a useful share, only `shares_seen` for a surplus one.
    Progressed(SsaRecoveryProgress<<S as PixSpec>::Pseudonym>),
    /// The share failed verification. Carries the SSA's aggregate fault total across all peers.
    InvalidShare(SsaId<<S as PixSpec>::Pseudonym>, u64),
    /// The early recovery threshold was crossed.
    EarlyRecovery(SsaRecoveryProgress<<S as PixSpec>::Pseudonym>),
    /// Full SSA was recovered. Carries the progress at the recovery instant; later buffered shares
    /// may advance only the compact tail's liveness counter.
    FullRecovery(
        RecoveredSsa<<S as PixSpec>::Pseudonym, <S as PixSpec>::AddressPrivateKey>,
        SsaRecoveryProgress<<S as PixSpec>::Pseudonym>,
    ),
}

/// Merges a snapshot into the batch's pending set, keeping the furthest-along one per SSA.
///
/// Concurrent batches share a cycle's counters, so snapshots taken microseconds apart can be
/// unordered. Keeping the maximum means one batch never reports its own SSA going backwards.
///
/// Merged **componentwise** rather than by choosing one snapshot whole, because the counters are two
/// independent axes and a batch can advance either alone. A surplus share moves only `shares_seen`,
/// so selecting on `useful_shares` discarded exactly the snapshots that say an Entry is still
/// serving during a surplus run — the run this crate emits `Progressed` for in the first place, and
/// the axis the Session layer's egress gate and idle deadline both read. Selecting on `shares_seen`
/// instead would only move the same defect onto the other axis, since the two orderings disagree
/// precisely when a snapshot is worth keeping.
///
/// Every field merged here is monotone over a cycle's life, so the componentwise maximum is itself a
/// reachable state rather than an artefact. `target_useful_shares` is fixed at construction and
/// `ssa_id` is the key, so neither is merged.
fn record_progress<P: PartialEq>(acc: &mut Vec<SsaRecoveryProgress<P>>, snapshot: SsaRecoveryProgress<P>) {
    match acc.iter_mut().find(|p| p.ssa_id == snapshot.ssa_id) {
        Some(existing) => {
            existing.useful_shares = existing.useful_shares.max(snapshot.useful_shares);
            existing.shares_seen = existing.shares_seen.max(snapshot.shares_seen);
            existing.recovered_polynomials = existing.recovered_polynomials.max(snapshot.recovered_polynomials);
        }
        None => acc.push(snapshot),
    }
}

/// One SSA's fault observation, with the relayer that carried the offending share.
type FaultObservation<P> = (Box<OffchainPublicKey>, SsaId<P>, u64);

/// Merges a fault observation into the batch's pending set, keeping the highest total per SSA.
///
/// The relayer travels with the total rather than being filled in at emission time: a batch can also
/// carry faults redeemed from deferral, and those were relayed by whoever held the share at the time
/// — not by the peer whose acknowledgements are being processed now.
fn record_fault<P: PartialEq>(acc: &mut Vec<FaultObservation<P>>, observation: FaultObservation<P>) {
    match acc.iter_mut().find(|(_, id, _)| *id == observation.1) {
        Some(existing) if existing.2 >= observation.2 => {}
        Some(existing) => *existing = observation,
        None => acc.push(observation),
    }
}

/// Appends a resolution unless an equal one is already present.
fn push_unique<P: PartialEq, A>(acc: &mut Vec<ShareResolution<P, A>>, resolution: ShareResolution<P, A>) {
    if !acc.contains(&resolution) {
        acc.push(resolution);
    }
}

impl<S: PixSpec + Clone> Default for SsaReconstructor<S> {
    fn default() -> Self {
        Self::new(Default::default())
    }
}

impl<S: PixSpec + Clone> SsaReconstructor<S> {
    fn retirement_frontier(&self, pseudonym: S::Pseudonym) -> SharedRetirementFrontier {
        self.retirement_frontiers
            .lock()
            .entry(pseudonym)
            .or_insert_with(|| std::sync::Arc::new(parking_lot::Mutex::new(RetirementFrontier::default())))
            .clone()
    }

    fn existing_retirement_frontier(&self, pseudonym: &S::Pseudonym) -> Option<SharedRetirementFrontier> {
        self.retirement_frontiers.lock().get(pseudonym).cloned()
    }

    fn retirement_frontier_for(&self, ssa_id: &SsaId<S::Pseudonym>) -> Option<SharedRetirementFrontier> {
        self.commitment_builder
            .get(ssa_id)
            .map(|registration| registration.retirement.clone())
            .or_else(|| {
                self.ssa_cycles
                    .get(ssa_id)
                    .map(|cycle| cycle.registration.retirement.clone())
            })
            .or_else(|| self.existing_retirement_frontier(ssa_id.pseudonym()))
    }

    /// Registers a Session pseudonym for compact retirement tracking.
    ///
    /// The returned scope belongs in the same owner as that Session's
    /// [`SsaCommitmentGuard`]s. Dropping it removes the map entry immediately, while any guards,
    /// published cycles or commitment workers still in flight retain the shared frontier until
    /// their own retirement race has settled.
    pub fn begin_retirement_scope(self: &std::sync::Arc<Self>, pseudonym: S::Pseudonym) -> SsaRetirementScope<S> {
        let retirement = self.retirement_frontier(pseudonym);
        SsaRetirementScope {
            reconstructor: std::sync::Arc::downgrade(self),
            pseudonym,
            retirement,
        }
    }

    /// Creates a new SSA reconstructor from the given configuration.
    ///
    /// Fails if the configuration does not validate. Prefer this over [`Self::new`] anywhere the
    /// configuration is assembled at runtime — a config built programmatically or read from a file
    /// is input, not a constant, and turning it into a panic makes it un-handleable by the caller.
    pub fn try_new(cfg: SsaReconstructorConfig) -> Result<Self, PixError<S::Pseudonym>> {
        cfg.validate()?;
        Ok(Self {
            commitment_builder: moka::sync::Cache::builder()
                .time_to_idle(cfg.incomplete_commitment_lifetime)
                .build(),
            // Indispensable per-cycle state: never size-evicted. Built without a `max_capacity`,
            // so only `time_to_idle` reclaims it. Active removal happens via `remove_cycle` on
            // full recovery and `retire_ssa` on session teardown; the TTL is the backstop.
            // A hard capacity would silently strand a live cycle.
            //
            // The idle timer is refreshed by an acknowledgement for *any* polynomial of the cycle,
            // because the whole cycle is one entry. That is what makes reclamation correct at any
            // line rate — see `SsaCycle`.
            ssa_cycles: moka::sync::Cache::builder()
                .time_to_idle(cfg.unused_verifier_lifetime)
                .build(),
            retirement_frontiers: parking_lot::Mutex::new(HashMap::new()),
            awaiting_acks: moka::sync::CacheBuilder::new(cfg.max_tracked_peers as u64)
                .time_to_idle(cfg.max_ack_await_time)
                // Dropping a peer entry drops its whole inner cache, and dropping a moka handle
                // does not run that cache's eviction listener — so without this, every entry the
                // peer still held would stay counted against the global budget forever. Invalidating
                // routes them through the inner listener instead.
                //
                // `run_pending_tasks()` is deliberately not called here: it is unbounded work on
                // whichever thread happened to trigger maintenance. That leaves the invalidation
                // best-effort, which is precisely why `resync_ack_buffer` exists — this narrows the
                // drift, it does not close it.
                .eviction_listener(|_, shares: EncryptedShareCache<S>, _| shares.invalidate_all())
                .build(),
            // One bucket per cycle, expiring on the same clock as the shares it belongs to: an ack
            // whose share has left `awaiting_acks` can never be used again, so there is nothing to
            // keep. `time_to_live`, not idle — appending to a bucket must not extend the life of
            // entries already in it.
            //
            // The capacity is in cycles. `MAX_POLYS_PER_SSA` is reused only as a generous count of
            // concurrently deferring cycles; a size eviction here is share loss, so it is
            // deliberately far above the pipelining factor and the TTL is the operative bound.
            pending_acks: moka::sync::CacheBuilder::new(MAX_POLYS_PER_SSA as u64)
                .time_to_live(cfg.max_ack_await_time)
                .build(),
            ready_resolutions: parking_lot::Mutex::new(Vec::new()),
            ready_resolutions_len: std::sync::atomic::AtomicUsize::new(0),
            ready_resolution_notifier: std::sync::OnceLock::new(),
            ack_buffer_entries: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            // At least one, so a budget rounded below one entry refuses everything rather than
            // dividing to zero and admitting everything.
            max_ack_buffer_entries: (cfg.max_ack_buffer_bytes / AWAITING_ACK_ENTRY_BYTES).max(1),
            ack_buffer_resync: parking_lot::Mutex::new(None),
            cfg,
        })
    }

    /// Creates a new SSA reconstructor from the given configuration.
    ///
    /// # Panics
    /// Panics if the configuration fails validation. Use [`Self::try_new`] to handle that case
    /// instead.
    pub fn new(cfg: SsaReconstructorConfig) -> Self {
        Self::try_new(cfg).expect("invalid SsaReconstructorConfig")
    }

    /// Returns the configuration of the reconstructor.
    #[inline]
    pub fn config(&self) -> &SsaReconstructorConfig {
        &self.cfg
    }

    /// Installs the non-blocking wake-up used when commitment installation resolves deferred
    /// acknowledgements.
    ///
    /// Returns `false` when a notifier was already installed. The callback is invoked only after
    /// the resolution-buffer mutex has been released, so it may immediately call
    /// [`take_ready_resolutions`](Self::take_ready_resolutions). If resolutions were queued before
    /// registration, registration wakes the consumer once before returning.
    pub fn set_ready_resolution_notifier(&self, notifier: impl Fn() + Send + Sync + 'static) -> bool {
        if self.ready_resolution_notifier.set(Box::new(notifier)).is_err() {
            return false;
        }

        if self.ready_resolutions_len.load(std::sync::atomic::Ordering::Acquire) > 0 {
            self.notify_ready_resolutions();
        }
        true
    }

    #[inline]
    fn notify_ready_resolutions(&self) {
        if let Some(notifier) = self.ready_resolution_notifier.get() {
            notifier();
        }
    }

    /// Returns `true` if the reconstructor still holds a builder (SSA-part
    /// builder or commitment builder) for the given cycle.  Used by tests to
    /// verify that [`retire_ssa`](ExitAcknowledgementShareProcessor::retire_ssa)
    /// cleaned up the expected state.
    pub fn contains_builder(&self, ssa_id: &SsaId<S::Pseudonym>) -> bool {
        self.ssa_cycles.contains_key(ssa_id) || self.commitment_builder.contains_key(ssa_id)
    }

    /// Polynomial-index runs of `ssa_id` whose constant-term commitment never arrived, so the caller
    /// can ask the peer to re-send exactly those.
    ///
    /// A cycle's commitment ships as hundreds of packets and completes only when every one of them
    /// has landed, so without this a single lost packet is a lost Session — nothing else in the
    /// protocol can repair it, since the peer has no way to learn which piece went missing.
    ///
    /// * `None` — no registration for `ssa_id`. It was retired, or it went unanswered for longer than
    ///   [`incomplete_commitment_lifetime`](SsaReconstructorConfig::incomplete_commitment_lifetime), and either way
    ///   there is nothing left for a retransmission to complete.
    /// * `Some([])` — the commitment is complete; there is nothing to ask for. This is the *only* thing an empty vector
    ///   means: a `max_runs` of zero is read as one rather than honoured.
    /// * `Some(runs)` — at most `max_runs` runs, lowest first. A caller that gets exactly `max_runs` back has not seen
    ///   the whole shortfall and should ask again once these are answered.
    ///
    /// `max_runs` is the asking message's capacity, which this crate cannot know, so the caller
    /// supplies it; `SsaCommitmentBuilder::missing_runs` documents how the runs are chosen.
    // That name is deliberately unlinked: `SsaCommitmentBuilder` is crate-private, so an intra-doc
    // link from this public method trips `rustdoc::private_intra_doc_links`, which the
    // `nix build .#docs` job builds as an error.
    pub fn missing_commitment_runs(
        &self,
        ssa_id: &SsaId<S::Pseudonym>,
        max_runs: usize,
    ) -> Option<Vec<(PolynomialIndex, PolynomialIndex)>> {
        self.commitment_builder
            .get(ssa_id)
            .map(|registration| registration.lock().missing_runs(max_runs))
    }

    /// Removes the heavyweight reconstructor state for a single SSA cycle.
    ///
    /// A successfully recovered cycle's compact accounting tail is owned by the Session retirement
    /// frontier, not by these caches, and intentionally survives this operation while paid SURBs
    /// drain.
    ///
    /// Idempotent: invalidating an absent key is a no-op.
    fn remove_cycle(&self, ssa_id: SsaId<S::Pseudonym>) {
        self.ssa_cycles.invalidate(&ssa_id);
        // Deferred acks for a retired cycle can never be redeemed — their part builders will not
        // come back and their shares are about to expire.
        self.pending_acks.invalidate(&ssa_id);
        self.commitment_builder.invalidate(&ssa_id);
    }

    /// Removes a cycle that has reached a terminal state and advances its Session frontier.
    ///
    /// Retirement is recorded *before* the removal, so a commitment completing concurrently sees
    /// the retirement and undoes its own publication rather than resurrecting the cycle.
    ///
    /// This is the operation for a cycle that went **live** and failed terminally. Successful
    /// recovery uses [`retire_recovered_cycle`](Self::retire_recovered_cycle), which performs the
    /// same heavyweight teardown but installs a bounded liveness tail. It is deliberately *not* the
    /// operation for a commitment released before going live: that index is retried at the same
    /// value by design, and retirement would block the retry permanently.
    /// [`release_abandoned_commitment`](Self::release_abandoned_commitment) draws exactly that line,
    /// and escalates to this for the live case.
    fn retire_cycle(&self, ssa_id: SsaId<S::Pseudonym>, retirement: &SharedRetirementFrontier) {
        // Keep the frontier lock through cache invalidation. Registration takes the same lock, so a
        // same-index retry cannot install itself between the retirement marker and the removal and
        // then be mistaken for the state this call intended to tear down.
        let mut frontier = retirement.lock();
        frontier.retire(ssa_id.ssa_index());
        self.remove_cycle(ssa_id);
    }

    /// Retires the heavyweight recovery state while retaining only the bounded allowance needed by
    /// SURBs that were minted before recovery and are still queued in the Exit's FIFO.
    fn retire_recovered_cycle(
        &self,
        ssa_id: SsaId<S::Pseudonym>,
        retirement: &SharedRetirementFrontier,
        tail: RecoveredSsaTail,
    ) {
        // Publication and retirement use this same lock. Installing the tail before invalidating
        // the live cache leaves no miss window in which a legitimate trailing acknowledgement is
        // mistaken for a not-yet-published cycle and deferred.
        let mut frontier = retirement.lock();
        frontier.retire_recovered(ssa_id.ssa_index(), tail);
        self.remove_cycle(ssa_id);
    }

    fn process_verified_ack(
        &self,
        ack: HalfKey,
        ack_challenge: HalfKeyChallenge,
        awaiting_ack_from_peer: &moka::sync::Cache<HalfKeyChallenge, TaggedEncryptedPartialSsaShare<S>>,
    ) -> Result<ProcessedAckResult<S>, PixError<S::Pseudonym>> {
        let Some(share) = awaiting_ack_from_peer.get(&ack_challenge) else {
            tracing::trace!(?ack_challenge, "received ack for unknown share");
            return Ok(ProcessedAckResult::NoProgress);
        };

        let spi = share.ssa_polynomial_id().ok_or(PixError::ShareIsEmpty)?;

        // One lookup for the whole cycle: the part builders and the accumulator are published and
        // reclaimed together, so there is no state in which one is reachable and the other is not.
        // The lookup also refreshes the cycle's idle timer, which is what keeps a cycle that is
        // still being served from being reclaimed underneath itself.
        let Some(cycle) = self.ssa_cycles.get(spi.as_ref()) else {
            // Two states share this miss and only one of them is worth deferring for. A cycle that
            // has not been published yet will arrive, so the share stays put and the caller buckets
            // the ack against it. A cycle that has been retired — recovered, or failed terminally —
            // never will, and deferring holds both the bucket and the share for `max_ack_await_time`
            // to no end.
            //
            // Not a corner case at the deployed dimensions: a conforming Entry ends every emission
            // window in a run of shares that advance nothing, so *every* recovered cycle is followed
            // by acks landing here. Deferring them would hold up to `MAX_DEFERRED_ACKS_PER_CYCLE`
            // dead entries per cycle, and — the part that actually costs something — keep their
            // shares in `awaiting_acks`, whose budget is global. There they crowd out live cycles
            // rather than merely wasting space.
            if let Some(frontier) = self.existing_retirement_frontier(spi.pseudonym())
                && frontier.lock().is_retired(spi.ssa_index())
            {
                // Frees the buffer entry: the per-peer cache's eviction listener decrements the
                // global count on an explicit removal, which is what the redeeming path below
                // relies on too.
                awaiting_ack_from_peer.remove(&ack_challenge);

                // Decrypt before taking the short Session frontier lock below. The tail itself is
                // immutable apart from atomics, but matching and crediting happen under that lock
                // so a newer recovered cycle cannot replace the one slot between those operations.
                let partial_share = share.partial_share.decrypt(spi.pseudonym(), &ack)?;
                let ssa_id = *spi.as_ref();
                let frontier = frontier.lock();
                let Some(tail) = frontier
                    .recovered_tail
                    .as_ref()
                    .filter(|tail| tail.ssa_index() == spi.ssa_index())
                else {
                    tracing::trace!(%spi, "ack for a retired cycle outside the active recovered tail — dropped");
                    return Ok(ProcessedAckResult::NoProgress);
                };

                return match tail.credit_share::<S>(ssa_id, spi.poly_index(), share.nonce, &partial_share) {
                    Ok(RecoveredTailCredit::Credited(progress)) => {
                        tracing::trace!(%spi, "credited buffered surplus against recovered cycle tail");
                        Ok(ProcessedAckResult::Progressed(progress))
                    }
                    Ok(RecoveredTailCredit::OverBudget) => {
                        tracing::debug!(%spi, "buffered share is past the recovered polynomial's negotiated surplus");
                        Ok(ProcessedAckResult::NoProgress)
                    }
                    Ok(RecoveredTailCredit::InvalidPolynomial) => {
                        tracing::error!(%spi, "buffered share names a polynomial outside its recovered cycle");
                        Err(PixError::InvalidInput)
                    }
                    Ok(RecoveredTailCredit::InvalidShare(observed_total)) => {
                        tracing::error!(%spi, observed_total, "invalid buffered share addressed to a recovered polynomial");
                        Ok(ProcessedAckResult::InvalidShare(ssa_id, observed_total))
                    }
                    Err(error) => Err(error),
                };
            }
            // Not an error: the constant-term set is still incomplete, so no part builder exists
            // yet. Leave the share in `awaiting_acks` and hand the caller the key it needs to
            // bucket the ack.
            //
            // A third state also lands here and is deliberately left to the TTL: a cycle that idled
            // out of `ssa_cycles` through `unused_verifier_lifetime` is still registered, not
            // retired, so its late acks are bucketed as though the cycle were still coming. Both
            // the bucket and retained share expire on `max_ack_await_time`; retirement paths above
            // are classified exactly without allocating one historical entry per cycle.
            return Ok(ProcessedAckResult::VerifierNotReady(spi));
        };

        // The polynomial index comes from the peer's own share, so it is untrusted. Once the cycle
        // is known its dimensions are too, which makes an out-of-range index definitively invalid
        // rather than merely early — there is no later state in which it becomes meaningful.
        let Some(part) = cycle.part(spi.poly_index()) else {
            tracing::error!(%spi, num_polys = cycle.num_polys(), "share names a polynomial outside the cycle");
            return Err(PixError::InvalidInput);
        };

        // Cycle confirmed — safe to consume the share.
        awaiting_ack_from_peer.remove(&ack_challenge);

        // The share cannot be empty at this point because we prevent empty share insertions
        let partial_share = share.partial_share.decrypt(spi.pseudonym(), &ack)?;

        let ssa_id = *spi.as_ref();

        // The part lock is released before the accumulator is taken below. That order is the one
        // callers must keep, and neither lock is ever held across the other.
        let ssa_part = match part.lock().add_share(share.nonce, partial_share) {
            Ok(AddShareOutcome::Completed(share)) => {
                tracing::trace!(%spi, "ssa part complete");
                cycle.record_useful_share();
                cycle.record_completed_part();
                share
            }
            Ok(AddShareOutcome::Useful) => {
                tracing::trace!(%spi, "ssa part not yet complete, waiting for more shares");
                cycle.record_useful_share();
                // Snapshot here rather than making the caller look the cycle up again: this is the
                // steady-state outcome for all but one share in `threshold`, so a second cache get
                // would double the lookups on the hot path.
                return Ok(ProcessedAckResult::Progressed(cycle.progress()));
            }
            // Expected traffic, not a fault: a conforming Entry emits `threshold + surplus` shares
            // per polynomial, so every polynomial ends its life absorbing surplus.
            //
            // Reported rather than swallowed, even though it advances nothing. Polynomials in an
            // emission window run in lockstep, so they reach `threshold` together and then take their
            // surplus together — `surplus × window` consecutive shares, 4096 at the deployed
            // dimensions, none of them useful. A consumer that only ever hears about useful shares
            // reads that as the Entry having gone quiet. The snapshot says which it is: `shares_seen`
            // moves, `useful_shares` does not.
            Ok(AddShareOutcome::Surplus) => {
                match cycle.credit_surplus(spi.poly_index()) {
                    Some(true) => {
                        tracing::trace!(%spi, "share arrived after its polynomial was reconstructed");
                        cycle.record_surplus_share();
                        return Ok(ProcessedAckResult::Progressed(cycle.progress()));
                    }
                    Some(false) => {
                        // Past the negotiated surplus the peer is emitting more for this polynomial
                        // than it said it would, and reconstruction has already released the share
                        // set, so these cannot even be told apart as duplicates. Withholding the
                        // snapshot prevents one completed polynomial from refreshing the Exit's
                        // gate indefinitely while the rest of the cycle is untouched.
                        tracing::debug!(%spi, "share arrived past the peer's negotiated surplus for this polynomial");
                        return Ok(ProcessedAckResult::NoProgress);
                    }
                    None => {
                        // `cycle.part` performed the same checked lookup above, so this can only be
                        // an internal disagreement between the part and budget dimensions.
                        tracing::error!(%spi, "surplus budget does not contain the cycle's polynomial");
                        return Err(PixError::InvalidInput);
                    }
                }
            }
            Ok(AddShareOutcome::Duplicate) => {
                tracing::trace!(%spi, "duplicate evaluation identifier");
                return Ok(ProcessedAckResult::NoProgress);
            }
            Ok(AddShareOutcome::Absorbed) => {
                tracing::trace!(%spi, "share for a polynomial that already failed its commitment");
                return Ok(ProcessedAckResult::NoProgress);
            }
            Err(PixError::VsssError(vsss_rs::Error::InvalidShare)) => {
                // Counted rather than raised: the caller reports it as a resolution, and the count
                // has to be taken here because the cycle that holds it is in hand.
                //
                // Almost always this means the polynomial's reconstructed constant term did not
                // open its commitment, in which case the offending share is one of the `threshold`
                // that went into it and cannot be singled out. The whole cycle is lost either way,
                // since the SSA needs every polynomial.
                let observed_total = cycle.record_invalid_share();
                tracing::error!(%spi, observed_total, "ssa part failed to open its commitment");
                return Ok(ProcessedAckResult::InvalidShare(ssa_id, observed_total));
            }
            Err(e) => return Err(e),
        };

        let mut builder_guard = cycle.builder().lock();
        let ssa = match builder_guard.add_recovered_ssa_part(spi.poly_index(), ssa_part) {
            Ok(ssa) => ssa,
            Err(error) => {
                // As terminal as `scalar_to_private_key` returning `None` below, and torn down the
                // same way. Propagating alone would leave the accumulator and every part builder in
                // place, and each further share for the cycle would refresh the idle timer that is
                // supposed to reclaim them — so a Session that keeps sending holds a cycle that can
                // never reconstruct for as long as it likes.
                //
                // The lock goes first: `retire_cycle` drops the last `Arc` to this very cycle.
                drop(builder_guard);
                tracing::error!(%spi, %error, "ssa part could not be added to its accumulator");
                self.retire_cycle(ssa_id, &cycle.registration.retirement);
                return Err(error);
            }
        };
        match ssa {
            Some(scalar) => {
                // Capture the recovery instant before retirement. The counters survive in the
                // compact tail, but any later movement is buffered liveness and must not be folded
                // backwards into the terminal recovery event.
                let progress = cycle.progress();
                // Release the accumulator lock before retiring, so `retire_cycle` — which drops
                // the last `Arc` to this very cycle — does not run while it is held.
                drop(builder_guard);
                let Some(ssa) = S::scalar_to_private_key(scalar) else {
                    tracing::error!(%spi, "ssa reconstruction failed");
                    self.retire_cycle(ssa_id, &cycle.registration.retirement);
                    return Err(PixError::InvalidSsa);
                };
                // Full recovery releases every heavyweight builder immediately. Only the shared
                // counters and per-polynomial remainder survive, so SURBs already queued with this
                // cycle's shares can drain as bounded liveness without retaining the recovered key
                // or giving the Entry a replayable cycle-wide allowance.
                let tail = cycle.recovered_tail();
                self.retire_recovered_cycle(ssa_id, &cycle.registration.retirement, tail);
                tracing::info!(%ssa_id, "ssa recovered");
                Ok(ProcessedAckResult::FullRecovery(RecoveredSsa { ssa_id, ssa }, progress))
            }
            None => {
                tracing::trace!(%spi, "ssa not yet complete, waiting for more ssa parts");
                // Check early threshold while we hold the lock
                let early = builder_guard.check_early_threshold(self.cfg.early_recovery_threshold);
                drop(builder_guard);
                let progress = cycle.progress();
                if early {
                    tracing::info!(%ssa_id, "early recovery threshold reached");
                    Ok(ProcessedAckResult::EarlyRecovery(progress))
                } else {
                    Ok(ProcessedAckResult::Progressed(progress))
                }
            }
        }
    }

    /// Buckets an acknowledgement whose cycle's part builders have not been installed yet.
    ///
    /// O(1) — this is the entire cost the acknowledgement path pays for a deferral.
    fn defer_ack(&self, spi: SsaPolynomialId<S::Pseudonym>, deferred: DeferredAck) {
        let bucket = self.pending_acks.get_with(*spi.as_ref(), || {
            std::sync::Arc::new(parking_lot::Mutex::new(Default::default()))
        });
        self.defer_ack_into(&bucket, spi, deferred);
    }

    /// The bucket half of [`defer_ack`](Self::defer_ack), taking the bucket rather than looking it
    /// up.
    ///
    /// Split out so a test can hold a handle across the drain that invalidates the cache key,
    /// which is the interleaving this guards against and the one thing a single thread cannot
    /// otherwise produce — after the invalidate, `get_with` hands out a *fresh* bucket.
    fn defer_ack_into(&self, bucket: &DeferredAckBucket, spi: SsaPolynomialId<S::Pseudonym>, deferred: DeferredAck) {
        let ssa_id = *spi.as_ref();
        let outcome = {
            let mut bucket = bucket.lock();
            if bucket.drained {
                Deferral::Orphaned
            } else if bucket.total >= MAX_DEFERRED_ACKS_PER_CYCLE {
                // The cycle as a whole is holding more than the shares it could plausibly have
                // received inside `max_ack_await_time`, so the excess cannot be redeemable.
                tracing::warn!(
                    %ssa_id,
                    cap = MAX_DEFERRED_ACKS_PER_CYCLE,
                    "dropping deferred acknowledgement: cycle bucket is full"
                );
                Deferral::Dropped
            } else {
                // Reborrowed off the guard so `by_poly` and `total` are disjoint field borrows.
                let bucket = &mut *bucket;
                let per_poly = bucket.by_poly.entry(spi.poly_index()).or_default();
                if per_poly.len() >= MAX_DEFERRED_ACKS_PER_POLYNOMIAL {
                    // Only reachable if the peer emits more shares for one polynomial than its own
                    // `threshold + surplus` budget allows, so the excess is almost certainly
                    // duplicate.
                    tracing::warn!(
                        %spi,
                        cap = MAX_DEFERRED_ACKS_PER_POLYNOMIAL,
                        "dropping deferred acknowledgement: polynomial bucket is full"
                    );
                    Deferral::Dropped
                } else {
                    per_poly.push(deferred);
                    bucket.total += 1;
                    Deferral::Buffered
                }
            }
        };

        match outcome {
            // The drain took this bucket while we were on our way into it. Redeeming here is what
            // makes the mutex the whole synchronisation point: the drain's take and this append
            // are serialised by it, so exactly one of them owns the ack.
            Deferral::Orphaned => {
                tracing::trace!(%spi, "redeeming an acknowledgement deferred into a drained bucket");
                self.redeem_deferred_acks(&ssa_id, std::iter::once(deferred));
            }
            // Close the race against a concurrent installation. The decision to defer was made on
            // a cycle lookup that missed; if the cycle has appeared since, the drain that would
            // have redeemed this ack may already have run against a bucket we never saw.
            Deferral::Buffered => {
                if self.ssa_cycles.contains_key(&ssa_id) {
                    self.drain_deferred_acks(&ssa_id);
                }
            }
            Deferral::Dropped => {}
        }
    }

    /// Redeems the acknowledgements that were waiting for this cycle's part builders.
    ///
    /// Called from the commitment path immediately after the cycle is installed, so each bucket is
    /// processed exactly once and never speculatively re-scanned. Resolutions are parked in
    /// [`ready_resolutions`](Self::ready_resolutions) for the next `acknowledge_shares` call, since
    /// the commitment path has no route to the upper layer.
    fn drain_deferred_acks(&self, ssa_id: &SsaId<S::Pseudonym>) {
        let Some(bucket) = self.pending_acks.get(ssa_id) else {
            return;
        };
        self.pending_acks.invalidate(ssa_id);

        // Take and tombstone in one critical section. A `defer_ack_into` that looked this bucket
        // up before the invalidate still holds an `Arc` to it; the flag is what stops its append
        // from disappearing into a bucket nothing will read again.
        let deferred = {
            let mut bucket = bucket.lock();
            bucket.drained = true;
            bucket.total = 0;
            std::mem::take(&mut bucket.by_poly)
        };
        if deferred.is_empty() {
            return;
        }

        self.redeem_deferred_acks(ssa_id, deferred.into_values().flatten());
    }

    /// Processes acknowledgements whose verifier has since been installed, parking whatever they
    /// resolve to.
    ///
    /// Shared by the two routes that can redeem a deferral — the drain on the commitment path and
    /// an [`orphaned`](Deferral::Orphaned) append — so both produce the same resolutions in the
    /// same order.
    fn redeem_deferred_acks(&self, ssa_id: &SsaId<S::Pseudonym>, deferred: impl IntoIterator<Item = DeferredAck>) {
        let mut resolved = Vec::new();
        // The furthest-along snapshot any redeemed ack produced. Shares recovered here would
        // otherwise be invisible to the consumer until some later batch happened to touch this same
        // SSA, since only `acknowledge_shares` emits snapshots.
        let mut progress = Vec::new();
        for (peer, challenge, ack) in deferred {
            // The share lives in the peer's own awaiting-acks cache; if the peer entry is gone the
            // share has expired with it and the ack is dead.
            let Some(awaiting) = self.awaiting_acks.get(&peer) else {
                continue;
            };
            match self.process_verified_ack(ack, challenge, &awaiting) {
                Ok(ProcessedAckResult::FullRecovery(ssa, snapshot)) => {
                    record_progress(&mut progress, snapshot);
                    resolved.push(ShareResolution::RecoveredSsa(ssa));
                }
                Ok(ProcessedAckResult::EarlyRecovery(snapshot)) => {
                    let id = snapshot.ssa_id;
                    record_progress(&mut progress, snapshot);
                    resolved.push(ShareResolution::AlmostRecoveredSsa(id));
                }
                Ok(ProcessedAckResult::Progressed(snapshot)) => record_progress(&mut progress, snapshot),
                Ok(ProcessedAckResult::InvalidShare(id, observed_total)) => {
                    tracing::error!(%id, observed_total, "deferred share could not be verified");
                    resolved.push(ShareResolution::InvalidShares {
                        peer: peer.into(),
                        ssa_id: id,
                        observed_total,
                    });
                }
                Ok(ProcessedAckResult::NoProgress) => {}
                Ok(ProcessedAckResult::VerifierNotReady(_)) => {
                    // The cycle was installed and then immediately withdrawn, which only the
                    // retirement path does. Re-bucketing would leak, so drop.
                    tracing::trace!(%ssa_id, "cycle withdrawn while draining deferred acknowledgements");
                }
                Err(error) => tracing::debug!(%ssa_id, %error, "failed to process deferred acknowledgement"),
            }
        }

        // Snapshots go in ahead of the terminal events they belong to, matching what
        // `acknowledge_shares` emits — the consumer sees one order regardless of which path resolved
        // a share.
        if !progress.is_empty() {
            let mut ordered = progress.into_iter().map(ShareResolution::Progress).collect::<Vec<_>>();
            ordered.append(&mut resolved);
            resolved = ordered;
        }

        if !resolved.is_empty() {
            tracing::debug!(%ssa_id, num = resolved.len(), "redeemed deferred acknowledgements");
            let mut ready = self.ready_resolutions.lock();
            ready.extend(resolved);
            self.ready_resolutions_len
                .store(ready.len(), std::sync::atomic::Ordering::Release);
            drop(ready);
            self.notify_ready_resolutions();
        }
    }

    /// Takes any resolutions parked by `drain_deferred_acks`.
    ///
    /// The take is atomic with respect to producers and other consumers. This lets the transport's
    /// notification stream and the acknowledgement fallback race safely: exactly one receives each
    /// queued resolution.
    ///
    /// One atomic load is the entire common-case cost — the buckets are empty whenever the Entry
    /// finishes the constant-term pass before the shares that reference it arrive.
    pub fn take_ready_resolutions(&self) -> Vec<ShareResolution<S::Pseudonym, S::AddressPrivateKey>> {
        if self.ready_resolutions_len.load(std::sync::atomic::Ordering::Acquire) == 0 {
            return Vec::new();
        }
        let mut ready = self.ready_resolutions.lock();
        self.ready_resolutions_len
            .store(0, std::sync::atomic::Ordering::Release);
        std::mem::take(&mut *ready)
    }

    /// Minimum wall time between two [`resync_ack_buffer`](Self::resync_ack_buffer) passes.
    ///
    /// A saturated buffer would otherwise turn every rejected insertion into an
    /// `O(max_tracked_peers)` scan — the overload path amplifying its own cost, which is the shape
    /// of bug this budget exists to prevent.
    ///
    /// Derived from [`max_ack_await_time`](SsaReconstructorConfig::max_ack_await_time) rather than
    /// fixed, because what a resync reclaims is entries that have aged out of *that* window. A
    /// constant would be wrong at both ends: too slow for a short window, so a drained buffer keeps
    /// refusing shares long after it emptied, and needlessly eager for a long one.
    ///
    /// The resulting staleness — up to ~1.9 s at the 30 s default — only bites when the counter has
    /// drifted high *and* nothing is touching the caches, since redemption and any cache access
    /// drive moka's expiry maintenance and fire the listener directly. That is the traffic-stopped
    /// case, where refusing a share costs nothing.
    fn ack_buffer_resync_interval(&self) -> std::time::Duration {
        (self.cfg.max_ack_await_time / 16).max(std::time::Duration::from_millis(1))
    }

    /// Recomputes [`ack_buffer_entries`](Self::ack_buffer_entries) from what the caches actually
    /// hold.
    ///
    /// # Why a counter needs a backstop at all
    ///
    /// Entries leave the buffer four ways: redeemed by their acknowledgement, expired, size-evicted
    /// from their peer's cache, or dropped wholesale when the peer's entry leaves `awaiting_acks`.
    /// The inner eviction listener catches the first three exactly. The fourth cannot be caught:
    /// dropping a moka handle does not run its eviction listener, so the outer listener falls back
    /// to `invalidate_all`, which is best-effort and races an insertion landing on the very cache
    /// being discarded.
    ///
    /// Left alone, that residue only ever accumulates *upward*, and an over-count is far worse than
    /// an under-count: it would eventually refuse every share while the buffer sat empty, turning a
    /// memory ceiling into a permanent outage of the acknowledgement path. (The sibling
    /// `HoprUnacknowledgedTicketProcessor` in `hopr-protocol-hopr` has the same nesting and the same
    /// residue; there it only skews metrics.)
    ///
    /// So the counter is treated as a hint that is allowed to be wrong, and ground truth is
    /// consulted at the one moment being wrong would cost something — when it says the buffer is
    /// full. `try_lock` rather than `lock`: a caller that finds a resync already running should
    /// proceed on the current estimate, not queue up behind it.
    fn resync_ack_buffer(&self) {
        let Some(mut last_run) = self.ack_buffer_resync.try_lock() else {
            return;
        };
        if last_run.is_some_and(|at| at.elapsed() < self.ack_buffer_resync_interval()) {
            return;
        }

        let held = self.count_ack_buffer_entries();
        let previous = self.ack_buffer_entries.swap(held, std::sync::atomic::Ordering::Relaxed);
        *last_run = Some(std::time::Instant::now());

        if previous != held {
            tracing::debug!(
                previous,
                held,
                "resynchronised the awaiting-acknowledgement buffer count"
            );
        }
    }

    /// Ground truth: the entries actually held across every peer.
    ///
    /// Never reads [`ack_buffer_entries`](Self::ack_buffer_entries), so a test asserting the two
    /// agree is testing the counter rather than agreeing with it — the same reason
    /// `deferred_ack_count` recomputes from `by_poly` instead of reading `DeferredAcks::total`.
    ///
    /// `O(max_tracked_peers)`, and each `run_pending_tasks` is bounded by that cache's pending write
    /// queue rather than its size. Both callers keep it off the steady-state path.
    fn count_ack_buffer_entries(&self) -> usize {
        self.awaiting_acks.run_pending_tasks();
        self.awaiting_acks
            .iter()
            .map(|(_, shares)| {
                shares.run_pending_tasks();
                shares.entry_count() as usize
            })
            .sum()
    }

    /// The published cycle, if it is still live.
    #[cfg(test)]
    fn cycle(&self, ssa_id: &SsaId<S::Pseudonym>) -> Option<std::sync::Arc<RegisteredCycle<S>>> {
        self.ssa_cycles.get(ssa_id)
    }

    /// Number of live cycles across all Sessions.
    #[cfg(test)]
    fn live_cycles(&self) -> u64 {
        self.ssa_cycles.run_pending_tasks();
        self.ssa_cycles.entry_count()
    }

    /// Number of part builders installed for a cycle, or `0` if the cycle is not live.
    ///
    /// The per-polynomial cache entry count used to express this. It has to be asked of the cycle
    /// now, because the cache holds one entry per cycle rather than one per polynomial — so
    /// `entry_count()` alone can no longer tell "every part installed" from "one part installed".
    #[cfg(test)]
    fn installed_parts(&self, ssa_id: &SsaId<S::Pseudonym>) -> usize {
        self.ssa_cycles.get(ssa_id).map(|c| c.num_polys()).unwrap_or(0)
    }

    /// Total deferred acknowledgements bucketed for a cycle.
    ///
    /// Recomputed from `by_poly` rather than read off `DeferredAcks::total` deliberately: a
    /// counter that has drifted from the map is exactly what this should catch, and an accessor
    /// reading the counter would agree with it whatever it said.
    #[cfg(test)]
    fn deferred_ack_count(&self, ssa_id: &SsaId<S::Pseudonym>) -> usize {
        self.pending_acks
            .get(ssa_id)
            .map(|b| b.lock().by_poly.values().map(Vec::len).sum())
            .unwrap_or(0)
    }
}

/// Ownership of an Exit SSA commitment, released when dropped.
///
/// Registering an Exit commitment is the first fallible step of many: the request still has to be
/// encoded, sent, and answered. Every early return between here and the point where a permanent
/// owner takes over would otherwise strand the commitment in the reconstructor until its own
/// lifetime expired, and a stranded commitment is not inert — its `SsaId` is occupied, so a retry at
/// the same index is rejected as a duplicate.
///
/// Move-only by design: no `Clone`, no `Copy`, so there is exactly one release point. A success path
/// hands ownership on with [`disarm`](Self::disarm) rather than letting the guard fall out of scope.
///
/// Dropping releases the registration **without** retiring the SSA, so the same index can be
/// requested again — see `SsaReconstructor::release_abandoned_commitment` for why that
/// distinction is load-bearing.
#[must_use = "dropping the guard immediately releases the SSA it owns"]
pub struct SsaCommitmentGuard<S: PixSpec + Clone> {
    /// `None` once disarmed, which is the only state in which `Drop` does nothing.
    owned: Option<OwnedCommitment<S>>,
}

/// What an [`SsaCommitmentGuard`] needs to release its SSA: where it is registered, which one, and
/// the Session frontier that must outlive a concurrent commitment worker.
type OwnedCommitment<S> = (
    std::sync::Arc<SsaReconstructor<S>>,
    SsaId<<S as PixSpec>::Pseudonym>,
    std::sync::Arc<RegisteredCommitment<S>>,
);

/// A registered Exit commitment, paired with ownership of its lifetime.
type GuardedExitCommitment<S> = (PixGroup<S>, SsaCommitmentGuard<S>);

impl<S: PixSpec + Clone> std::fmt::Debug for SsaCommitmentGuard<S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SsaCommitmentGuard")
            .field("ssa_id", &self.owned.as_ref().map(|(_, id, _)| id))
            .finish()
    }
}

impl<S: PixSpec + Clone> SsaCommitmentGuard<S> {
    /// The SSA this guard owns, or `None` if it has been disarmed.
    pub fn ssa_id(&self) -> Option<&SsaId<S::Pseudonym>> {
        self.owned.as_ref().map(|(_, id, _)| id)
    }

    /// Gives up ownership without retiring, returning the SSA that is now the caller's to release.
    ///
    /// Returns `None` if the guard was already disarmed.
    pub fn disarm(mut self) -> Option<SsaId<S::Pseudonym>> {
        self.owned.take().map(|(_, ssa_id, _)| ssa_id)
    }
}

impl<S: PixSpec + Clone> Drop for SsaCommitmentGuard<S> {
    fn drop(&mut self) {
        if let Some((reconstructor, ssa_id, registration)) = self.owned.take() {
            reconstructor.release_abandoned_commitment(ssa_id, &registration);
        }
    }
}

impl<S: PixSpec + Clone> SsaReconstructor<S> {
    fn register_exit_commitment(
        &self,
        id: SsaId<S::Pseudonym>,
        params: PixParams,
    ) -> Result<RegisteredExitCommitment<S>, PixError<S::Pseudonym>> {
        let exit_commitment_secret = PixScalar::<S>::random(&mut hopr_types::crypto_random::rng());
        let exit_commitment_public = PixGroup::<S>::mul_by_generator(&exit_commitment_secret);
        let retirement = self.retirement_frontier(*id.pseudonym());
        let registration = std::sync::Arc::new(RegisteredCommitment {
            builder: parking_lot::Mutex::new(SsaCommitmentBuilder::new(
                id,
                params,
                exit_commitment_secret,
                exit_commitment_public,
            )),
            retirement: retirement.clone(),
        });

        // Registration and retirement serialize on this Session-local lock. The cache insertion is
        // inside the critical section so a retiring call cannot remove the old registration and
        // accidentally remove a same-index retry installed between its marker and invalidation.
        let mut frontier = retirement.lock();
        if frontier.is_retired(id.ssa_index()) || self.ssa_cycles.contains_key(&id) {
            return Err(PixError::DuplicateCommitment);
        }

        self.commitment_builder
            .entry(id)
            .and_try_compute_with(|entry| match entry {
                Some(_) => Err(PixError::DuplicateCommitment),
                None => Ok(moka::ops::compute::Op::Put(registration.clone())),
            })?;

        // The retired check above and this mutation share the same lock, so this cannot fail unless
        // the frontier's own invariants were broken locally.
        let newly_registered = frontier.register(id.ssa_index());
        debug_assert!(newly_registered, "registration raced the frontier under its own lock");
        drop(frontier);

        Ok((exit_commitment_public, registration))
    }

    /// Releases a commitment that was registered but never taken over by an owner.
    ///
    /// Deliberately **not** [`retire_ssa`](ExitAcknowledgementShareProcessor::retire_ssa), which
    /// permanently moves the index behind the Session's retirement frontier. A pre-live release
    /// instead marks the index retriable, so a replacement registration can use the same value.
    ///
    /// Same-index retry is not a corner case: the SSA index is advanced only after every fallible
    /// step of a request has succeeded, so a request that failed keeps its index by design and the
    /// next attempt reuses it.
    ///
    /// Escalates to a full retirement if a cycle did go live, which means the peer was asked and
    /// answered — and therefore that ownership should already have been transferred with
    /// [`disarm`](SsaCommitmentGuard::disarm). That branch is a caller error, and retiring is the
    /// safe response to it, because a live cycle is exactly what the retirement frontier protects.
    fn release_abandoned_commitment(
        &self,
        ssa_id: SsaId<S::Pseudonym>,
        registration: &std::sync::Arc<RegisteredCommitment<S>>,
    ) {
        // Hold the frontier lock through both classification and removal. A completion that already
        // published is retired and removed here; one that publishes later sees either retirement or
        // an abandoned registration and withdraws its own state.
        let mut frontier = registration.retirement.lock();
        let owns_live_cycle = self
            .ssa_cycles
            .get(&ssa_id)
            .is_some_and(|cycle| std::sync::Arc::ptr_eq(&cycle.registration, registration));
        let owns_pending_commitment = self
            .commitment_builder
            .get(&ssa_id)
            .is_some_and(|current| std::sync::Arc::ptr_eq(&current, registration));

        if owns_live_cycle {
            tracing::warn!(%ssa_id, "abandoned ssa commitment was already live — retiring it");
            frontier.retire(ssa_id.ssa_index());
        } else if owns_pending_commitment {
            tracing::debug!(%ssa_id, "releasing ssa commitment abandoned by its owner");
            frontier.abandon(ssa_id.ssa_index());
        } else {
            // This guard belonged to an expired registration that has since been retried. Pointer
            // identity is what prevents the old owner from tearing down the replacement.
            return;
        }
        self.remove_cycle(ssa_id);
    }

    /// [`new_exit_commitment`](ExitAcknowledgementShareProcessor::new_exit_commitment), with the
    /// registration owned by an [`SsaCommitmentGuard`].
    ///
    /// No guard is produced on failure, so a rejected duplicate never retires the registration that
    /// caused the rejection.
    pub fn new_guarded_exit_commitment(
        self: &std::sync::Arc<Self>,
        id: SsaId<S::Pseudonym>,
        params: PixParams,
    ) -> Result<GuardedExitCommitment<S>, PixError<S::Pseudonym>> {
        let (exit_commitment, registration) = self.register_exit_commitment(id, params)?;
        Ok((
            exit_commitment,
            SsaCommitmentGuard {
                owned: Some((self.clone(), id, registration)),
            },
        ))
    }
}

impl<S: PixSpec> Drop for SsaReconstructor<S> {
    /// Reports terminal resolutions that were never collected.
    ///
    /// The transport normally drains this buffer as soon as the commitment path wakes it, with an
    /// acknowledgement batch retained as a fallback consumer. Reaching `Drop` with anything left
    /// therefore means every delivery route has gone away. A `RecoveredSsa` reported here is a
    /// deposit key the Exit held and never handed on.
    fn drop(&mut self) {
        if self.ready_resolutions_len.load(std::sync::atomic::Ordering::Acquire) == 0 {
            return;
        }
        for resolution in self.ready_resolutions.lock().drain(..) {
            tracing::error!(
                ?resolution,
                "pix resolution was never collected and is lost with the reconstructor"
            );
        }
    }
}

impl<S: PixSpec + Clone> ExitAcknowledgementShareProcessor<S> for SsaReconstructor<S> {
    type Error = PixError<S::Pseudonym>;

    fn has_pending_shares(&self, peer: &OffchainPublicKey) -> bool {
        // The parked-resolution check is not redundant with the per-peer one. Callers use this to
        // skip `acknowledge_shares` entirely. The transport notification stream is the primary
        // consumer of `ready_resolutions`, but an acknowledgement batch remains a useful fallback
        // if it wins the race to this global buffer. Its entries name their own `SsaId`, so any
        // batch can correctly carry them out; gating collection behind the producing peer's
        // `awaiting_acks` entry, which expires on its own timer, would strand them for no reason.
        self.ready_resolutions_len.load(std::sync::atomic::Ordering::Acquire) > 0
            || self.awaiting_acks.contains_key(peer)
    }

    fn is_expected_error(&self, error: &Self::Error) -> bool {
        matches!(error, PixError::UnexpectedShare)
    }

    fn retire_ssa(&self, ssa_id: SsaId<S::Pseudonym>) {
        // Every heavyweight key is the SsaId itself, so there is nothing to enumerate and no way for
        // one builder to survive the removal of the rest. The frontier-before-removal ordering the
        // commitment completion path depends on lives in `retire_cycle`, which the terminal
        // acknowledgement paths share with this one.
        //
        // A successfully recovered id is already retired, so `RetirementFrontier::retire` returns
        // false and deliberately leaves its compact tail intact. The Session tombstone is a short
        // diagnostic record; the FIFO can take longer to drain. The one-slot tail is replaced by
        // the next successful recovery and finally released with the Session retirement scope. An
        // entirely unknown id has no frontier to advance and remains the documented idempotent no-op.
        if let Some(retirement) = self.retirement_frontier_for(&ssa_id) {
            self.retire_cycle(ssa_id, &retirement);
        } else {
            self.remove_cycle(ssa_id);
        }
    }

    /// No range check on the dimensions: holding a [`PixParams`] is what proves them in range, since
    /// [`PixParams::try_new`] is the only way to make one and the two bounded fields are as wide as
    /// their types. The check this replaced tested exactly those bounds.
    ///
    /// "The two bounded fields" is `polys_per_ssa` and `shares_per_poly`, and the omission of
    /// `surplus_shares` is deliberate rather than an oversight — worth stating, because the surplus
    /// does reach `SsaPartBuilder::max_credited_surplus` and so does size how much post-reconstruction
    /// traffic one completed polynomial may credit as liveness. `try_new` accepts the whole `u8`
    /// range for it on purpose: the surplus is a *wire* value, `surplus_must_not_exceed_threshold`
    /// bounds only what this node will locally configure, and an extravagant surplus is caught where
    /// it costs the peer something — the Exit's `quota_range`, which prices `polys × (threshold +
    /// surplus)` and so rises with it. A peer declaring a surplus of 255 against a threshold of 2 is
    /// asking to be charged 128× and is refused unless the operator configured a range that admits
    /// it, in which case the traffic it credits is traffic it paid for.
    fn new_exit_commitment(&self, id: SsaId<S::Pseudonym>, params: PixParams) -> Result<PixGroup<S>, Self::Error> {
        self.register_exit_commitment(id, params)
            .map(|(commitment, _)| commitment)
    }

    fn insert_coefficient_commitments(
        &self,
        ssa_id: SsaId<S::Pseudonym>,
        index: CoefficientIndex,
        proof: Option<SsaCommitmentProof<S>>,
        commitments: impl Iterator<Item = (PolynomialIndex, PixGroupRepr<S>)>,
    ) -> Result<SsaCommitmentState<S::Pseudonym, S::DepositAddress>, Self::Error> {
        let mut res = SsaCommitmentState::new(ssa_id);

        // The Server commitment must be present first
        let Some(builder) = self.commitment_builder.get(&ssa_id) else {
            return Err(PixError::MissingSsaCommitment);
        };
        let retirement = builder.retirement.clone();

        let progress = {
            let mut builder = builder.lock();
            res.is_first_encountered = builder.is_empty();
            res.ssa_deposit_address = builder.get_deposit_address().copied();
            builder.add_transposed(index, proof, commitments)?
        };

        // `ssa_deposit_address` was read *before* the insertion, so it being absent here means the
        // address (if any) was discovered by this very call.
        res.deposit_address_first_encountered = res.ssa_deposit_address.is_none();

        let Some(full_ssa_commitment) = progress.full_commitment else {
            res.deposit_address_first_encountered = false; // Not yet encountered
            tracing::trace!(%ssa_id, "ssa commitment not yet complete, waiting for more constant terms");
            return Ok(res);
        };
        res.ssa_deposit_address = Some(S::group_to_deposit_address(full_ssa_commitment).ok_or(PixError::InvalidSsa)?);

        // The accumulator and every part builder go in as one entry. There is deliberately no
        // ordering to get right here: a share can never observe a cycle in which one is reachable
        // and the other is not, which used to be a permanent-drop hazard.
        let mut publication_rejected = false;
        let installed = if let Some(ssa_builder) = progress.ssa_builder {
            let num_polys = ssa_builder.num_polys();
            let cycle = SsaCycle::new(ssa_id, ssa_builder, progress.new_verifiers)?;
            let registered_cycle = std::sync::Arc::new(RegisteredCycle {
                cycle,
                registration: builder.clone(),
            });

            // Registration, abandonment and retirement all take this same Session-local lock. A
            // stale worker therefore cannot overwrite a newer retry: it verifies pointer identity
            // and publishes while every state transition for this index is excluded.
            let frontier = retirement.lock();
            let registration_is_current = self
                .commitment_builder
                .get(&ssa_id)
                .is_some_and(|current| std::sync::Arc::ptr_eq(&current, &builder));
            let was_retired = frontier.is_retired(ssa_id.ssa_index());
            if !registration_is_current || was_retired {
                publication_rejected = true;
                false
            } else {
                self.ssa_cycles.insert(ssa_id, registered_cycle);

                // Keep the check after publication as well. Cache expiry is independent of the
                // frontier lock; if it withdraws this registration during insertion, undo the cycle
                // before releasing the lock and allowing a retry.
                let registration_is_current = self
                    .commitment_builder
                    .get(&ssa_id)
                    .is_some_and(|current| std::sync::Arc::ptr_eq(&current, &builder));
                if !registration_is_current {
                    self.remove_cycle(ssa_id);
                    publication_rejected = true;
                    false
                } else {
                    tracing::debug!(%ssa_id, num_polys, "ssa commitment known — cycle is live");
                    true
                }
            }
        } else {
            false
        };

        if publication_rejected {
            tracing::trace!(%ssa_id, "ssa commitment progressed after its registration ended — dropped published state");
            res.deposit_address_first_encountered = false;
            return Ok(res);
        }

        // Installing the cycle unblocks every acknowledgement bucketed under it. Doing this here
        // rather than on the acknowledgement path is what keeps `acknowledge_shares` free of retry
        // scanning.
        if installed {
            self.drain_deferred_acks(&ssa_id);
        }

        res.is_verifiable = progress.fully_committed;
        if progress.fully_committed {
            tracing::trace!(%ssa_id, "ssa commitment completed");
        }

        Ok(res)
    }

    /// Buffers an encrypted share until its acknowledgement arrives, subject to the global byte
    /// budget.
    ///
    /// # The bound is global, and it is not the product of the two caps
    ///
    /// [`max_tracked_peers`](SsaReconstructorConfig::max_tracked_peers) and
    /// [`max_awaiting_acks`](SsaReconstructorConfig::max_awaiting_acks) bound one dimension each,
    /// and their product — 2 000 × 1 000 000 by default, some 800 GB — is neither reachable nor the
    /// right thing to bound. Not reachable, because an entry exists only for a share this node has
    /// already *sent*, so filling it would take 66 M packets/s of egress inside the default 30 s
    /// window. Not right, because the two guard **mutually exclusive** concentrations:
    /// `max_awaiting_acks` sizes one cache per peer and has to cover every Session returning through
    /// a single first-relayer, while `max_tracked_peers` covers traffic spread thin. Squeezing the
    /// product would push one of them below what its own case needs, and a `max_awaiting_acks` set
    /// too low does not save memory — it size-evicts shares before their acknowledgements arrive.
    ///
    /// So the real bound is [`max_ack_buffer_bytes`](SsaReconstructorConfig::max_ack_buffer_bytes),
    /// counted here across all peers at once. Validating a workload model instead would not do:
    /// a model has to assume a Session count and a packet rate, and this node enforces neither.
    ///
    /// # Behaviour at the ceiling
    ///
    /// The newest share is refused, rather than the oldest evicted: there is no cheap global
    /// "oldest" across per-peer caches, and the oldest is nearest its TTL anyway. Either way a full
    /// buffer means share loss — the packet is already on the wire and its acknowledgement will find
    /// nothing — which is the honest cost of a hard ceiling. [`PixError::AckBufferFull`] is
    /// deliberately not an expected error so the caller logs it.
    ///
    /// The check and the insertion are not atomic. Concurrent inserters can overshoot the ceiling by
    /// their own number, which is the right trade: a lock on this path would cost more than the few
    /// hundred kilobytes of overshoot it would prevent.
    fn insert_encrypted_share(
        &self,
        peer: &OffchainPublicKey,
        challenge: HalfKeyChallenge,
        tagged_enc_share: TaggedEncryptedPartialSsaShare<S>,
    ) -> Result<(), Self::Error> {
        if tagged_enc_share.partial_share.is_empty() {
            return Err(PixError::ShareIsEmpty);
        }

        // One relaxed load in the steady state. Only a buffer that has actually reached its ceiling
        // pays for the ground-truth pass, and even then at most one caller at a time does.
        if self.ack_buffer_entries.load(std::sync::atomic::Ordering::Relaxed) >= self.max_ack_buffer_entries {
            self.resync_ack_buffer();
            if self.ack_buffer_entries.load(std::sync::atomic::Ordering::Relaxed) >= self.max_ack_buffer_entries {
                tracing::error!(
                    %peer,
                    budget_bytes = self.cfg.max_ack_buffer_bytes,
                    "awaiting-acknowledgement buffer is full — dropping share"
                );
                return Err(PixError::AckBufferFull);
            }
        }

        // Incremented unconditionally, including when this replaces an entry under the same
        // challenge. The inner listener fires for `Replaced` too, so the pair nets to zero.
        self.ack_buffer_entries
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.awaiting_acks
            .get_with_by_ref(peer, || {
                // Inner cache keyed by HalfKeyChallenge — each entry gets its own TTL
                // so a late-arriving share gets the full max_ack_await_time window.
                let released = self.ack_buffer_entries.clone();
                moka::sync::CacheBuilder::new(self.cfg.max_awaiting_acks as u64)
                    .time_to_live(self.cfg.max_ack_await_time)
                    // Every way an entry can leave this cache — redeemed by its acknowledgement,
                    // expired, or size-evicted — arrives here, which is what keeps the global count
                    // falling without the removal sites having to know about it.
                    .eviction_listener(move |_, _, _| {
                        released.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
                    })
                    .build()
            })
            .insert(challenge, tagged_enc_share);

        Ok(())
    }

    fn acknowledge_shares(
        &self,
        peer: OffchainPublicKey,
        acks: Vec<Acknowledgement>,
    ) -> Result<Vec<ShareResolution<S::Pseudonym, S::AddressPrivateKey>>, Self::Error> {
        let Some((awaiting_ack_from_peer, half_keys_challenges)) = crate::ack_verify::verify_expected_acknowledgements(
            peer,
            acks,
            &self.awaiting_acks,
            self.cfg.use_batch_verification,
        ) else {
            return Err(PixError::UnexpectedShare);
        };

        // Three accumulators rather than one deduplicating set, because the emission contract is an
        // ordering as well as a multiplicity: every SSA's counters precede its terminal event.
        //
        // All three are `Vec`s deduplicated by linear scan. A per-peer batch resolves shares for the
        // cycle that peer is relaying for — one, or a small handful while a Session pipelines the
        // next — so the scan is over a one- or two-element list and a hash set would cost more than
        // it saves.
        let mut progress: Vec<SsaRecoveryProgress<S::Pseudonym>> = Vec::new();
        let mut faults: Vec<FaultObservation<S::Pseudonym>> = Vec::new();
        let mut terminal: Vec<ShareResolution<S::Pseudonym, S::AddressPrivateKey>> = Vec::new();

        // Collect anything redeemed while verifiers were being installed. No retry scanning happens
        // here: a deferred ack is retried exactly once, by the installation of the verifier it was
        // waiting for (see `drain_deferred_acks`).
        for resolution in self.take_ready_resolutions() {
            match resolution {
                ShareResolution::Progress(snapshot) => record_progress(&mut progress, snapshot),
                ShareResolution::InvalidShares {
                    peer,
                    ssa_id,
                    observed_total,
                } => record_fault(&mut faults, (peer, ssa_id, observed_total)),
                other => push_unique(&mut terminal, other),
            }
        }

        for (ack, ack_challenge) in half_keys_challenges {
            match self.process_verified_ack(ack, ack_challenge, &awaiting_ack_from_peer) {
                Ok(ProcessedAckResult::FullRecovery(ssa, snapshot)) => {
                    record_progress(&mut progress, snapshot);
                    push_unique(&mut terminal, ShareResolution::RecoveredSsa(ssa));
                }
                Ok(ProcessedAckResult::EarlyRecovery(snapshot)) => {
                    let ssa_id = snapshot.ssa_id;
                    record_progress(&mut progress, snapshot);
                    push_unique(&mut terminal, ShareResolution::AlmostRecoveredSsa(ssa_id));
                }
                Ok(ProcessedAckResult::Progressed(snapshot)) => record_progress(&mut progress, snapshot),
                Ok(ProcessedAckResult::InvalidShare(ssa_id, observed_total)) => {
                    tracing::error!(%ssa_id, observed_total, "encountered share that could not be verified");
                    record_fault(&mut faults, (Box::new(peer), ssa_id, observed_total));
                }
                Ok(ProcessedAckResult::NoProgress) => {}
                Ok(ProcessedAckResult::VerifierNotReady(spi)) => {
                    // The share stays in `awaiting_acks`; bucket the ack under the polynomial whose
                    // verifier it needs, so installing that verifier redeems it.
                    tracing::trace!(%peer, %spi, "verifier not yet installed, deferring acknowledgement");
                    self.defer_ack(spi, (peer, ack_challenge, ack));
                }
                Err(PixError::ShareIsEmpty) => tracing::trace!(%peer, "received empty share"),
                Err(error) => {
                    tracing::error!(%error, "failed to process acknowledgement");
                }
            }
        }

        let mut res = Vec::with_capacity(progress.len() + faults.len() + terminal.len());
        res.extend(progress.into_iter().map(ShareResolution::Progress));
        res.extend(
            faults
                .into_iter()
                .map(|(peer, ssa_id, observed_total)| ShareResolution::InvalidShares {
                    peer,
                    ssa_id,
                    observed_total,
                }),
        );
        res.append(&mut terminal);
        Ok(res)
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use anyhow::Context;
    use hopr_types::{
        crypto::{crypto_traits, prelude::*},
        crypto_random::Randomizable,
        internal::prelude::VerifiedAcknowledgement,
    };
    use vsss_rs::elliptic_curve::Field;

    use super::{utils::SsaBuilder, *};
    use crate::{
        DEFAULT_POLY_THRESHOLD, DEFAULT_POLYS_PER_SSA, DEFAULT_SURPLUS_SHARES, GroupEncoding, PartialSsaShare,
        SsaGeneratorConfig, SsaIndex, SsaShareGenerator,
        tests::TestSpec,
        traits::{EntryShareGenerator, ExitAcknowledgementShareProcessor},
    };

    /// The two structural claims [`peak_cycle_bytes`] rests on, pinned where CI can see them.
    ///
    /// The empirical check is `exit_reconstructor_worst_case_share_order` in
    /// `tests/memory_profile.rs`, which needs a tracking allocator and a production-width cycle and
    /// is therefore `#[ignore]`d. This catches the two ways the model can silently stop being a
    /// ceiling without anyone running that: a part builder growing past the per-polynomial constant,
    /// and `Vec`'s growth policy allocating more than the next power of two for a buffer that stops
    /// one share short of its threshold.
    #[test]
    fn peak_cycle_bytes_rests_on_the_sizes_it_claims() {
        // `Vec` doubles from a minimum of four, which is what makes `next_power_of_two` the
        // allocation for a buffer holding `threshold - 1` shares. Modelled with a same-sized array
        // so the check does not need a live cycle; the equality below keeps the proxy honest.
        assert_eq!(
            64,
            size_of::<CompletedShare<TestSpec>>(),
            "the growth model below assumes a 64 B share"
        );
        for threshold in [2usize, 3, 4, 5, 64, 65, 128, 255] {
            let mut shares: Vec<[u8; 64]> = Vec::new();
            for _ in 0..threshold - 1 {
                shares.push([0; 64]);
            }
            assert!(
                shares.capacity() as u64 <= (threshold as u64).next_power_of_two().max(4),
                "a buffer of {} shares allocated {} slots, past the modelled {}",
                threshold - 1,
                shares.capacity(),
                (threshold as u64).next_power_of_two().max(4)
            );
        }

        // The share buffers are what the model is *for*, so they must dominate it at the shipped
        // dimensions. If the per-polynomial terms ever caught up, the whole "worst share order is
        // the worst case" argument would stop being the thing worth bounding.
        let params = PixParams::try_from_config::<TestSpec>(&SsaGeneratorConfig {
            polynomials_per_ssa: DEFAULT_POLYS_PER_SSA,
            threshold: DEFAULT_POLY_THRESHOLD,
            surplus_shares: DEFAULT_SURPLUS_SHARES,
        })
        .expect("the shipped dimensions must be valid");
        let total = peak_cycle_bytes::<TestSpec>(&params);
        let buffers = DEFAULT_POLYS_PER_SSA as u64
            * (DEFAULT_POLY_THRESHOLD as u64).next_power_of_two()
            * size_of::<CompletedShare<TestSpec>>() as u64;
        assert!(
            buffers * 2 > total,
            "share buffers ({buffers} B) must dominate the modelled cycle ({total} B)"
        );
    }

    #[test]
    fn retirement_frontier_keeps_out_of_order_and_retriable_indices_live() -> anyhow::Result<()> {
        let first = SsaIndex::new(1).context("one is a non-zero SSA index")?;
        let second = SsaIndex::new(2).context("two is a non-zero SSA index")?;
        let unknown = SsaIndex::new(3).context("three is a non-zero SSA index")?;
        let mut frontier = RetirementFrontier::default();

        assert!(!frontier.retire(unknown), "retiring an unknown index must be a no-op");
        assert!(frontier.register(first));
        assert!(frontier.register(second));
        assert!(frontier.retire(second));
        assert!(frontier.is_retired(second));
        assert!(
            !frontier.is_retired(first),
            "a registered lower batch member must survive a later member's retirement"
        );

        frontier.abandon(first);
        assert!(
            !frontier.is_retired(first),
            "an abandoned pre-live index remains reusable below the watermark"
        );
        assert!(frontier.register(first), "same-index retry must be admitted");
        assert!(frontier.retire(first));
        assert!(frontier.is_retired(first));

        Ok(())
    }

    #[test]
    fn sequential_retirements_use_one_compact_frontier() -> anyhow::Result<()> {
        const CYCLES: u32 = 100_000;
        let mut frontier = RetirementFrontier::default();

        for raw in 1..=CYCLES {
            let index = SsaIndex::new(raw).context("the test range contains no zero index")?;
            assert!(frontier.register(index));
            assert!(frontier.retire(index));
        }

        assert_eq!(SsaIndex::new(CYCLES), frontier.highest_retired);
        assert!(frontier.registered.is_empty());
        assert!(frontier.retriable.is_empty());

        Ok(())
    }

    #[test]
    fn dropping_a_retirement_scope_removes_its_session_entry() {
        let reconstructor = std::sync::Arc::new(SsaReconstructor::<TestSpec>::default());
        let pseudonym = SimplePseudonym::random();

        let scope = reconstructor.begin_retirement_scope(pseudonym);
        assert!(reconstructor.retirement_frontiers.lock().contains_key(&pseudonym));

        drop(scope);
        assert!(
            !reconstructor.retirement_frontiers.lock().contains_key(&pseudonym),
            "Session teardown must release its frontier immediately"
        );
    }

    #[test]
    fn ssa_reconstructor_try_new_should_reject_an_invalid_config_without_panicking() {
        let cfg = SsaReconstructorConfig {
            // Outside the validated 0.0..=1.0 range.
            early_recovery_threshold: 1.5,
            ..Default::default()
        };

        assert!(matches!(
            SsaReconstructor::<TestSpec>::try_new(cfg),
            Err(PixError::InvalidConfiguration(_))
        ));
    }

    #[test]
    #[should_panic(expected = "invalid SsaReconstructorConfig")]
    fn ssa_reconstructor_new_should_still_panic_on_an_invalid_config() {
        let _ = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            early_recovery_threshold: 1.5,
            ..Default::default()
        });
    }

    /// A distinct, insertable share and the acknowledgement that redeems it.
    ///
    /// The contents do not matter to the budget — only that each entry has its own key — so this
    /// skips the generator entirely and builds a `PartialSsaShare::default()` under a fresh
    /// acknowledgement key.
    fn budget_share(
        spi: &SsaPolynomialId<SimplePseudonym>,
    ) -> anyhow::Result<(HalfKey, HalfKeyChallenge, TaggedEncryptedPartialSsaShare<TestSpec>)> {
        let ack_key = HalfKey::random();
        let challenge = ack_key.to_challenge()?;
        Ok((
            ack_key,
            challenge,
            TaggedEncryptedPartialSsaShare {
                pseudonym: *spi.pseudonym(),
                nonce: crypto_traits::elliptic_curve::Scalar::<Secp256k1>::random(&mut hopr_types::crypto_random::rng()),
                partial_share: PartialSsaShare::default().encrypt(spi, &ack_key)?,
            },
        ))
    }

    /// A reconstructor whose acknowledgement buffer holds exactly `entries`.
    fn reconstructor_with_ack_budget(entries: usize) -> SsaReconstructor<TestSpec> {
        SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            max_ack_buffer_bytes: entries * AWAITING_ACK_ENTRY_BYTES,
            ..Default::default()
        })
    }

    #[test]
    fn ack_buffer_refuses_shares_past_its_byte_budget() -> anyhow::Result<()> {
        // The validated minimum, so the budget is the smallest one a real config can ask for.
        const BUDGET: usize = 64;

        let reconstructor = reconstructor_with_ack_budget(BUDGET);
        let spi = SsaPolynomialId::new(SsaId::new(SimplePseudonym::random(), 1.try_into()?), 0);
        let peer = OffchainKeypair::random();

        for i in 0..BUDGET {
            let (_, challenge, share) = budget_share(&spi)?;
            reconstructor
                .insert_encrypted_share(peer.public(), challenge, share)
                .map_err(|error| anyhow::anyhow!("insertion {i} must fit the budget: {error}"))?;
        }

        let (_, challenge, share) = budget_share(&spi)?;
        assert!(
            matches!(
                reconstructor.insert_encrypted_share(peer.public(), challenge, share),
                Err(PixError::AckBufferFull)
            ),
            "the share past the budget must be refused rather than buffered"
        );
        assert_eq!(
            BUDGET,
            reconstructor.count_ack_buffer_entries(),
            "the buffer must hold exactly its budget — no more, and the refusal must not have dropped any"
        );

        Ok(())
    }

    /// Redeeming an acknowledgement must give its budget back.
    ///
    /// This is the release path the inner eviction listener exists for: `process_verified_ack`
    /// calls `remove`, which moka reports as an `Explicit` removal. A listener that only handled
    /// expiry and size eviction would leak here, and the buffer would fill permanently on a node
    /// that was working perfectly.
    #[test]
    fn redeeming_an_acknowledgement_returns_its_budget() -> anyhow::Result<()> {
        const BUDGET: usize = 64;

        let reconstructor = reconstructor_with_ack_budget(BUDGET);
        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into()?);
        let spi = SsaPolynomialId::new(ssa_id, 0);
        let peer = OffchainKeypair::random();
        reconstructor.new_exit_commitment(ssa_id, test_params(DEFAULT_POLYS_PER_SSA, DEFAULT_POLY_THRESHOLD))?;

        let mut redeemable = None;
        for _ in 0..BUDGET {
            let (ack_key, challenge, share) = budget_share(&spi)?;
            reconstructor.insert_encrypted_share(peer.public(), challenge, share)?;
            redeemable = Some((ack_key, challenge));
        }

        let (ack_key, challenge) = redeemable.expect("the loop ran at least once");
        let (_, refused, share) = budget_share(&spi)?;
        assert!(
            matches!(
                reconstructor.insert_encrypted_share(peer.public(), refused, share),
                Err(PixError::AckBufferFull)
            ),
            "precondition: the buffer is full"
        );

        // The cycle is not published, so this defers rather than redeeming — and deliberately
        // leaves the share in place. Budget must therefore *not* be returned yet.
        let peer_cache = reconstructor
            .awaiting_acks
            .get(peer.public())
            .expect("the peer holds shares");
        assert!(matches!(
            reconstructor.process_verified_ack(ack_key, challenge, &peer_cache),
            Ok(ProcessedAckResult::VerifierNotReady(_))
        ));
        assert_eq!(
            BUDGET,
            reconstructor.count_ack_buffer_entries(),
            "a deferral retains the share, so it must still be charged for"
        );

        // An outright removal is the redemption path's effect on the buffer.
        peer_cache.remove(&challenge);
        assert_eq!(
            BUDGET - 1,
            reconstructor.count_ack_buffer_entries(),
            "redeeming must free the entry"
        );
        reconstructor
            .insert_encrypted_share(peer.public(), refused, share)
            .map_err(|error| anyhow::anyhow!("the freed slot must be reusable: {error}"))?;

        Ok(())
    }

    /// Expiry must give budget back too — the release path that runs constantly in production.
    ///
    /// Most shares are never acknowledged in the window that matters; they age out. If the TTL did
    /// not release budget, a busy Exit would fill its buffer once and never accept another share.
    #[test]
    fn expiring_shares_return_their_budget() -> anyhow::Result<()> {
        const BUDGET: usize = 64;
        const WINDOW: std::time::Duration = std::time::Duration::from_millis(100);

        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            max_ack_buffer_bytes: BUDGET * AWAITING_ACK_ENTRY_BYTES,
            max_ack_await_time: WINDOW,
            ..Default::default()
        });
        let spi = SsaPolynomialId::new(SsaId::new(SimplePseudonym::random(), 1.try_into()?), 0);
        let peer = OffchainKeypair::random();

        for _ in 0..BUDGET {
            let (_, challenge, share) = budget_share(&spi)?;
            reconstructor.insert_encrypted_share(peer.public(), challenge, share)?;
        }
        let (_, challenge, share) = budget_share(&spi)?;
        assert!(
            matches!(
                reconstructor.insert_encrypted_share(peer.public(), challenge, share),
                Err(PixError::AckBufferFull)
            ),
            "precondition: the buffer is full"
        );

        std::thread::sleep(WINDOW * 2);

        // moka expires lazily, so the entries are still charged for until maintenance runs. The
        // insertion path reaches that through `resync_ack_buffer`, which is what this asserts: a
        // buffer full of expired shares must let the next one in without any caller intervening.
        let (_, challenge, share) = budget_share(&spi)?;
        reconstructor
            .insert_encrypted_share(peer.public(), challenge, share)
            .map_err(|error| anyhow::anyhow!("expired entries must free their budget: {error}"))?;

        assert_eq!(
            1,
            reconstructor.count_ack_buffer_entries(),
            "only the share inserted after the window should remain"
        );

        Ok(())
    }

    /// The counter is a hint; this asserts it does not become a lying one.
    ///
    /// The churn below deliberately includes the drift source no listener can catch — evicting a
    /// peer's whole entry from `awaiting_acks`, which drops its inner cache rather than draining
    /// it. Over-counting is the dangerous direction: it would eventually refuse every share while
    /// the buffer sat empty. `count_ack_buffer_entries` recomputes from the caches instead of
    /// reading the counter, so this compares the counter against the truth rather than against
    /// itself.
    #[test]
    fn the_ack_buffer_counter_does_not_inflate_as_peers_churn() -> anyhow::Result<()> {
        const PEERS: usize = 40;
        const TRACKED: usize = 10;
        const PER_PEER: usize = 5;

        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            // Well above anything inserted here: the point is peer churn, not the ceiling.
            max_ack_buffer_bytes: 4096 * AWAITING_ACK_ENTRY_BYTES,
            // Forces the outer cache to evict peers — and with them, whole inner caches — four
            // times over.
            max_tracked_peers: TRACKED,
            ..Default::default()
        });
        let spi = SsaPolynomialId::new(SsaId::new(SimplePseudonym::random(), 1.try_into()?), 0);

        for _ in 0..PEERS {
            let peer = OffchainKeypair::random();
            for _ in 0..PER_PEER {
                let (_, challenge, share) = budget_share(&spi)?;
                reconstructor.insert_encrypted_share(peer.public(), challenge, share)?;
            }
        }

        let held = reconstructor.count_ack_buffer_entries();
        assert!(
            held <= TRACKED * PER_PEER,
            "the outer cache bounds what can be held: {held} entries across at most {TRACKED} peers"
        );

        // A resync is what makes the counter authoritative again; without the backstop the residue
        // from dropped inner caches would sit on it permanently.
        reconstructor.resync_ack_buffer();
        assert_eq!(
            held,
            reconstructor
                .ack_buffer_entries
                .load(std::sync::atomic::Ordering::Relaxed),
            "the counter must agree with what the caches actually hold after churn"
        );

        Ok(())
    }

    /// The bound no longer depends on a workload model — which is the whole of M2.
    ///
    /// Both caps are set so their product is orders of magnitude past the budget, exactly the
    /// configuration the old modelled validation would have waved through. What is held is decided
    /// by the byte budget and nothing else: no Session count, no assumed packet rate, no
    /// acknowledgement window.
    #[test]
    fn the_ack_buffer_budget_binds_regardless_of_the_configured_caps() -> anyhow::Result<()> {
        const BUDGET: usize = 64;
        /// Enough peers that neither cap is anywhere near binding, few enough that the test is not
        /// dominated by keypair generation.
        const PEERS: usize = 16;

        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            max_ack_buffer_bytes: BUDGET * AWAITING_ACK_ENTRY_BYTES,
            max_tracked_peers: 2000,
            max_awaiting_acks: 1_000_000,
            // An hour, against a 30 s default: the dial the modelled check keyed on, turned to a
            // value it would have rejected outright.
            max_ack_await_time: std::time::Duration::from_secs(3600),
            ..Default::default()
        });
        let spi = SsaPolynomialId::new(SsaId::new(SimplePseudonym::random(), 1.try_into()?), 0);

        // Spread across peers so neither per-peer nor per-dimension cap is anywhere near binding.
        let peers = (0..PEERS).map(|_| OffchainKeypair::random()).collect::<Vec<_>>();
        let mut accepted = 0;
        for i in 0..(BUDGET * 4) {
            let (_, challenge, share) = budget_share(&spi)?;
            if reconstructor
                .insert_encrypted_share(peers[i % PEERS].public(), challenge, share)
                .is_ok()
            {
                accepted += 1;
            }
        }

        assert_eq!(
            BUDGET, accepted,
            "the byte budget must be what binds, not the caps whose product is 2e9 entries"
        );
        Ok(())
    }

    /// Pulls from the generator until it yields a share for `poly_index`, discarding the rest.
    ///
    /// Shares are emitted round-robin across [`crate::SHARE_EMISSION_WINDOW`] polynomials, so a test
    /// that needs to drive one polynomial to its threshold cannot assume consecutive calls stay on
    /// it. Discarding the others is sound here: these tests assert on one polynomial's behaviour,
    /// and a polynomial that never receives its shares simply stays incomplete.
    fn next_share_for_poly(
        generator: &SsaShareGenerator<TestSpec>,
        pseudonym: &SimplePseudonym,
        poly_index: PolynomialIndex,
    ) -> anyhow::Result<([u8; 20], crate::GeneratedShare<TestSpec>)> {
        loop {
            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            let share = generator
                .next_share(pseudonym, &msg)?
                .ok_or_else(|| anyhow::anyhow!("generator must yield a share"))?;
            if share.id.poly_index() == poly_index {
                return Ok((msg, share));
            }
        }
    }

    /// The proof a generated commitment carries, attached only to the constant-term batch — the
    /// shape the wire uses, since that batch is what determines the commitment being opened.
    fn proof_of(
        commitment: &crate::SsaCommitment<TestSpec>,
        coeff_index: CoefficientIndex,
    ) -> Option<SsaCommitmentProof<TestSpec>> {
        (coeff_index == 0).then_some(commitment.commitment_proof)
    }

    /// Proof matching the all-identity commitment sets some fixtures use as filler: their sum is the
    /// identity, whose discrete logarithm is zero, so the proof is honest rather than a bypass.
    fn identity_proof(ssa_id: &SsaId<SimplePseudonym>) -> SsaCommitmentProof<TestSpec> {
        let zero = <PixScalar<TestSpec> as Field>::ZERO;
        SsaCommitmentProof::prove(ssa_id, &zero, &PixGroup::<TestSpec>::mul_by_generator(&zero))
            .expect("identity proof must be constructible")
    }

    /// Negotiated dimensions for a test cycle.
    ///
    /// The surplus is [`DEFAULT_SURPLUS_SHARES`] rather than zero so that no test trips the
    /// per-polynomial liveness credit cap by accident — the tests that mean to exercise that cap
    /// build their own [`PixParams`] with a surplus small enough to reach it.
    fn test_params(polys: u16, threshold: u8) -> PixParams {
        PixParams::try_new_for::<TestSpec>(polys, threshold, DEFAULT_SURPLUS_SHARES)
            .expect("test dimensions must be in range")
    }

    /// The dimension bounds live in [`PixParams::try_new`] now, so what used to be five refusals
    /// from `new_exit_commitment` is five values that cannot be built in the first place.
    ///
    /// Kept rather than deleted, and pointed at the constructor: the property is unchanged — these
    /// dimensions must never reach a commitment builder — and only its enforcement point moved. A
    /// future signature that took the fields loose again would drop it silently.
    #[test]
    fn out_of_range_exit_commitment_dimensions_must_be_unrepresentable() -> anyhow::Result<()> {
        for (polys, threshold) in [
            (0, 2),                     // polys_per_ssa == 0
            (MAX_POLYS_PER_SSA + 1, 2), // polys_per_ssa exceeds MAX
            (2, 0),                     // shares_per_poly == 0
            (2, 1),                     // shares_per_poly below the minimum of 2
        ] {
            assert!(
                PixParams::try_new_for::<TestSpec>(polys, threshold, 0).is_err(),
                "{polys} x {threshold} must not be constructible"
            );
        }

        // The fifth case — a threshold above the maximum — has no representation at all now that
        // the field is as wide as its bound.
        assert_eq!(u8::MAX, crate::MAX_POLY_THRESHOLD);

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into().unwrap());
        assert!(reconstructor.new_exit_commitment(ssa_id, test_params(2, 2)).is_ok());

        Ok(())
    }

    #[test]
    fn reconstructor_invalid_commitment_inputs() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());

        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into()?);

        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;

        // 1. Any non-constant coefficient index is ignored rather than rejected, whether or not it is within the
        //    threshold — PIX commits to nothing but the constant term, and a peer that still sends a full Feldman
        //    matrix must not be treated as hostile.
        for coeff_index in [1, 2, CoefficientIndex::MAX] {
            let ignored =
                reconstructor.insert_coefficient_commitments(ssa_id, coeff_index, None, HashMap::new().into_iter())?;
            assert!(ignored.ssa_deposit_address.is_none());
            assert!(!ignored.is_verifiable);
        }
        assert!(
            reconstructor
                .commitment_builder
                .get(&ssa_id)
                .ok_or_else(|| anyhow::anyhow!("missing builder"))?
                .lock()
                .is_empty(),
            "ignored commitments must not enter the builder's state"
        );

        // 2. Invalid polynomial index (>= polys_per_ssa)
        let mut invalid_poly_map = HashMap::new();
        invalid_poly_map.insert(2 as PolynomialIndex, PixGroupRepr::<TestSpec>::default());
        let result = reconstructor.insert_coefficient_commitments(ssa_id, 0, None, invalid_poly_map.into_iter());
        assert!(matches!(result, Err(PixError::InvalidInput)));

        Ok(())
    }

    #[test]
    fn reconstructor_should_not_accept_client_commitments_without_priod_exit_commitment() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());

        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into()?);

        let mut poly_map = HashMap::new();
        for poly in 0..2 {
            poly_map.insert(poly as PolynomialIndex, PixGroupRepr::<TestSpec>::default());
        }

        let res = reconstructor.insert_coefficient_commitments(ssa_id, 0, None, poly_map.into_iter());

        assert!(matches!(res, Err(PixError::MissingSsaCommitment)));

        Ok(())
    }

    /// Commitments arriving after the set is complete are absorbed, and still report the cycle as
    /// complete.
    ///
    /// This is what makes retransmission safe. The peer re-sends parts of the set on request, and a
    /// re-sent copy travels the same mixed path as the original it repairs — so it can arrive after
    /// the gap it was asked for was filled by the original showing up late. Refusing those as
    /// duplicates would turn every successful repair into an error on the next packet. Nothing can
    /// change at this point anyway: every slot is filled, and the per-slot check cannot even speak
    /// for these, because the map is drained when the part builders are handed out.
    #[test]
    fn commitments_arriving_after_completion_are_absorbed() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());

        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into()?);

        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;

        // Fill every constant term, which is the whole commitment
        let mut poly_map = HashMap::new();
        for poly in 0..2 {
            poly_map.insert(poly as PolynomialIndex, PixGroupRepr::<TestSpec>::default());
        }
        reconstructor.insert_coefficient_commitments(ssa_id, 0, Some(identity_proof(&ssa_id)), poly_map.into_iter())?;

        // A re-delivery of a slot that is already filled, which is what a repair racing its own
        // original looks like.
        let mut resent = HashMap::new();
        resent.insert(0 as PolynomialIndex, PixGroupRepr::<TestSpec>::default());
        let absorbed = reconstructor.insert_coefficient_commitments(ssa_id, 0, None, resent.into_iter())?;
        assert!(
            absorbed.is_verifiable,
            "absorbing a re-delivery must not make a completed cycle look incomplete"
        );

        // A trailing non-constant coefficient is ignored for its own reason: a peer that still emits
        // the full Feldman matrix sends the bulk of it *after* the constant-term pass has completed.
        let mut trailing = HashMap::new();
        trailing.insert(0 as PolynomialIndex, PixGroupRepr::<TestSpec>::default());
        let ignored = reconstructor.insert_coefficient_commitments(ssa_id, 1, None, trailing.into_iter())?;
        assert!(
            ignored.is_verifiable,
            "ignoring a message must not make a completed cycle look incomplete"
        );

        Ok(())
    }

    /// A slot re-offered the commitment it holds is a no-op; re-offered a *different* one, it is
    /// still refused.
    ///
    /// The single-assignment invariant is the one that matters: a slot that has been decoded and
    /// accepted can never be rebound, or a peer could displace a commitment the deposit address was
    /// derived from. Re-delivering the *same* bytes cannot do that, and has to be tolerated because
    /// retransmission produces exactly that case.
    #[test]
    fn a_slot_tolerates_its_own_commitment_and_refuses_a_different_one() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());

        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into()?);

        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;

        // Insert the constant term of poly 0
        let mut poly_map_1 = HashMap::new();
        poly_map_1.insert(0 as PolynomialIndex, PixGroupRepr::<TestSpec>::default());
        reconstructor.insert_coefficient_commitments(ssa_id, 0, None, poly_map_1.into_iter())?;

        // The same commitment for the same slot: absorbed, and it must not consume the slot twice —
        // proven below by the set still completing on poly 1 alone.
        let mut resent = HashMap::new();
        resent.insert(0 as PolynomialIndex, PixGroupRepr::<TestSpec>::default());
        let absorbed = reconstructor.insert_coefficient_commitments(ssa_id, 0, None, resent.into_iter())?;
        assert!(!absorbed.is_verifiable, "poly 1 has still not been committed");

        // A *different* commitment for that slot is a contradiction and must be refused.
        let mut conflicting = HashMap::new();
        conflicting.insert(0 as PolynomialIndex, PixGroup::<TestSpec>::generator().to_bytes());
        let result = reconstructor.insert_coefficient_commitments(ssa_id, 0, None, conflicting.into_iter());
        assert!(matches!(result, Err(PixError::DuplicateCommitment)));

        // Poly 1's constant term is a different slot and must still complete the set.
        let mut poly_map_3 = HashMap::new();
        poly_map_3.insert(1 as PolynomialIndex, PixGroupRepr::<TestSpec>::default());
        let completed = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            Some(identity_proof(&ssa_id)),
            poly_map_3.into_iter(),
        )?;
        assert!(
            completed.is_verifiable,
            "the absorbed re-delivery must not have consumed poly 1's slot"
        );

        Ok(())
    }

    /// The missing-commitment scope is what the Exit puts on the wire to repair a lost packet, so it
    /// has to name the gaps exactly, compactly, and within one message.
    #[test]
    fn missing_commitment_runs_report_the_gaps_and_nothing_else() -> anyhow::Result<()> {
        const POLYS: u16 = 8;
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into()?);

        // An unregistered cycle is distinguishable from a complete one: there is nothing left for a
        // retransmission to complete, which the caller has to treat differently from "ask again".
        assert_eq!(None, reconstructor.missing_commitment_runs(&ssa_id, 8));

        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, 2))?;
        assert_eq!(
            Some(vec![(0, POLYS as PolynomialIndex - 1)]),
            reconstructor.missing_commitment_runs(&ssa_id, 8),
            "before anything arrives the whole set is one run"
        );

        // Two chunk-shaped holes: 2..=3 and 6..=6. Each is what one lost packet leaves behind, since
        // the sender slices the set into consecutive polynomial indices.
        let arrived: HashMap<PolynomialIndex, PixGroupRepr<TestSpec>> = [0, 1, 4, 5, 7]
            .into_iter()
            .map(|poly| (poly as PolynomialIndex, PixGroupRepr::<TestSpec>::default()))
            .collect();
        reconstructor.insert_coefficient_commitments(ssa_id, 0, None, arrived.into_iter())?;
        assert_eq!(
            Some(vec![(2, 3), (6, 6)]),
            reconstructor.missing_commitment_runs(&ssa_id, 8),
            "consecutive gaps must coalesce into one run each"
        );

        // A scope larger than the asking message can carry is truncated lowest-first, so successive
        // asks converge on a prefix instead of interleaving.
        assert_eq!(
            Some(vec![(2, 3)]),
            reconstructor.missing_commitment_runs(&ssa_id, 1),
            "the run budget must bound the reported scope"
        );

        // A budget of zero is read as one rather than honoured, and this is the branch that makes an
        // empty result mean "complete" and nothing else. Honouring it would report an incomplete
        // commitment as having nothing to ask for, and the caller — which branches on exactly that —
        // would stop repairing without a word.
        assert_eq!(
            Some(vec![(2, 3)]),
            reconstructor.missing_commitment_runs(&ssa_id, 0),
            "a zero run budget must not be mistaken for a complete commitment"
        );

        // Filling the gaps completes the set, and a complete set has nothing to ask for. Note the map
        // is *drained* at completion, so reporting emptiness here is not incidental.
        let repaired: HashMap<PolynomialIndex, PixGroupRepr<TestSpec>> = [2, 3, 6]
            .into_iter()
            .map(|poly| (poly as PolynomialIndex, PixGroupRepr::<TestSpec>::default()))
            .collect();
        let state = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            Some(identity_proof(&ssa_id)),
            repaired.into_iter(),
        )?;
        assert!(state.is_verifiable);
        assert_eq!(Some(Vec::new()), reconstructor.missing_commitment_runs(&ssa_id, 8));

        Ok(())
    }

    /// The whole point of the feature: a commitment burst that lost a slice still completes, because
    /// the Exit can name the slice and the Entry can re-send exactly it.
    ///
    /// Without this a single dropped packet costs the cycle — and by then the Entry has usually
    /// already been told to fund its deposit address.
    #[test]
    fn a_lost_slice_of_a_commitment_is_repaired_by_a_scoped_retransmission() -> anyhow::Result<()> {
        const POLYS: u16 = 16;
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, 2))?;

        // Deliver every constant term except one packet's worth, the middle quarter of the set.
        let full = coefficient_of(&commitment, 0, None)?;
        let lost: std::collections::HashSet<PolynomialIndex> = (4..8).collect();
        let delivered: Vec<_> = full
            .iter()
            .copied()
            .filter(|(poly_index, _)| !lost.contains(poly_index))
            .collect();
        let partial =
            reconstructor.insert_coefficient_commitments(ssa_id, 0, proof_of(&commitment, 0), delivered.into_iter())?;
        assert!(
            !partial.is_verifiable && partial.ssa_deposit_address.is_none(),
            "a set one packet short yields no deposit address — this is the failure being repaired"
        );

        // The Exit names the gap; the Entry answers it from the same bytes it sent the first time.
        let runs = reconstructor
            .missing_commitment_runs(&ssa_id, 64)
            .expect("the registration must still be live");
        assert_eq!(vec![(4, 7)], runs);

        let repair = generator.recommit(&pseudonym, SsaIndex::MIN, runs)?;
        let repaired = coefficient_of(&repair, 0, None)?;
        assert_eq!(
            lost,
            repaired.iter().map(|(poly_index, _)| *poly_index).collect(),
            "the retransmission must carry exactly the missing polynomials"
        );
        for (poly_index, resent) in &repaired {
            assert_eq!(
                Some(resent),
                full.iter()
                    .find(|(index, _)| index == poly_index)
                    .map(|(_, original)| original),
                "the retransmission must repeat the original bytes rather than derive new ones"
            );
        }

        let completed =
            reconstructor.insert_coefficient_commitments(ssa_id, 0, proof_of(&repair, 0), repaired.into_iter())?;
        assert!(
            completed.is_verifiable && completed.ssa_deposit_address.is_some(),
            "the repaired set must complete the commitment and yield the deposit address"
        );
        assert!(completed.deposit_address_first_encountered);

        Ok(())
    }

    #[test]
    fn reconstructor_missing_verifier_retains_share() -> anyhow::Result<()> {
        // Regression test for the share-loss race:
        // When the polynomial's verifier is not installed yet, the share must NOT be removed
        // from the awaiting_acks cache — it must remain available for the retry that happens
        // when the verifier arrives.
        //
        // The implementation guarantees this: `process_verified_ack` looks the share up with
        // `.get()` and only `.remove()`s it after the verifier lookup succeeds, so the
        // `VerifierNotReady` deferral leaves the share in place. This test asserts that retention.
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig { ..Default::default() });

        let ack_key = HalfKey::random();
        let challenge = ack_key.to_challenge()?;

        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into()?);
        let spi = SsaPolynomialId::new(ssa_id, 0);

        let partial_share = PartialSsaShare::default().encrypt(&spi, &ack_key)?;
        let peer = OffchainKeypair::random();
        let nonce = crypto_traits::elliptic_curve::Scalar::<Secp256k1>::random(&mut hopr_types::crypto_random::rng());

        reconstructor.new_exit_commitment(ssa_id, test_params(DEFAULT_POLYS_PER_SSA, DEFAULT_POLY_THRESHOLD))?;

        reconstructor.insert_encrypted_share(
            peer.public(),
            challenge,
            TaggedEncryptedPartialSsaShare {
                pseudonym: *spi.pseudonym(),
                nonce,
                partial_share,
            },
        )?;

        // Verify the share exists before processing
        let peer_cache = reconstructor.awaiting_acks.get(peer.public());
        assert!(peer_cache.is_some(), "share must be inserted before processing");
        assert!(
            peer_cache.as_ref().unwrap().contains_key(&challenge),
            "share must be present in the peer cache before processing"
        );

        // Process the ack — the verifier is not installed, so this defers rather than failing,
        // and reports the polynomial the ack must be bucketed under.
        let peer_cache_ref = reconstructor.awaiting_acks.get(peer.public()).unwrap();
        let result = reconstructor.process_verified_ack(ack_key, challenge, &peer_cache_ref);
        assert!(
            matches!(result, Ok(ProcessedAckResult::VerifierNotReady(reported)) if reported == spi),
            "expected deferral naming polynomial {spi:?}"
        );

        // The share MUST NOT be destroyed by the deferral: the implementation only removes it
        // after the verifier lookup succeeds, so it stays available for the retry.
        let peer_cache_after = reconstructor.awaiting_acks.get(peer.public());
        assert!(peer_cache_after.is_some(), "share must be retained when deferred");
        assert!(
            peer_cache_after.as_ref().unwrap().contains_key(&challenge),
            "share must be retained when deferred"
        );

        Ok(())
    }

    #[test]
    fn reconstructor_defers_ack_when_verifier_is_not_installed() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig { ..Default::default() });

        let ack_key = HalfKey::random();
        let challenge = ack_key.to_challenge()?;

        // Add a pending share but NO commitment (so no verifier is created)
        let ssa_id = SsaId::new(SimplePseudonym::random(), 1.try_into()?);
        let spi = SsaPolynomialId::new(ssa_id, 0);

        // We need a valid-looking encrypted share even if it's junk.
        // EncryptedPartialSsaShare is basically a wrapper around bytes.
        let partial_share = PartialSsaShare::default().encrypt(&spi, &ack_key)?;

        let peer = OffchainKeypair::random();
        let nonce = crypto_traits::elliptic_curve::Scalar::<Secp256k1>::random(&mut hopr_types::crypto_random::rng());

        reconstructor.new_exit_commitment(ssa_id, test_params(DEFAULT_POLYS_PER_SSA, DEFAULT_POLY_THRESHOLD))?;

        reconstructor.insert_encrypted_share(
            peer.public(),
            challenge,
            TaggedEncryptedPartialSsaShare {
                pseudonym: *spi.pseudonym(),
                nonce,
                partial_share,
            },
        )?;

        let result = reconstructor.process_verified_ack(
            ack_key,
            challenge,
            reconstructor
                .awaiting_acks
                .get(peer.public())
                .as_ref()
                .ok_or(anyhow::anyhow!("missing peer"))?,
        );
        assert!(
            matches!(result, Ok(ProcessedAckResult::VerifierNotReady(reported)) if reported == spi),
            "an ack with no installed verifier must be deferred, not failed"
        );

        Ok(())
    }

    #[test]
    fn reconstructor_rejects_duplicate_share_via_different_challenges() -> anyhow::Result<()> {
        // 1 poly, threshold=2 → need 2 shares per polynomial to reconstruct.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 1,
            threshold: 2,
            surplus_shares: 0,
        });

        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, 1.try_into()?);

        let commitment_msg = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        let _server_commitment = reconstructor.new_exit_commitment(ssa_id, test_params(1, 2))?;
        commitment_msg.process_into_reconstructor(&reconstructor)?;

        // --- Step 1: Generate the first share ---
        let msg1: [u8; 20] = hopr_types::crypto_random::random_bytes();
        let Some(first) = generator.next_share(&pseudonym, &msg1)? else {
            anyhow::bail!("expected first share");
        };
        // Clone the PartialSsaShare so we can re-encrypt it as a duplicate later
        let first_share = first.share.clone();
        let ack1 = HalfKey::random();
        let challenge1 = ack1.to_challenge()?;
        let enc1 = first.share.encrypt(&first.id, &ack1)?;
        reconstructor.insert_encrypted_share(
            peer.public(),
            challenge1,
            TaggedEncryptedPartialSsaShare::new(pseudonym, &msg1, enc1)?,
        )?;

        // --- Step 2: Re-encrypt the SAME share under a different challenge (true duplicate) ---
        // The PartialSsaShare retains the same scalar value and derives the same identifier
        // (X-coordinate) from msg1, so it will be recognised as a duplicate at share-insertion time.
        let dup_ack = HalfKey::random();
        let dup_challenge = dup_ack.to_challenge()?;
        let enc_dup = first_share.encrypt(&first.id, &dup_ack)?;
        reconstructor.insert_encrypted_share(
            peer.public(),
            dup_challenge,
            TaggedEncryptedPartialSsaShare::new(pseudonym, &msg1, enc_dup)?,
        )?;

        // --- Step 3: Process the first ack — share accepted, not yet complete ---
        let resolution1 = reconstructor.process_verified_ack(
            ack1,
            challenge1,
            reconstructor
                .awaiting_acks
                .get(peer.public())
                .as_ref()
                .ok_or(anyhow::anyhow!("missing peer"))?,
        )?;
        assert!(
            matches!(resolution1, ProcessedAckResult::Progressed(p) if p.useful_shares == 1),
            "first share should count as progress without completing the SSA"
        );

        // --- Step 4: Process the duplicate ---
        // The SsaPartBuilder has 1/2 shares. The duplicate share has the same identifier
        // (same X-coordinate from msg1), so it hits the
        // `any(|s| s.identifier == share.identifier)` check in SsaPartBuilder::add_share.
        //
        // The point of the assertion is the contrast with step 3: a duplicate is not merely
        // "incomplete", it is *not progress*, so it must leave the counters where they were.
        let resolution_dup = reconstructor.process_verified_ack(
            dup_ack,
            dup_challenge,
            reconstructor
                .awaiting_acks
                .get(peer.public())
                .as_ref()
                .ok_or(anyhow::anyhow!("missing peer"))?,
        )?;
        assert!(
            matches!(resolution_dup, ProcessedAckResult::NoProgress),
            "duplicate share must not register as progress"
        );
        assert_eq!(
            1,
            reconstructor
                .cycle(&ssa_id)
                .ok_or(anyhow::anyhow!("cycle went away"))?
                .progress()
                .useful_shares,
            "the duplicate must not have advanced the useful-share count"
        );

        // --- Step 5: Generate and process the second distinct share ---
        let msg2: [u8; 20] = hopr_types::crypto_random::random_bytes();
        let Some(second) = generator.next_share(&pseudonym, &msg2)? else {
            anyhow::bail!("expected second share");
        };
        let ack2 = HalfKey::random();
        let challenge2 = ack2.to_challenge()?;
        let enc2 = second.share.encrypt(&second.id, &ack2)?;
        reconstructor.insert_encrypted_share(
            peer.public(),
            challenge2,
            TaggedEncryptedPartialSsaShare::new(pseudonym, &msg2, enc2)?,
        )?;

        let resolution2 = reconstructor.process_verified_ack(
            ack2,
            challenge2,
            reconstructor
                .awaiting_acks
                .get(peer.public())
                .as_ref()
                .ok_or(anyhow::anyhow!("missing peer"))?,
        )?;
        assert!(
            matches!(resolution2, ProcessedAckResult::FullRecovery(ref r, _) if r.ssa_id == ssa_id),
            "second unique share should complete SSA reconstruction"
        );

        Ok(())
    }

    #[test]
    fn reconstructor_must_not_accept_empty_encrypted_share() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig { ..Default::default() });

        let ack_key = HalfKey::random();
        let challenge = ack_key.to_challenge()?;

        let peer = OffchainKeypair::random();

        assert!(
            reconstructor
                .insert_encrypted_share(
                    peer.public(),
                    challenge,
                    TaggedEncryptedPartialSsaShare {
                        pseudonym: SimplePseudonym::random(),
                        nonce: Default::default(),
                        partial_share: Default::default(),
                    }
                )
                .is_err()
        );

        Ok(())
    }

    // -----------------------------------------------------------------------
    // early_recovery_threshold tests
    // -----------------------------------------------------------------------

    /// A non-finite early-recovery threshold must not survive configuration validation.
    ///
    /// The `range(min = 0.0, max = 1.0)` this replaced could not reject `NaN`: IEEE comparison against
    /// it is false in both directions, so it satisfied both bounds. And it did not stay harmless —
    /// `check_early_threshold` multiplies by `num_polys`, calls `ceil`, and casts to `usize`, and a
    /// `NaN` survives all three to saturate at zero. The Exit would then announce early recovery on
    /// its first reconstructed polynomial, which is the earliest instant it can ask for the next batch
    /// and precisely what the Entry's successor gate refuses.
    ///
    /// Asserted through `validate` rather than through `SsaReconstructor::new`, whose `expect` would
    /// make the failure a panic and the passing cases the only observable ones.
    #[test]
    fn a_non_finite_early_recovery_threshold_is_not_a_valid_configuration() {
        use validator::Validate;

        for bad in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY, 1.5, -0.1] {
            let cfg = SsaReconstructorConfig {
                early_recovery_threshold: bad,
                ..Default::default()
            };
            assert!(cfg.validate().is_err(), "{bad} is not a fraction and must be rejected");
        }

        for good in [0.0, 0.85, 1.0] {
            let cfg = SsaReconstructorConfig {
                early_recovery_threshold: good,
                ..Default::default()
            };
            assert!(cfg.validate().is_ok(), "{good} is a valid fraction");
        }
    }

    /// Helper: create an SsaBuilder that accepts zero-valued sub-secrets.
    ///
    /// These tests exercise part accumulation rather than progress reporting, so the recovery target
    /// only has to be self-consistent: one useful share per polynomial is the smallest value that
    /// keeps `SsaCycle::new`'s cross-check satisfiable for callers that go on to build a cycle.
    fn make_builder(num_polys: usize) -> SsaBuilder<TestSpec> {
        let exit_secret = PixScalar::<TestSpec>::default();
        let full_commitment = PixGroup::<TestSpec>::default();
        SsaBuilder::new(full_commitment, exit_secret, num_polys, num_polys as u64)
    }

    /// Helper: add `n` zero-valued polynomial parts to `builder`, returning
    /// the result of each call.
    fn add_parts(
        builder: &mut SsaBuilder<TestSpec>,
        n: usize,
    ) -> crate::errors::Result<Vec<Option<PixScalar<TestSpec>>>, <TestSpec as PixSpec>::Pseudonym> {
        let mut results = Vec::with_capacity(n);
        for i in 0..n {
            let sub = PixScalar::<TestSpec>::default();
            results.push(builder.add_recovered_ssa_part(i as PolynomialIndex, sub)?);
        }
        Ok(results)
    }

    /// A cycle with no polynomials can never recover, so the constructor refuses to build one.
    ///
    /// Unreachable through the commitment path — `PixParams::try_new` rejects `polys_per_ssa == 0` —
    /// but `SsaCycle::new` and `SsaBuilder::new` are both public, so this is a property of the type
    /// rather than of its current callers. It is also the case that must behave identically in debug
    /// and release: the guard predates the cross-check below it precisely because
    /// `debug_assert_eq!` compiles away and anything fallible inside it would not.
    #[test]
    fn a_cycle_with_no_polynomials_is_refused() {
        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);
        assert!(
            matches!(
                SsaCycle::new(ssa_id, make_builder(0), Vec::new()),
                Err(PixError::InvalidInput)
            ),
            "a zero-part cycle is unusable and must not be constructible"
        );
    }

    #[test]
    fn ssa_builder_early_threshold_below() -> anyhow::Result<()> {
        // num_polys=10, threshold=0.85 → ceil(0.85×10)=9.
        // Adding 8 parts should NOT reach the threshold.
        let mut builder = make_builder(10);
        add_parts(&mut builder, 8)?;
        assert!(!builder.check_early_threshold(0.85));
        Ok(())
    }

    #[test]
    fn ssa_builder_early_threshold_hits_ceil_at_9() -> anyhow::Result<()> {
        // num_polys=10, threshold=0.85 → ceil(0.85×10)=9.
        // Adding 9 parts SHOULD fire on the first check.
        let mut builder = make_builder(10);
        add_parts(&mut builder, 9)?;
        assert!(builder.check_early_threshold(0.85));
        // Second call must return false (idempotent guard).
        assert!(!builder.check_early_threshold(0.85));
        Ok(())
    }

    #[test]
    fn ssa_builder_threshold_1_dot_0_fires_at_full_recovery() -> anyhow::Result<()> {
        // num_polys=10, threshold=1.0 → ceil(1.0×10)=10.
        // Only fires when ALL 10 polynomial parts are received.
        let mut builder = make_builder(10);
        add_parts(&mut builder, 9)?;
        assert!(!builder.check_early_threshold(1.0));
        add_parts(&mut builder, 1)?; // 10th part → completes SSA
        // After full recovery, early_notified is set by add_recovered_ssa_part.
        // check_early_threshold should still report false.
        assert!(!builder.check_early_threshold(1.0));
        Ok(())
    }

    #[test]
    fn process_verified_ack_emits_early_and_full_recovery() -> anyhow::Result<()> {
        // Use a small SSA config where we can observe both events.
        // 4 polynomials, threshold=4, surplus=0 → 16 shares total.
        // early_recovery_threshold=0.5 → ceil(0.5×4)=2.
        // After 2 polynomial parts → EarlyRecovery.
        // After all 4         → FullRecovery.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 4,
            threshold: 4,
            surplus_shares: 0,
        });

        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, 1.try_into()?);

        let commitment_msg = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            early_recovery_threshold: 0.5,
            ..Default::default()
        });

        let _server_commitment = reconstructor.new_exit_commitment(ssa_id, test_params(4, 4))?;

        commitment_msg.process_into_reconstructor(&reconstructor)?;

        // No shares inserted yet — has_pending_shares must be false.
        assert!(
            !reconstructor.has_pending_shares(peer.public()),
            "no shares inserted yet"
        );

        // Generate and insert all 16 encrypted shares
        let mut acks = Vec::new();
        while let Some((msg, share)) = {
            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            generator.next_share(&pseudonym, &msg).map(|v| v.map(|u| (msg, u)))
        }? {
            let ack = HalfKey::random();
            let ack_challenge = ack.to_challenge()?;
            let enc_share = share.share.encrypt(&share.id, &ack)?;

            reconstructor.insert_encrypted_share(
                peer.public(),
                ack_challenge,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc_share)?,
            )?;
            acks.push(VerifiedAcknowledgement::new(ack, &peer).leak());
        }

        // After inserting encrypted shares, the peer must have pending shares.
        assert!(
            reconstructor.has_pending_shares(peer.public()),
            "shares were just inserted"
        );

        // Process all acks in one batch
        let resolutions = reconstructor.acknowledge_shares(*peer.public(), acks)?;

        // Both events MUST be present
        let has_early = resolutions
            .iter()
            .any(|r| matches!(r, ShareResolution::AlmostRecoveredSsa(id) if *id == ssa_id));
        let has_full = resolutions
            .iter()
            .any(|r| matches!(r, ShareResolution::RecoveredSsa(r) if r.ssa_id == ssa_id));

        assert!(has_early, "expected AlmostRecoveredSsa event");
        assert!(has_full, "expected RecoveredSsa event");

        Ok(())
    }

    /// Sets up one fully committed cycle and returns every share's acknowledgement, unprocessed.
    ///
    /// Shares are inserted but not acknowledged, so the caller can feed acks in whatever grouping
    /// the test needs and read the cycle's counters in between.
    #[allow(clippy::type_complexity)]
    fn cycle_with_pending_acks(
        polys: u16,
        threshold: u8,
        surplus: u8,
        peer: &OffchainKeypair,
    ) -> anyhow::Result<(SsaReconstructor<TestSpec>, SsaId<SimplePseudonym>, Vec<Acknowledgement>)> {
        cycle_with_declared_surplus(polys, threshold, surplus, surplus, peer)
    }

    /// As [`cycle_with_pending_acks`], but lets the Entry emit a different surplus from the one it
    /// negotiated.
    ///
    /// The two are equal for a conforming peer, which is why every other caller uses the wrapper
    /// above. Separating them is what makes an over-emitting Entry expressible: the Exit sizes its
    /// per-polynomial liveness credit on `declared_surplus`, and nothing but that bound stops the
    /// extra shares from counting.
    fn cycle_with_declared_surplus(
        polys: u16,
        threshold: u8,
        emitted_surplus: u8,
        declared_surplus: u8,
        peer: &OffchainKeypair,
    ) -> anyhow::Result<(SsaReconstructor<TestSpec>, SsaId<SimplePseudonym>, Vec<Acknowledgement>)> {
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: polys,
            threshold,
            surplus_shares: emitted_surplus,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let commitment_msg = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let reconstructor = SsaReconstructor::<TestSpec>::default();
        reconstructor.new_exit_commitment(
            ssa_id,
            PixParams::try_new_for::<TestSpec>(polys, threshold, declared_surplus)?,
        )?;
        commitment_msg.process_into_reconstructor(&reconstructor)?;

        let mut acks = Vec::new();
        while let Some((msg, share)) = {
            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            generator.next_share(&pseudonym, &msg).map(|v| v.map(|u| (msg, u)))
        }? {
            let ack = HalfKey::random();
            let ack_challenge = ack.to_challenge()?;
            let enc_share = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                ack_challenge,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc_share)?,
            )?;
            acks.push(VerifiedAcknowledgement::new(ack, peer).leak());
        }

        Ok((reconstructor, ssa_id, acks))
    }

    #[test]
    fn progress_counts_only_the_shares_that_advance_reconstruction() -> anyhow::Result<()> {
        // 2 polynomials × threshold 2, plus 1 surplus share each: 6 shares on the wire, of which
        // only 4 can ever be useful. The gap between those two numbers is the whole point — a
        // consumer sizing a progress ratio against packets received would read 6/4.
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;
        let peer = OffchainKeypair::random();
        let (reconstructor, ssa_id, acks) = cycle_with_pending_acks(POLYS, THRESHOLD, 1, &peer)?;
        assert_eq!(
            6,
            acks.len(),
            "generator should emit (threshold + surplus) per polynomial"
        );

        let mut snapshots = Vec::new();
        let mut recovered = false;
        // One ack at a time, so each share's individual contribution is observable.
        for ack in acks {
            for resolution in reconstructor.acknowledge_shares(*peer.public(), vec![ack])? {
                match resolution {
                    ShareResolution::Progress(p) => snapshots.push(p),
                    ShareResolution::RecoveredSsa(r) => {
                        assert_eq!(ssa_id, r.ssa_id);
                        recovered = true;
                    }
                    _ => {}
                }
            }
        }

        assert!(recovered, "the cycle must reconstruct from its own shares");

        // Shares are emitted polynomial-major, so the useful ones are shares 1,2 and 4,5 — the
        // third share of each polynomial arrives after that polynomial is already reconstructed.
        assert_eq!(
            vec![1, 2, 3, 4, 4, 4],
            snapshots.iter().map(|p| p.useful_shares).collect::<Vec<_>>(),
            "the first four snapshots advance reconstruction; the buffered tail reports liveness only"
        );
        let last = snapshots.last().ok_or(anyhow::anyhow!("no progress emitted"))?;
        assert_eq!(
            (POLYS as u64 * THRESHOLD as u64),
            last.target_useful_shares,
            "target must be polynomials × threshold, matching the negotiated dimensions"
        );
        assert_eq!(
            last.useful_shares, last.target_useful_shares,
            "a completed cycle must report itself as complete"
        );
        assert_eq!(
            POLYS, last.recovered_polynomials,
            "every polynomial must be accounted for"
        );
        assert_eq!(
            POLYS as u64 * (THRESHOLD as u64 + 1),
            last.shares_seen,
            "every negotiated share, including the post-recovery tail, must be visible as liveness"
        );

        Ok(())
    }

    /// A surplus share must report itself as liveness without reporting itself as progress.
    ///
    /// This is the distinction the Exit's egress gate and recovery-idle deadline turn on. A conforming
    /// Entry emits `threshold + surplus` shares per polynomial, and a whole emission window reaches its
    /// threshold in lockstep, so `surplus × window` consecutive shares advance nothing — 4096 packets
    /// at the deployed dimensions. A consumer told only about *useful* shares reads that run as the
    /// Entry having gone quiet, and if it answers by withholding service it removes the only thing that
    /// could have produced the next useful share.
    ///
    /// Acks are fed polynomial-major rather than in emission order, which is what puts a surplus share
    /// in front of a still-incomplete cycle. In emission order both surplus shares of a
    /// two-polynomial cycle arrive after full recovery has already removed it, which is why
    /// [`progress_counts_only_the_shares_that_advance_reconstruction`] never reaches this path.
    #[test]
    fn a_surplus_share_reports_liveness_but_not_progress() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;
        let peer = OffchainKeypair::random();
        let (reconstructor, ssa_id, acks) = cycle_with_pending_acks(POLYS, THRESHOLD, 1, &peer)?;
        assert_eq!(6, acks.len(), "(threshold + surplus) per polynomial");

        // Emission alternates between the two polynomials, so the even positions are one polynomial's
        // three shares and the odd positions the other's. Draining the evens first completes that
        // polynomial and then hands its third share to a cycle that is still short.
        let (evens, odds): (Vec<_>, Vec<_>) = acks.into_iter().enumerate().partition(|(i, _)| i % 2 == 0);
        let ordered = evens.into_iter().chain(odds).map(|(_, ack)| ack);

        let mut snapshots = Vec::new();
        let mut recovered = false;
        for ack in ordered {
            for resolution in reconstructor.acknowledge_shares(*peer.public(), vec![ack])? {
                match resolution {
                    ShareResolution::Progress(p) => snapshots.push((p.useful_shares, p.shares_seen)),
                    ShareResolution::RecoveredSsa(r) => {
                        assert_eq!(ssa_id, r.ssa_id);
                        recovered = true;
                    }
                    _ => {}
                }
            }
        }

        assert!(recovered, "the cycle must still reconstruct from its own shares");
        assert_eq!(
            vec![(1, 1), (2, 2), (2, 3), (3, 4), (4, 5), (4, 6)],
            snapshots,
            "(useful, seen): the third entry is the surplus share — seen advances, useful holds. The sixth share \
             arrives after full recovery removed the heavy cycle, and its compact tail still reports liveness."
        );

        Ok(())
    }

    /// A batch that ends on a surplus share reports that share's liveness, not the batch's last
    /// *useful* one.
    ///
    /// Every other test in this module feeds acknowledgements one at a time, so each batch carries a
    /// single snapshot and `record_progress` never has to merge anything. Batching is what makes the
    /// merge observable, and the merge is where a surplus run can be lost: the two counters are
    /// independent axes, and a selector keyed on `useful_shares` alone keeps the older snapshot for
    /// every share that advances only `shares_seen`.
    ///
    /// That is not a cosmetic loss. `shares_seen` is the axis the Session layer's egress gate and
    /// recovery-idle deadline both read, precisely so that the `surplus × window` run at the end of
    /// every emission window is not mistaken for the Entry falling silent. Dropping the surplus
    /// snapshot inside a batch reinstates exactly the reading those consumers were built to avoid.
    #[test]
    fn a_batch_ending_in_surplus_reports_the_surplus_share() -> anyhow::Result<()> {
        let peer = OffchainKeypair::random();
        let (reconstructor, _ssa_id, acks) = cycle_with_pending_acks(2, 2, 1, &peer)?;

        // Emission alternates polynomials, so the even positions are polynomial zero's three shares:
        // two useful, then one surplus. Handed over as a single batch, they collapse to one snapshot.
        let batch: Vec<_> = acks.into_iter().step_by(2).collect();
        assert_eq!(3, batch.len(), "threshold 2 plus 1 surplus for polynomial zero");

        let snapshots: Vec<_> = reconstructor
            .acknowledge_shares(*peer.public(), batch)?
            .into_iter()
            .filter_map(|r| match r {
                ShareResolution::Progress(p) => Some((p.useful_shares, p.shares_seen)),
                _ => None,
            })
            .collect();

        assert_eq!(
            vec![(2, 3)],
            snapshots,
            "(useful, seen): the batch's last share is surplus, so `seen` must reach 3 while `useful` holds at 2. \
             Reporting (2, 2) means the surplus snapshot was discarded for having an equal `useful_shares`, and the \
             Exit sees a batch it was served as one it was not."
        );

        Ok(())
    }

    /// A polynomial credits at most the negotiated surplus, however many shares arrive for it.
    ///
    /// Reconstruction releases the collected share set, so from that moment the builder cannot tell
    /// one post-completion share from another — every replay of a single share looks exactly like a
    /// fresh one. Without a bound, an Entry completes one polynomial of a large cycle, replays its
    /// shares forever, and every replay refreshes the Exit's egress gate and recovery-idle deadline
    /// while the remaining polynomials are never touched: unbounded service, no recovery, and so no
    /// payment. The bound is what makes that ride finite.
    ///
    /// The Entry here emits four surplus shares per polynomial having negotiated one, which is the
    /// only way to express over-emission — a conforming generator cannot exceed its own budget.
    #[test]
    fn a_polynomial_credits_no_more_surplus_than_was_negotiated() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        // Threshold and emitted surplus are equal because that is the largest over-emission
        // `SsaGeneratorConfig` will construct: `surplus_must_not_exceed_threshold` bounds the
        // surplus by the threshold, so expressing "emits four where it negotiated one" needs a
        // threshold of at least four.
        const THRESHOLD: u8 = 4;
        const DECLARED_SURPLUS: u8 = 1;
        const EMITTED_SURPLUS: u8 = 4;

        let peer = OffchainKeypair::random();
        let (reconstructor, _ssa_id, acks) =
            cycle_with_declared_surplus(POLYS, THRESHOLD, EMITTED_SURPLUS, DECLARED_SURPLUS, &peer)?;
        assert_eq!(16, acks.len(), "(threshold + emitted surplus) per polynomial");

        // Drain one polynomial's six shares before touching the other, so the cycle stays live while
        // that polynomial takes every surplus share the Entry has for it. In emission order the
        // second polynomial would complete the cycle first and retire it.
        let (evens, odds): (Vec<_>, Vec<_>) = acks.into_iter().enumerate().partition(|(i, _)| i % 2 == 0);

        let mut seen = Vec::new();
        for (_, ack) in evens {
            for resolution in reconstructor.acknowledge_shares(*peer.public(), vec![ack])? {
                if let ShareResolution::Progress(p) = resolution {
                    seen.push((p.useful_shares, p.shares_seen));
                }
            }
        }

        assert_eq!(
            vec![(1, 1), (2, 2), (3, 3), (4, 4), (4, 5)],
            seen,
            "(useful, seen): four useful shares complete the polynomial, then exactly one surplus share is credited. \
             The remaining three are over the negotiated budget and must report nothing at all — had they been \
             credited, `seen` would have run to 8."
        );

        // The other polynomial is untouched, so the cycle is still short of recovery — which is the
        // state the attack wants to hold the Exit in.
        let (_, odds_first) = odds.into_iter().next().expect("second polynomial has shares");
        assert!(
            reconstructor
                .acknowledge_shares(*peer.public(), vec![odds_first])?
                .iter()
                .any(|r| matches!(r, ShareResolution::Progress(_))),
            "the cycle must still be live and still accepting useful shares elsewhere"
        );

        Ok(())
    }

    /// Once a polynomial is reconstructed, the surplus path must not let malformed shares become
    /// liveness signals. Otherwise an Entry can keep the funded-service gate open indefinitely by
    /// naming a completed polynomial: `SsaPartBuilder::add_share` returns `Surplus` before even the
    /// inexpensive zero-share validation that applies while the polynomial is incomplete.
    #[test]
    fn an_invalid_share_for_a_completed_polynomial_is_not_liveness() -> anyhow::Result<()> {
        let peer = OffchainKeypair::random();
        // The negotiated surplus must be non-zero, and that is the whole reason this test can fail.
        // At a surplus of zero `max_credited_surplus` is zero too, so a share arriving for a
        // completed polynomial is refused because no per-polynomial credit remains, whichever side
        // of the `reconstructed` check the zero-coordinate validation sits on — and the assertion
        // below would hold even with the laundering channel wide open.
        let (reconstructor, ssa_id, acks) = cycle_with_pending_acks(2, 2, 1, &peer)?;

        // Emission alternates polynomials, so the even positions are polynomial zero's shares. Only
        // `threshold` of them are needed to complete it; taking exactly two leaves its surplus
        // credit unspent, which is the state an Entry would want to launder a malformed share
        // through. Polynomial one is never touched, so the cycle stays incomplete.
        for ack in acks.into_iter().step_by(2).take(2) {
            reconstructor.acknowledge_shares(*peer.public(), vec![ack])?;
        }

        let spi = SsaPolynomialId::new(ssa_id, 0);
        let ack = HalfKey::random();
        let challenge = ack.to_challenge()?;
        let encrypted = PartialSsaShare::<TestSpec>::default().encrypt(&spi, &ack)?;
        let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
        reconstructor.insert_encrypted_share(
            peer.public(),
            challenge,
            TaggedEncryptedPartialSsaShare::new(*spi.pseudonym(), &msg, encrypted)?,
        )?;

        let resolutions =
            reconstructor.acknowledge_shares(*peer.public(), vec![VerifiedAcknowledgement::new(ack, &peer).leak()])?;
        assert!(
            resolutions.iter().all(|r| !matches!(r, ShareResolution::Progress(_))),
            "a provably invalid zero share must not reset the service gate; got {resolutions:?}"
        );
        // Stated positively as well, so the test cannot pass by the share being dropped somewhere
        // upstream of the validation it exists to pin.
        assert!(
            resolutions
                .iter()
                .any(|r| matches!(r, ShareResolution::InvalidShares { ssa_id: id, .. } if *id == ssa_id)),
            "the share must be reported as a fault, not merely absorbed; got {resolutions:?}"
        );

        Ok(())
    }

    #[test]
    fn progress_precedes_the_terminal_event_it_belongs_to() -> anyhow::Result<()> {
        // The emission order is a contract: a consumer that acts on RecoveredSsa must already have
        // been told the counters that justify it, including for the batch that completes the cycle —
        // the one whose snapshot is taken from a cycle that no longer exists by the time the batch
        // returns.
        let peer = OffchainKeypair::random();
        let (reconstructor, ssa_id, acks) = cycle_with_pending_acks(2, 2, 0, &peer)?;

        let resolutions = reconstructor.acknowledge_shares(*peer.public(), acks)?;

        let first_terminal = resolutions
            .iter()
            .position(|r| {
                matches!(
                    r,
                    ShareResolution::RecoveredSsa(_) | ShareResolution::AlmostRecoveredSsa(_)
                )
            })
            .ok_or(anyhow::anyhow!("expected a terminal resolution"))?;
        let last_progress = resolutions
            .iter()
            .rposition(|r| matches!(r, ShareResolution::Progress(_)))
            .ok_or(anyhow::anyhow!("expected a progress resolution"))?;
        assert!(
            last_progress < first_terminal,
            "every Progress must precede every terminal event, got {resolutions:?}"
        );

        // Exactly one snapshot per SSA per batch, even though four shares moved the counters.
        let progress = resolutions
            .iter()
            .filter_map(|r| match r {
                ShareResolution::Progress(p) => Some(p),
                _ => None,
            })
            .collect::<Vec<_>>();
        assert_eq!(1, progress.len(), "one snapshot per SSA per batch, got {progress:?}");
        assert_eq!(ssa_id, progress[0].ssa_id);
        assert_eq!(
            4, progress[0].useful_shares,
            "the surviving snapshot must be the furthest-along one"
        );

        Ok(())
    }

    #[test]
    fn an_abandoned_commitment_guard_releases_its_registration() -> anyhow::Result<()> {
        let reconstructor = std::sync::Arc::new(SsaReconstructor::<TestSpec>::default());
        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);

        let (_commitment, guard) = reconstructor.new_guarded_exit_commitment(ssa_id, test_params(2, 2))?;
        assert_eq!(Some(&ssa_id), guard.ssa_id());
        assert!(
            reconstructor.contains_builder(&ssa_id),
            "the commitment must be registered while the guard holds it"
        );

        drop(guard);
        assert!(
            !reconstructor.contains_builder(&ssa_id),
            "dropping the guard must release the registration it owned"
        );

        Ok(())
    }

    /// Abandoning a guard whose cycle already went live escalates to a full retirement.
    ///
    /// Reaching this is a caller error — a live cycle means the peer was asked and answered, so
    /// ownership should already have been handed on with `disarm()`. Retiring is the safe response
    /// rather than the merely tidy one: the frontier is what stops a commitment completion racing
    /// the teardown from republishing the cycle that was just dismantled.
    #[test]
    fn abandoning_a_live_cycle_retires_it_rather_than_just_releasing_it() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });

        let reconstructor = std::sync::Arc::new(SsaReconstructor::<TestSpec>::default());
        let (_commitment, guard) = reconstructor.new_guarded_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;

        // Take the cycle live under the guard, which is the state a correct caller would have
        // disarmed out of.
        generator
            .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
            .process_into_reconstructor(reconstructor.as_ref())?;
        assert!(reconstructor.cycle(&ssa_id).is_some(), "the cycle must be live");

        drop(guard);
        assert!(
            reconstructor.cycle(&ssa_id).is_none(),
            "abandoning a live cycle must tear it down"
        );

        assert!(
            matches!(
                reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD)),
                Err(PixError::DuplicateCommitment)
            ),
            "the Session frontier must reject a retired SsaId before accepting replacement commitments"
        );

        Ok(())
    }

    #[test]
    fn a_disarmed_commitment_guard_leaves_its_ssa_alone() -> anyhow::Result<()> {
        let reconstructor = std::sync::Arc::new(SsaReconstructor::<TestSpec>::default());
        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);

        let (_commitment, guard) = reconstructor.new_guarded_exit_commitment(ssa_id, test_params(2, 2))?;
        assert_eq!(Some(ssa_id), guard.disarm());
        assert!(
            reconstructor.contains_builder(&ssa_id),
            "a disarmed guard must not retire the commitment it handed on"
        );

        Ok(())
    }

    /// Abandoning a request must leave its SSA index usable, because that index is what the next
    /// attempt will use — it is advanced only once every fallible step has succeeded.
    ///
    /// Releasing through the full retirement path instead advances the Session frontier and makes
    /// the retry fail as a duplicate before the peer can be asked to fund it.
    #[test]
    fn an_ssa_index_stays_usable_after_its_request_is_abandoned() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let reconstructor = std::sync::Arc::new(SsaReconstructor::<TestSpec>::default());

        // First attempt is abandoned before the peer is ever asked for commitments.
        let (_commitment, guard) = reconstructor.new_guarded_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;
        drop(guard);

        // Retry at the same index, which is what a failed request leaves behind.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let mut final_state = None;
        for (coeff_idx, coeffs) in commitment.verifiers.clone() {
            let proof = (coeff_idx == crate::CONSTANT_TERM_COEFFICIENT).then_some(commitment.commitment_proof);
            final_state =
                Some(reconstructor.insert_coefficient_commitments(ssa_id, coeff_idx, proof, coeffs.into_iter())?);
        }
        let state = final_state.ok_or(anyhow::anyhow!("commitment carried no batches"))?;

        assert!(
            state.is_verifiable,
            "the retry must become verifiable — otherwise the peer is asked to fund a cycle that cannot reconstruct"
        );
        assert!(state.ssa_deposit_address.is_some(), "a verifiable cycle has an address");
        assert!(
            reconstructor.cycle(&ssa_id).is_some(),
            "the retry's cycle must be live, not published and then withdrawn"
        );

        Ok(())
    }

    #[test]
    fn a_rejected_duplicate_registration_does_not_retire_the_original() -> anyhow::Result<()> {
        // The failure path must not produce a guard: if it did, dropping the error's guard would
        // retire the very registration whose presence caused the rejection — turning a harmless
        // duplicate request into the loss of a live cycle.
        let reconstructor = std::sync::Arc::new(SsaReconstructor::<TestSpec>::default());
        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);

        let (_commitment, guard) = reconstructor.new_guarded_exit_commitment(ssa_id, test_params(2, 2))?;

        assert!(
            matches!(
                reconstructor.new_guarded_exit_commitment(ssa_id, test_params(2, 2)),
                Err(PixError::DuplicateCommitment)
            ),
            "a second registration at the same index must be rejected"
        );
        assert!(
            reconstructor.contains_builder(&ssa_id),
            "the rejected duplicate must leave the original registration intact"
        );

        drop(guard);
        Ok(())
    }

    #[test]
    fn an_expired_registration_guard_does_not_release_its_same_index_retry() -> anyhow::Result<()> {
        let reconstructor = std::sync::Arc::new(SsaReconstructor::<TestSpec>::default());
        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);

        let (_commitment, stale_guard) = reconstructor.new_guarded_exit_commitment(ssa_id, test_params(2, 2))?;
        // Model the commitment cache's independent TTL expiry while its owner is still alive.
        reconstructor.commitment_builder.invalidate(&ssa_id);

        let (_commitment, retry_guard) = reconstructor.new_guarded_exit_commitment(ssa_id, test_params(2, 2))?;
        drop(stale_guard);

        assert!(
            reconstructor.contains_builder(&ssa_id),
            "a stale guard must not remove the replacement registration at the same index"
        );

        drop(retry_guard);
        Ok(())
    }

    #[test]
    fn a_live_cycle_stays_duplicate_after_its_commitment_registration_expires() -> anyhow::Result<()> {
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let params = test_params(2, 2);
        let reconstructor = std::sync::Arc::new(SsaReconstructor::<TestSpec>::default());
        let (_commitment, guard) = reconstructor.new_guarded_exit_commitment(ssa_id, params)?;

        SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        })
        .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
        .process_into_reconstructor(reconstructor.as_ref())?;
        reconstructor.commitment_builder.invalidate(&ssa_id);

        assert!(
            matches!(
                reconstructor.new_guarded_exit_commitment(ssa_id, params),
                Err(PixError::DuplicateCommitment)
            ),
            "a live cycle must continue occupying its index after the commitment cache entry expires"
        );

        drop(guard);
        Ok(())
    }

    /// **L2 regression.** A polynomial index repeated inside one batch must not consume its slot
    /// twice, and a repeat carrying a *different* commitment must be refused.
    ///
    /// The two-phase check tested each entry against `committed_polynomials`, which holds only what
    /// earlier calls inserted, so two entries sharing an index both found the slot vacant. The
    /// second insert then rebound the first — the single-assignment invariant the two phases exist
    /// to enforce — and `total_committed` counted two occupants of one slot. The practical bite:
    /// a batch carrying every polynomial with a repeat among them could never complete the set, and
    /// the peer had no way to supply the one it displaced.
    ///
    /// Both halves are checked here because they now take different paths: an identical repeat is
    /// *absorbed*, since retransmission legitimately produces one, while a conflicting one is still
    /// an error. Either way the slot is occupied exactly once, which is the invariant.
    ///
    /// The wire decoder rejects intra-message duplicates today. This builder is not meant to depend
    /// on that, which is why the check is here.
    #[test]
    fn a_polynomial_repeated_within_one_batch_occupies_its_slot_once() -> anyhow::Result<()> {
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;

        // Polynomial 0 twice with conflicting commitments: a contradiction, refused outright.
        let mut conflicting = coefficient_of(&commitment, 0, Some(0))?;
        conflicting.push((conflicting[0].0, PixGroup::<TestSpec>::generator().to_bytes()));
        let result =
            reconstructor.insert_coefficient_commitments(ssa_id, 0, proof_of(&commitment, 0), conflicting.into_iter());
        assert!(
            matches!(&result, Err(crate::errors::PixError::DuplicateCommitment)),
            "a conflicting repeat inside the batch must be rejected, got {result:?}"
        );

        // Rejected transactionally: neither entry was written, so polynomial 0 is still vacant. Now
        // the same index twice with the *same* commitment — accepted, occupying one slot, which is
        // the shape that would otherwise leave the set permanently one short while reporting two
        // commitments received.
        let mut repeated = coefficient_of(&commitment, 0, Some(0))?;
        repeated.push(repeated[0]);
        let partial =
            reconstructor.insert_coefficient_commitments(ssa_id, 0, proof_of(&commitment, 0), repeated.into_iter())?;
        assert!(
            !partial.is_verifiable,
            "two entries for one polynomial must not stand in for the other one"
        );

        // ...and the set completes on the polynomial the repeat did not cover.
        let retry = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(1))?.into_iter(),
        )?;
        assert!(
            retry.is_verifiable && retry.ssa_deposit_address.is_some(),
            "the remaining polynomial must complete the commitment"
        );

        Ok(())
    }

    #[test]
    fn malformed_commitment_does_not_poison_corrected_retransmission() -> anyhow::Result<()> {
        // Regression test for M2: a malformed coefficient that fails EC point
        // decoding must NOT leave the commitment builder permanently poisoned.
        // After submitting malformed bytes, a retry with correct bytes must
        // succeed and complete the SSA.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;

        // Step 1: Submit polynomial 0's constant term — must succeed, but the SSA commitment needs
        // every constant term, so no deposit address yet.
        let partial = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(0))?.into_iter(),
        )?;
        assert!(
            partial.ssa_deposit_address.is_none(),
            "the deposit address needs every constant term"
        );
        assert!(!partial.is_verifiable, "not yet complete");

        // Step 2: Submit a malformed constant term for polynomial 1 (bytes with an invalid EC
        // compressed-point prefix of 0xff) — must return InvalidInput.
        let mut malformed = PixGroupRepr::<TestSpec>::default(); // zero-filled
        AsMut::<[u8]>::as_mut(&mut malformed).fill(0xff);
        let malformed_result = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            [(1, malformed)].into_iter(),
        );
        assert!(
            matches!(&malformed_result, Err(crate::errors::PixError::InvalidInput)),
            "malformed commitment must be rejected, got {malformed_result:?}"
        );

        // Step 3: Retry with the correct bytes — must succeed and complete the SSA commitment.
        let retry = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(1))?.into_iter(),
        );
        assert!(
            matches!(&retry, Ok(state) if state.is_verifiable && state.ssa_deposit_address.is_some()),
            "corrected retransmission must complete the SSA, got {retry:?}"
        );

        Ok(())
    }

    /// Regression test for M13: the validation applied when a commitment *arrives* is the only
    /// validation it ever gets, so it must include the prime-order-subgroup test.
    ///
    /// A commitment that passes occupies its polynomial's slot permanently — re-insertion is
    /// rejected as a duplicate. A decodable-but-small-order point admitted here would take the
    /// slot and then make every reconstruction of that polynomial fail, with no way to retransmit
    /// a correction. This matters in practice: the default build uses BabyJubJub, whose cofactor is
    /// 8, so small-order points do exist and do pass a plain on-curve check.
    ///
    /// What this test covers is the *arrival point*, not the subgroup filter itself: `TestSpec` is
    /// secp256k1, cofactor 1, so no small-order point exists to feed it. The subgroup case is
    /// `pix_group_element_rejects_a_small_order_point` in `hopr-crypto-packet`, which also records
    /// why the filter cannot be isolated by any test — the Baby JubJub backend's own `from_bytes`
    /// already rejects, so `is_torsion_free` is defence in depth rather than the acting check.
    #[test]
    fn decode_commitment_is_the_single_validation_point() {
        use vsss_rs::elliptic_curve::group::GroupEncoding;

        use crate::SsaPartCommitment;

        // A well-formed commitment (the generator) decodes.
        let generator_repr = PixGroup::<TestSpec>::generator().to_bytes();
        assert!(
            SsaPartCommitment::<TestSpec>::decode_commitment(&generator_repr).is_ok(),
            "the generator is a valid constant-term commitment (scalar coefficient 1)"
        );

        // Undecodable bytes are rejected.
        let mut malformed = PixGroupRepr::<TestSpec>::default();
        AsMut::<[u8]>::as_mut(&mut malformed).fill(0xff);
        assert!(
            matches!(
                SsaPartCommitment::<TestSpec>::decode_commitment(&malformed),
                Err(PixError::InvalidInput)
            ),
            "undecodable bytes must be rejected"
        );
    }

    /// A reconstructed polynomial part must release its collected shares, while **keeping its
    /// cache entry**.
    ///
    /// That buffer dominates reconstructor memory — `threshold` shares held for every one of
    /// `polys` polynomials — and it cannot be read again once the part is reconstructed.
    ///
    /// Retaining the (now-stripped) cache entry is equally deliberate: evicting it would make every
    /// late or surplus share for that polynomial look like a not-yet-installed verifier, so the ack
    /// would be deferred into a bucket that nothing will ever drain. The stripped builder keeps the
    /// cheap already-reconstructed path instead.
    #[test]
    fn reconstructed_polynomial_releases_verification_state_but_keeps_its_entry() -> anyhow::Result<()> {
        // 2 polynomials, threshold 2, no surplus: finishing polynomial 0 does *not* complete the
        // SSA, so the cycle is still live and its per-polynomial state is observable.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let poly_0 = SsaPolynomialId::new(ssa_id, 0);

        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;
        commitment.process_into_reconstructor(&reconstructor)?;

        // Freshly installed: no shares collected yet.
        let cycle = reconstructor
            .cycle(&ssa_id)
            .ok_or_else(|| anyhow::anyhow!("cycle must be live"))?;
        assert_eq!(
            0,
            cycle
                .part(poly_0.poly_index())
                .ok_or_else(|| anyhow::anyhow!("part builder for polynomial 0 must be installed"))?
                .lock()
                .verification_state_len()
        );
        drop(cycle);

        // Feed exactly enough shares to reconstruct polynomial 0, skipping the ones the round-robin
        // emission hands out for polynomial 1 in between.
        for _ in 0..2 {
            let (msg, share) = next_share_for_poly(&generator, &pseudonym, 0)?;

            let ack = HalfKey::random();
            let challenge = ack.to_challenge()?;
            let enc = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                challenge,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
            )?;
            reconstructor.acknowledge_shares(*peer.public(), vec![VerifiedAcknowledgement::new(ack, &peer).leak()])?;
        }

        // The SSA as a whole is not recovered (polynomial 1 is untouched), so the cycle — and
        // polynomial 0's cache entry — must still be there.
        assert!(
            reconstructor.contains_builder(&ssa_id),
            "the cycle must still be live while polynomial 1 is outstanding"
        );
        let cycle = reconstructor
            .cycle(&ssa_id)
            .ok_or_else(|| anyhow::anyhow!("a reconstructed polynomial must keep its slot"))?;
        assert_eq!(
            0,
            cycle
                .part(poly_0.poly_index())
                .ok_or_else(|| anyhow::anyhow!("polynomial 0 must keep its slot"))?
                .lock()
                .verification_state_len(),
            "a reconstructed polynomial must hold no shares"
        );

        // Polynomial 1 is untouched and must still be installed, awaiting its own shares.
        assert!(
            cycle.part(1).is_some(),
            "part builder for polynomial 1 must be installed"
        );

        Ok(())
    }

    /// A failed interpolation must leave the part terminal, exactly as a failed commitment opening
    /// does.
    ///
    /// The two paths were asymmetric: the commitment mismatch released the share buffer and set
    /// `failed`, while a `combine()` error propagated with `?` and did neither. That left the part
    /// holding a full share set with no terminal flag, so neither early return in `add_share`
    /// fired and every remaining share for the polynomial was pushed and re-ran the interpolation
    /// over a larger set — `O(threshold²)` per share, against a buffer that should already have
    /// been released, re-reporting the same fault each time.
    ///
    /// The interpolation is forced to fail here by giving the builder a one-share threshold, which
    /// `vsss_rs` rejects outright. How the failure arises is not the property under test; that the
    /// part is left terminal either way is.
    #[test]
    fn a_failed_interpolation_leaves_the_part_terminal() -> anyhow::Result<()> {
        use utils::{AddShareOutcome, SsaPartBuilder};

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 1,
            threshold: 2,
            surplus_shares: 2,
        });
        let pseudonym = SimplePseudonym::random();
        let spi = SsaPolynomialId::new(SsaId::new(pseudonym, SsaIndex::MIN), 0);
        generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        // The commitment is never reached — `verify_reconstructed` runs only on a value that
        // interpolated — so the generator is a stand-in for any well-formed constant term.
        let mut part = SsaPartBuilder::<TestSpec>::new(
            crate::SsaPartCommitment::from_decoded_commitment(spi, PixGroup::<TestSpec>::generator()),
            1,
            0,
        );

        let mut rng = hopr_types::crypto_random::rng();
        let (_, share) = next_share_for_poly(&generator, &pseudonym, 0)?;
        assert!(
            part.add_share(PixScalar::<TestSpec>::random(&mut rng), share.share)
                .is_err(),
            "one share is below what `combine` accepts, so the interpolation must fail"
        );
        assert_eq!(
            0,
            part.verification_state_len(),
            "a part that can never reconstruct must release its share buffer"
        );

        // Every later share is absorbed instead of re-running the interpolation and re-reporting.
        let (_, share) = next_share_for_poly(&generator, &pseudonym, 0)?;
        assert!(
            matches!(
                part.add_share(PixScalar::<TestSpec>::random(&mut rng), share.share)?,
                AddShareOutcome::Absorbed
            ),
            "a failed part must absorb later shares silently"
        );
        assert_eq!(0, part.verification_state_len());

        Ok(())
    }

    #[test]
    fn full_recovery_retires_all_reconstructor_state() -> anyhow::Result<()> {
        // 4 polynomials, threshold 4, no surplus → 16 shares, fully recoverable.
        // Multiple polynomials on purpose: this exercises the whole `0..num_polys`
        // cleanup loop, so a wrong-key or off-by-one in `remove_cycle` would surface.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 4,
            threshold: 4,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let commitment_msg = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(4, 4))?;
        commitment_msg.process_into_reconstructor(&reconstructor)?;

        // Precondition: the completed cycle holds every part builder and its accumulator.
        assert_eq!(
            4,
            reconstructor.installed_parts(&ssa_id),
            "4 part builders present after completion"
        );
        assert_eq!(1, reconstructor.live_cycles(), "the cycle is live after completion");

        // Drive full recovery: generate every share, insert it encrypted, acknowledge.
        let mut acks = Vec::new();
        while let Some((msg, share)) = {
            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            generator.next_share(&pseudonym, &msg).map(|v| v.map(|u| (msg, u)))
        }? {
            let ack = HalfKey::random();
            let ack_challenge = ack.to_challenge()?;
            let enc_share = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                ack_challenge,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc_share)?,
            )?;
            acks.push(VerifiedAcknowledgement::new(ack, &peer).leak());
        }
        let resolutions = reconstructor.acknowledge_shares(*peer.public(), acks)?;
        assert!(
            resolutions
                .iter()
                .any(|r| matches!(r, ShareResolution::RecoveredSsa(r) if r.ssa_id == ssa_id)),
            "cycle must fully recover"
        );

        // Behaviour under test: full recovery must retire ALL of the cycle's
        // reconstructor state, rather than leave it to linger until the idle TTL.
        reconstructor.commitment_builder.run_pending_tasks();
        assert_eq!(
            0,
            reconstructor.live_cycles(),
            "the cycle must be retired on full recovery"
        );
        assert!(
            !reconstructor.commitment_builder.contains_key(&ssa_id),
            "commitment builder must be retired on full recovery"
        );

        Ok(())
    }

    #[test]
    fn retire_ssa_removes_cycle_state_and_is_idempotent() -> anyhow::Result<()> {
        // 3 polynomials so the cleanup loop is again exercised over several keys.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 3,
            threshold: 4,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let commitment_msg = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(3, 4))?;
        commitment_msg.process_into_reconstructor(&reconstructor)?;

        assert_eq!(
            3,
            reconstructor.installed_parts(&ssa_id),
            "3 part builders present after completion"
        );

        // Explicit retirement (as invoked on session teardown) drops everything.
        reconstructor.retire_ssa(ssa_id);
        reconstructor.commitment_builder.run_pending_tasks();
        assert_eq!(
            0,
            reconstructor.live_cycles(),
            "the cycle must be removed by retire_ssa"
        );
        assert!(!reconstructor.commitment_builder.contains_key(&ssa_id));

        // Idempotent: retiring the same (now-empty) cycle again is a harmless no-op.
        reconstructor.retire_ssa(ssa_id);

        // Retiring a cycle that was never created must not panic and must leave the caches
        // untouched.
        let never_seen = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);
        reconstructor.retire_ssa(never_seen);
        assert_eq!(0, reconstructor.live_cycles());

        Ok(())
    }

    #[test]
    fn retiring_a_later_cycle_does_not_block_an_earlier_registered_cycle() -> anyhow::Result<()> {
        let pseudonym = SimplePseudonym::random();
        let first_index = SsaIndex::new(1).context("one is a non-zero SSA index")?;
        let second_index = SsaIndex::new(2).context("two is a non-zero SSA index")?;
        let first = SsaId::new(pseudonym, first_index);
        let second = SsaId::new(pseudonym, second_index);
        let params = test_params(2, 2);
        let reconstructor = SsaReconstructor::<TestSpec>::default();

        reconstructor.new_exit_commitment(first, params)?;
        reconstructor.new_exit_commitment(second, params)?;
        reconstructor.retire_ssa(second);

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        });
        generator
            .new_ssa_commitment(&pseudonym, first_index)?
            .process_into_reconstructor(&reconstructor)?;

        assert!(
            reconstructor.cycle(&first).is_some(),
            "the registered lower index must win over the higher retirement watermark"
        );
        assert!(
            matches!(
                reconstructor.new_exit_commitment(second, params),
                Err(PixError::DuplicateCommitment)
            ),
            "the exact retired index must remain spent"
        );

        Ok(())
    }

    /// Deferred acknowledgements are bucketed by the cycle they are waiting for, sub-bucketed by
    /// polynomial, and never by peer.
    ///
    /// All three halves matter. Keying by cycle is what lets a bucket be drained by exactly one
    /// event (its own cycle installing) instead of being rescanned speculatively. The
    /// per-polynomial sub-bucket is what the cap is expressed against. And a sub-bucket
    /// deliberately holding several peers' acks is not an accident: one polynomial's shares are
    /// spread across return paths, hence across first-relayers, so the peer has to be carried per
    /// entry rather than being the key.
    #[test]
    fn deferred_acks_are_bucketed_by_cycle_and_polynomial_across_peers() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());

        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);
        let other_ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);
        let spi_0 = SsaPolynomialId::new(ssa_id, 0);
        let spi_1 = SsaPolynomialId::new(ssa_id, 1);
        let spi_other = SsaPolynomialId::new(other_ssa_id, 0);

        let peer_a = OffchainKeypair::random();
        let peer_b = OffchainKeypair::random();
        let ack_a = HalfKey::random();
        let ack_b = HalfKey::random();
        let ack_other_poly = HalfKey::random();
        let ack_other_cycle = HalfKey::random();

        // Two peers defer for the same polynomial; a third ack belongs to another polynomial of the
        // same cycle; a fourth to a different cycle entirely.
        reconstructor.defer_ack(spi_0, (*peer_a.public(), ack_a.to_challenge()?, ack_a));
        reconstructor.defer_ack(spi_0, (*peer_b.public(), ack_b.to_challenge()?, ack_b));
        reconstructor.defer_ack(
            spi_1,
            (*peer_a.public(), ack_other_poly.to_challenge()?, ack_other_poly),
        );
        reconstructor.defer_ack(
            spi_other,
            (*peer_a.public(), ack_other_cycle.to_challenge()?, ack_other_cycle),
        );

        let bucket = reconstructor
            .pending_acks
            .get(&ssa_id)
            .ok_or_else(|| anyhow::anyhow!("missing bucket for the cycle"))?;
        {
            let bucket = bucket.lock();
            assert_eq!(
                2,
                bucket.by_poly.get(&0).map(Vec::len).unwrap_or(0),
                "one sub-bucket holds both peers' acks"
            );
            assert_eq!(
                1,
                bucket.by_poly.get(&1).map(Vec::len).unwrap_or(0),
                "a different polynomial keeps its own sub-bucket"
            );
        }
        assert_eq!(3, reconstructor.deferred_ack_count(&ssa_id));
        assert_eq!(1, reconstructor.deferred_ack_count(&other_ssa_id));

        // Draining one cycle's bucket must not touch another's. No share exists in `awaiting_acks`,
        // so nothing is redeemed — the point is the bucket bookkeeping.
        reconstructor.drain_deferred_acks(&ssa_id);
        assert!(
            !reconstructor.pending_acks.contains_key(&ssa_id),
            "a drained bucket is removed, with all of its polynomials"
        );
        assert!(
            reconstructor.pending_acks.contains_key(&other_ssa_id),
            "draining one cycle must not disturb another"
        );

        Ok(())
    }

    /// Every polynomial becomes reconstructible on the call that completes the constant-term set,
    /// and none before it.
    ///
    /// A polynomial's whole commitment *is* its constant term, so there is no partially committed
    /// row to wait on; but a part builder is still useless until the [`SsaBuilder`] exists to
    /// receive what it reconstructs, and that needs the constant terms of *all* polynomials.
    /// The two therefore coincide, and this pins that they do.
    #[test]
    fn verifiers_are_installed_when_the_constant_term_set_completes() -> anyhow::Result<()> {
        const POLYS: u16 = 6;
        const THRESHOLD: u8 = 4;

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;

        // One polynomial's constant term per call. Nothing is published until the last one.
        for poly in 0..POLYS as PolynomialIndex {
            let state = reconstructor.insert_coefficient_commitments(
                ssa_id,
                0,
                proof_of(&commitment, 0),
                coefficient_of(&commitment, 0, Some(poly))?.into_iter(),
            )?;

            let last = poly == POLYS as PolynomialIndex - 1;
            assert_eq!(
                state.ssa_deposit_address.is_some(),
                last,
                "the deposit address is the sum of every constant term (poly {poly})"
            );
            assert_eq!(
                state.is_verifiable, last,
                "the cycle becomes verifiable exactly when the constant-term set closes (poly {poly})"
            );

            let expected = if last { POLYS as usize } else { 0 };
            assert_eq!(
                expected,
                reconstructor.installed_parts(&ssa_id),
                "after polynomial {poly} there must be {expected} part builders installed"
            );
        }

        Ok(())
    }

    /// An Entry that still emits the full Feldman matrix must not break the cycle: the Exit ignores
    /// every non-constant coefficient, whether it arrives before, during or after the constant-term
    /// pass, and recovery proceeds unaffected.
    #[test]
    fn non_constant_coefficients_are_ignored_wherever_they_arrive() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            early_recovery_threshold: 1.0,
            ..Default::default()
        });
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;

        // Stand-in for what a Feldman-emitting Entry would send: a well-formed commitment under a
        // non-constant coefficient index, for every polynomial.
        let higher: Vec<(PolynomialIndex, PixGroupRepr<TestSpec>)> = (0..POLYS as PolynomialIndex)
            .map(|poly| {
                (
                    poly,
                    PixGroup::<TestSpec>::mul_by_generator(&PixScalar::<TestSpec>::random(
                        &mut hopr_types::crypto_random::rng(),
                    ))
                    .to_bytes(),
                )
            })
            .collect();

        // Before the constant-term pass.
        reconstructor.insert_coefficient_commitments(ssa_id, 1, None, higher.clone().into_iter())?;

        // Interleaved with it.
        reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(0))?.into_iter(),
        )?;
        reconstructor.insert_coefficient_commitments(ssa_id, 1, None, higher.clone().into_iter())?;
        let state = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(1))?.into_iter(),
        )?;
        assert!(state.is_verifiable, "the constant-term pass alone completes the cycle");

        // And after it — the bulk of what such an Entry sends. Must not read as a duplicate.
        reconstructor.insert_coefficient_commitments(ssa_id, 1, None, higher.into_iter())?;

        // Recovery is unaffected.
        let mut acks = Vec::new();
        while let Some((msg, share)) = {
            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            generator.next_share(&pseudonym, &msg).map(|v| v.map(|u| (msg, u)))
        }? {
            let ack = HalfKey::random();
            let enc_share = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                ack.to_challenge()?,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc_share)?,
            )?;
            acks.push(VerifiedAcknowledgement::new(ack, &peer).leak());
        }

        let resolutions = reconstructor.acknowledge_shares(*peer.public(), acks)?;
        assert!(
            resolutions
                .iter()
                .any(|r| matches!(r, ShareResolution::RecoveredSsa(r) if r.ssa_id == ssa_id)),
            "the SSA must recover from the constant terms alone, got {resolutions:?}"
        );

        Ok(())
    }

    /// A single corrupted share is invisible until the polynomial is interpolated — that is the
    /// price of dropping the per-coefficient commitments — and it must surface exactly then, once,
    /// with no further noise from the shares that follow it.
    #[test]
    fn a_corrupted_share_is_reported_once_at_the_threshold_th_share() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 4;
        const SURPLUS: u8 = 2;

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: SURPLUS,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;
        generator
            .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
            .process_into_reconstructor(&reconstructor)?;

        // Corrupt the very first share of polynomial 0, then feed its whole budget one at a time.
        let mut invalid_reports: Vec<u64> = Vec::new();
        for i in 0..THRESHOLD as usize + SURPLUS as usize {
            let (msg, mut share) = next_share_for_poly(&generator, &pseudonym, 0)?;

            if i == 0 {
                // Flip a low bit of the share value. It stays a valid field element, so nothing
                // rejects it on arrival — only the interpolation notices.
                AsMut::<[u8]>::as_mut(&mut share.share.0)[31] ^= 1;
            }

            let ack = HalfKey::random();
            let enc = share.share.clone().encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                ack.to_challenge()?,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
            )?;
            let resolutions = reconstructor
                .acknowledge_shares(*peer.public(), vec![VerifiedAcknowledgement::new(ack, &peer).leak()])?;

            let reported = resolutions
                .iter()
                .filter_map(|r| match r {
                    ShareResolution::InvalidShares {
                        ssa_id: id,
                        observed_total,
                        ..
                    } if *id == ssa_id => Some(*observed_total),
                    _ => None,
                })
                .collect::<Vec<_>>();
            if i + 1 < THRESHOLD as usize {
                assert!(
                    reported.is_empty(),
                    "share {i} must pass unremarked — nothing checks it yet"
                );
            }
            invalid_reports.extend(reported);
        }

        assert_eq!(
            vec![1],
            invalid_reports,
            "the corrupted set must be reported exactly once, at the threshold-th share, as the cycle's first fault"
        );

        Ok(())
    }

    /// The fault total is a property of the cycle, not of whoever relayed the offending share.
    ///
    /// A Session's shares reach the Exit through whichever relayer the return path happens to use,
    /// so a per-peer total would let an Entry stay under any limit by spreading its bad shares —
    /// and would make the number the consumer enforces on depend on routing rather than on conduct.
    #[test]
    fn fault_totals_aggregate_over_the_cycle_not_the_peer() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        // One relayer per polynomial, which is what shares arriving over different return paths
        // looks like from here.
        let relayers = [OffchainKeypair::random(), OffchainKeypair::random()];

        let reconstructor = SsaReconstructor::<TestSpec>::default();
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;
        generator
            .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
            .process_into_reconstructor(&reconstructor)?;

        // Emission is round-robin across the window, and with no surplus every share matters — so
        // the cycle's whole budget is drawn first and grouped, rather than filtered as it comes.
        let mut by_poly: std::collections::BTreeMap<PolynomialIndex, Vec<_>> = Default::default();
        for _ in 0..POLYS as usize * THRESHOLD as usize {
            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            let share = generator
                .next_share(&pseudonym, &msg)?
                .ok_or_else(|| anyhow::anyhow!("generator must yield a share"))?;
            by_poly.entry(share.id.poly_index()).or_default().push((msg, share));
        }

        let mut totals = Vec::new();
        for poly in 0..POLYS {
            let relayer = &relayers[poly as usize];
            let shares = by_poly
                .remove(&poly)
                .ok_or_else(|| anyhow::anyhow!("no shares for polynomial {poly}"))?;
            assert_eq!(THRESHOLD as usize, shares.len());

            for (i, (msg, mut share)) in shares.into_iter().enumerate() {
                // Corrupt one share of each polynomial, so each one fails on its own threshold-th.
                if i == 0 {
                    AsMut::<[u8]>::as_mut(&mut share.share.0)[31] ^= 1;
                }

                let ack = HalfKey::random();
                let enc = share.share.clone().encrypt(&share.id, &ack)?;
                reconstructor.insert_encrypted_share(
                    relayer.public(),
                    ack.to_challenge()?,
                    TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
                )?;
                totals.extend(
                    reconstructor
                        .acknowledge_shares(
                            *relayer.public(),
                            vec![VerifiedAcknowledgement::new(ack, relayer).leak()],
                        )?
                        .into_iter()
                        .filter_map(|r| match r {
                            ShareResolution::InvalidShares { observed_total, .. } => Some(observed_total),
                            _ => None,
                        }),
                );
            }
        }

        assert_eq!(
            vec![1, 2],
            totals,
            "the second relayer's fault must be charged to the cycle's running total, not restart it"
        );
        assert_eq!(
            2,
            reconstructor
                .cycle(&ssa_id)
                .ok_or(anyhow::anyhow!("cycle went away"))?
                .invalid_shares(),
            "the cycle must hold the aggregate"
        );

        Ok(())
    }

    /// End-to-end deferral: shares that arrive before their polynomial's row is committed must be
    /// redeemed when the verifier installs, and the resulting resolutions must reach the caller.
    ///
    /// This is the path that used to be a full stash re-scan on every acknowledgement batch.
    #[test]
    fn shares_arriving_before_their_verifier_are_redeemed_on_installation() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;

        // Only part of the constant-term pass so far, so the SSA commitment is still unknown and no
        // part builder can be installed — a recovered part would have nowhere to go.
        reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(0))?.into_iter(),
        )?;

        // All shares for the whole cycle arrive now — every one of them ahead of its verifier.
        let mut acks = Vec::new();
        while let Some((msg, share)) = {
            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            generator.next_share(&pseudonym, &msg).map(|v| v.map(|u| (msg, u)))
        }? {
            let ack = HalfKey::random();
            let ack_challenge = ack.to_challenge()?;
            let enc_share = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                ack_challenge,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc_share)?,
            )?;
            acks.push(VerifiedAcknowledgement::new(ack, &peer).leak());
        }

        // Nothing can resolve yet, and every ack must have been bucketed rather than dropped.
        let resolutions = reconstructor.acknowledge_shares(*peer.public(), acks)?;
        assert!(
            resolutions.is_empty(),
            "no share can resolve before its polynomial is committed"
        );
        assert_eq!(
            (POLYS as usize * THRESHOLD as usize),
            reconstructor.deferred_ack_count(&ssa_id),
            "every early ack must be bucketed under its own cycle"
        );

        // The last constant term closes the set, which installs every part builder and redeems the
        // deferred acknowledgements.
        let state = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(1))?.into_iter(),
        )?;
        assert!(state.is_verifiable);

        // The bucket is consumed by the drain, not left to be re-scanned.
        assert!(
            !reconstructor.pending_acks.contains_key(&ssa_id),
            "the cycle's bucket must be consumed by the drain"
        );

        // The redeemed resolutions surface on the next acknowledgement batch. An empty ack batch is
        // enough — the work is already done, only the hand-off remains.
        let resolutions = reconstructor.acknowledge_shares(*peer.public(), Vec::new())?;
        assert!(
            resolutions
                .iter()
                .any(|r| matches!(r, ShareResolution::RecoveredSsa(r) if r.ssa_id == ssa_id)),
            "deferred shares must recover the SSA once their verifiers install, got {resolutions:?}"
        );

        Ok(())
    }

    /// Acknowledgements trailing a fully recovered cycle must use the compact tail, not be deferred.
    ///
    /// A missing cycle means one of two opposite things, and the acknowledgement path cannot tell
    /// them apart from the cache miss alone: a cycle not yet published, whose acks are worth
    /// bucketing, or one already retired. Advancing the Session frontier at recovery separates
    /// them, and the one recovered-tail slot preserves only the latter cycle's paid remainder.
    ///
    /// This is the common case rather than an edge: a conforming Entry emits `threshold + surplus`
    /// per polynomial, so the run of shares that advance nothing lands *after* the last polynomial
    /// completes — every recovered cycle is followed by acks in this state. Dropping their liveness
    /// makes the Session supervisor charge the FIFO drain to the successor; deferring them costs a
    /// bucket that can never be redeemed and holds the global acknowledgement budget.
    #[test]
    fn acks_trailing_a_recovered_cycle_report_liveness_without_deferral() -> anyhow::Result<()> {
        // A single polynomial, so its recovery *is* the SSA's and the surplus shares are
        // unambiguously post-recovery. With more, emission interleaves and which shares land after
        // the last completion stops being obvious from the dimensions.
        const POLYS: u16 = 1;
        const THRESHOLD: u8 = 2;
        const SURPLUS: u8 = 2;

        let peer = OffchainKeypair::random();
        let (reconstructor, ssa_id, acks) = cycle_with_pending_acks(POLYS, THRESHOLD, SURPLUS, &peer)?;
        assert_eq!(
            (THRESHOLD + SURPLUS) as usize,
            acks.len(),
            "the generator emits threshold + surplus for the single polynomial"
        );
        assert_eq!(
            acks.len(),
            reconstructor.count_ack_buffer_entries(),
            "every emitted share is awaiting its acknowledgement"
        );

        // One at a time: the two shares after the recovering one are the subject, and feeding the
        // batch whole would let a single dropped resolution hide which of them did what.
        let mut recovered_after = None;
        let mut tail_progress = Vec::new();
        for (i, ack) in acks.into_iter().enumerate() {
            let already_recovered = recovered_after.is_some();
            let mut recovered_now = false;
            for resolution in reconstructor.acknowledge_shares(*peer.public(), vec![ack])? {
                match resolution {
                    ShareResolution::Progress(progress) if already_recovered => {
                        tail_progress.push((progress.useful_shares, progress.shares_seen));
                    }
                    ShareResolution::RecoveredSsa(r) => {
                        assert_eq!(ssa_id, r.ssa_id);
                        recovered_after = Some(i + 1);
                        recovered_now = true;
                    }
                    _ => {}
                }
            }
            if recovered_now {
                // Mirrors supervisor tombstone cleanup. The compact Session tail deliberately
                // outlives that short diagnostic record; otherwise a buffer taking longer than the
                // retention window to drain would recreate the same liveness hole later.
                reconstructor.retire_ssa(ssa_id);
            }
        }
        assert_eq!(
            Some(THRESHOLD as usize),
            recovered_after,
            "the SSA must recover on the threshold-th share, leaving the surplus to trail it"
        );
        assert_eq!(
            vec![
                (THRESHOLD as u64, THRESHOLD as u64 + 1),
                (THRESHOLD as u64, THRESHOLD as u64 + 2)
            ],
            tail_progress,
            "each negotiated post-recovery share must advance only the liveness counter"
        );

        assert_eq!(
            0,
            reconstructor.deferred_ack_count(&ssa_id),
            "a retired cycle must not accumulate deferred acknowledgements"
        );
        assert!(
            !reconstructor.pending_acks.contains_key(&ssa_id),
            "no bucket may be created for a cycle that cannot come back"
        );
        // The load-bearing memory property: the tail consumes the encrypted entries immediately;
        // it does not retain the heavyweight cycle or park acknowledgements until their TTL.
        assert_eq!(
            0,
            reconstructor.count_ack_buffer_entries(),
            "the trailing shares must leave the acknowledgement buffer after bounded accounting"
        );

        Ok(())
    }

    /// A recovered tail must retain the live path's *per-polynomial* security boundary.
    ///
    /// A cycle-wide remainder is not equivalent. Once reconstruction releases duplicate state, an
    /// Entry can replay one completed polynomial; with an aggregate allowance, that one polynomial
    /// could spend the paid surplus belonging to every other polynomial and keep the Exit serving
    /// without ever presenting their buffered SURBs. Each polynomial therefore owns exactly its
    /// negotiated remainder before and after retirement.
    #[test]
    fn a_recovered_polynomial_cannot_borrow_another_polynomials_surplus_budget() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 4;
        const DECLARED_SURPLUS: u8 = 1;
        const EMITTED_SURPLUS: u8 = 4;

        let peer = OffchainKeypair::random();
        let (reconstructor, ssa_id, mut acks) =
            cycle_with_declared_surplus(POLYS, THRESHOLD, EMITTED_SURPLUS, DECLARED_SURPLUS, &peer)?;

        // Round-robin emission puts the first `polys × threshold` acknowledgements entirely in the
        // useful prefix. Its last acknowledgement recovers the cycle; every remaining one exercises
        // the compact tail rather than an `SsaPartBuilder`.
        let tail = acks.split_off(POLYS as usize * THRESHOLD as usize);
        let prefix = reconstructor.acknowledge_shares(*peer.public(), acks)?;
        assert!(
            prefix
                .iter()
                .any(|resolution| matches!(resolution, ShareResolution::RecoveredSsa(recovered) if recovered.ssa_id == ssa_id)),
            "the useful prefix must recover the cycle before the attack starts"
        );

        let (polynomial_zero, polynomial_one): (Vec<_>, Vec<_>) =
            tail.into_iter().enumerate().partition(|(index, _)| index % 2 == 0);
        let mut zero_snapshots = Vec::new();
        for (_, ack) in polynomial_zero {
            zero_snapshots.extend(
                reconstructor
                    .acknowledge_shares(*peer.public(), vec![ack])?
                    .into_iter()
                    .filter_map(|resolution| match resolution {
                        ShareResolution::Progress(progress) => Some(progress.shares_seen),
                        _ => None,
                    }),
            );
        }
        assert_eq!(
            vec![POLYS as u64 * THRESHOLD as u64 + 1],
            zero_snapshots,
            "replaying one recovered polynomial may consume only its own single negotiated slot"
        );

        let (_, other_ack) = polynomial_one.into_iter().next().expect("second polynomial has a tail");
        let other_progress = reconstructor
            .acknowledge_shares(*peer.public(), vec![other_ack])?
            .into_iter()
            .find_map(|resolution| match resolution {
                ShareResolution::Progress(progress) => Some(progress.shares_seen),
                _ => None,
            });
        assert_eq!(
            Some(POLYS as u64 * THRESHOLD as u64 + 2),
            other_progress,
            "the attack must not consume the independent allowance of the untouched polynomial"
        );

        Ok(())
    }

    /// Malformed shares cannot be laundered through a recovered polynomial as paid liveness.
    #[test]
    fn an_invalid_recovered_tail_share_is_neither_liveness_nor_budget_spend() -> anyhow::Result<()> {
        let peer = OffchainKeypair::random();
        let (reconstructor, ssa_id, mut acks) = cycle_with_pending_acks(1, 2, 1, &peer)?;
        let legitimate_tail = acks.pop().expect("one negotiated surplus acknowledgement");

        let prefix = reconstructor.acknowledge_shares(*peer.public(), acks)?;
        assert!(
            prefix
                .iter()
                .any(|resolution| matches!(resolution, ShareResolution::RecoveredSsa(recovered) if recovered.ssa_id == ssa_id)),
            "the threshold prefix must recover the cycle"
        );

        let spi = SsaPolynomialId::new(ssa_id, 0);
        let ack = HalfKey::random();
        let challenge = ack.to_challenge()?;
        let encrypted = PartialSsaShare::<TestSpec>::default().encrypt(&spi, &ack)?;
        let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
        reconstructor.insert_encrypted_share(
            peer.public(),
            challenge,
            TaggedEncryptedPartialSsaShare::new(*spi.pseudonym(), &msg, encrypted)?,
        )?;

        let invalid =
            reconstructor.acknowledge_shares(*peer.public(), vec![VerifiedAcknowledgement::new(ack, &peer).leak()])?;
        assert!(
            invalid
                .iter()
                .all(|resolution| !matches!(resolution, ShareResolution::Progress(_))),
            "a zero share addressed to the recovered tail must not reopen service: {invalid:?}"
        );
        assert!(
            invalid.iter().any(
                |resolution| matches!(resolution, ShareResolution::InvalidShares { ssa_id: id, .. } if *id == ssa_id)
            ),
            "the malformed share must reach validation and be reported: {invalid:?}"
        );

        let legitimate = reconstructor.acknowledge_shares(*peer.public(), vec![legitimate_tail])?;
        assert!(
            legitimate.iter().any(
                |resolution| matches!(resolution, ShareResolution::Progress(progress) if progress.shares_seen == 3)
            ),
            "rejecting malformed traffic must not consume the paid slot of a legitimate buffered SURB: {legitimate:?}"
        );

        Ok(())
    }

    /// A fault found while draining deferred acknowledgements must stay attributed to the relayer
    /// that carried the offending share.
    ///
    /// The drain runs on the commitment path, so its findings are handed to whichever
    /// `acknowledge_shares` call happens to collect them next — routinely a different peer's. Filling
    /// the relayer in at emission time would name that unrelated peer, which is worse than naming
    /// none: the field exists to attribute misbehaviour.
    #[test]
    fn a_fault_redeemed_from_deferral_keeps_its_own_relayer() -> anyhow::Result<()> {
        const POLYS: u16 = 1;
        const THRESHOLD: u8 = 2;

        // One surplus share, kept back to give `collector` an `awaiting_acks` entry.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 1,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        // `carrier` relays the shares; `collector` is an unrelated peer whose later batch picks up
        // the parked resolutions.
        let carrier = OffchainKeypair::random();
        let collector = OffchainKeypair::random();

        let reconstructor = SsaReconstructor::<TestSpec>::default();
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;

        // Shares arrive over `carrier` before any commitment, so they defer. One is corrupted, so the
        // drain is what discovers the fault.
        let mut acks = Vec::new();
        for i in 0..THRESHOLD {
            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            let mut share = generator
                .next_share(&pseudonym, &msg)?
                .ok_or_else(|| anyhow::anyhow!("generator must yield a share"))?;
            if i == 0 {
                AsMut::<[u8]>::as_mut(&mut share.share.0)[31] ^= 1;
            }
            let ack = HalfKey::random();
            let enc = share.share.clone().encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                carrier.public(),
                ack.to_challenge()?,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
            )?;
            acks.push(VerifiedAcknowledgement::new(ack, &carrier).leak());
        }
        assert!(
            reconstructor.acknowledge_shares(*carrier.public(), acks)?.is_empty(),
            "shares must defer while the commitment is unknown"
        );

        // Completing the commitment installs the part builder and drains the bucket, which is where
        // the corrupted set is detected.
        reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, None)?.into_iter(),
        )?;

        // The surplus share gives `collector` an entry in `awaiting_acks`, which is what makes its
        // batch acceptable. It is never acknowledged — the batch below is empty, so the only thing
        // `collector` contributes is being the wrong answer for the `peer` field.
        let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
        let surplus = generator
            .next_share(&pseudonym, &msg)?
            .ok_or_else(|| anyhow::anyhow!("generator must yield the surplus share"))?;
        let filler = HalfKey::random();
        reconstructor.insert_encrypted_share(
            collector.public(),
            filler.to_challenge()?,
            TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, surplus.share.encrypt(&surplus.id, &filler)?)?,
        )?;

        let faults = reconstructor
            .acknowledge_shares(*collector.public(), Vec::new())?
            .into_iter()
            .filter_map(|r| match r {
                ShareResolution::InvalidShares { peer, ssa_id, .. } => Some((peer, ssa_id)),
                _ => None,
            })
            .collect::<Vec<_>>();

        assert_eq!(1, faults.len(), "the drained fault must surface exactly once");
        assert_eq!(ssa_id, faults[0].1);
        assert_eq!(
            carrier.public(),
            faults[0].0.as_ref(),
            "the fault must name the relayer that carried the share, not the peer that collected it"
        );

        Ok(())
    }

    /// Deferring is decided on a cycle lookup that missed, so a cycle installing concurrently would
    /// leave the ack in a bucket whose one and only drain has already run — a share silently lost
    /// to a microsecond-wide window. `defer_ack` re-probes and drains itself in that case.
    #[test]
    fn deferring_against_an_installed_verifier_drains_immediately() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;
        generator
            .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
            .process_into_reconstructor(&reconstructor)?;

        // Every part builder is installed and every bucket already drained.
        let spi = SsaPolynomialId::new(ssa_id, 0);
        assert!(reconstructor.cycle(&ssa_id).is_some_and(|c| c.part(0).is_some()));

        // Simulate the racing path: an ack deferred *after* its cycle appeared.
        let ack = HalfKey::random();
        reconstructor.defer_ack(spi, (*peer.public(), ack.to_challenge()?, ack));

        assert!(
            !reconstructor.pending_acks.contains_key(&ssa_id),
            "a bucket created after its cycle installed must drain itself, not linger unclaimed"
        );

        Ok(())
    }

    /// The per-polynomial sub-cap must hold, so a peer cannot make the Exit buffer without bound by
    /// emitting more shares for one polynomial than its own share budget permits.
    #[test]
    fn deferred_ack_buckets_are_capped() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);
        let spi = SsaPolynomialId::new(ssa_id, 0);
        let peer = OffchainKeypair::random();

        for _ in 0..MAX_DEFERRED_ACKS_PER_POLYNOMIAL + 32 {
            let ack = HalfKey::random();
            reconstructor.defer_ack(spi, (*peer.public(), ack.to_challenge()?, ack));
        }

        assert_eq!(
            MAX_DEFERRED_ACKS_PER_POLYNOMIAL,
            reconstructor.deferred_ack_count(&ssa_id),
            "a polynomial's sub-bucket must not grow past the cap"
        );

        Ok(())
    }

    /// The per-cycle ceiling must hold even when the peer spreads its acknowledgements across many
    /// polynomials, each of which stays under the per-polynomial sub-cap.
    ///
    /// Without it the cycle total is `num_polys × 128` — a million entries at production
    /// dimensions, which is no bound at all.
    #[test]
    fn deferred_ack_buckets_are_capped_per_cycle() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);
        let peer = OffchainKeypair::random();

        // Spread over enough polynomials that no sub-bucket ever reaches its own cap, so the
        // per-cycle ceiling is provably the thing doing the work.
        let polys = (MAX_DEFERRED_ACKS_PER_CYCLE / (MAX_DEFERRED_ACKS_PER_POLYNOMIAL / 2)) + 8;
        'outer: for poly in 0..polys as PolynomialIndex {
            let spi = SsaPolynomialId::new(ssa_id, poly);
            for _ in 0..MAX_DEFERRED_ACKS_PER_POLYNOMIAL / 2 {
                let ack = HalfKey::random();
                reconstructor.defer_ack(spi, (*peer.public(), ack.to_challenge()?, ack));
                if reconstructor.deferred_ack_count(&ssa_id) > MAX_DEFERRED_ACKS_PER_CYCLE {
                    break 'outer;
                }
            }
        }

        assert_eq!(
            MAX_DEFERRED_ACKS_PER_CYCLE,
            reconstructor.deferred_ack_count(&ssa_id),
            "the cycle total must not grow past the ceiling"
        );

        Ok(())
    }

    /// An acknowledgement appended to a bucket the drain has already taken must still be redeemed.
    ///
    /// A bucket is reachable two ways — through the `pending_acks` key, and through an `Arc` a
    /// `defer_ack` obtained before the drain invalidated that key. The drain can only remove the
    /// first. The `ssa_cycles` re-probe was supposed to cover the resulting window, but it called
    /// `drain_deferred_acks`, which returns early on a cache miss — and a miss is exactly what this
    /// interleaving produces. The append landed in an orphaned bucket, and the share sat in
    /// `awaiting_acks` until `max_ack_await_time` discarded it. Silently: no error, no counter, and
    /// a polynomial losing more than `surplus_shares` this way strands the whole cycle without any
    /// check ever failing.
    ///
    /// Forcing the interleaving needs the stale handle, which is why `defer_ack_into` takes the
    /// bucket: after the invalidate, `defer_ack`'s own `get_with` would hand out a fresh one and
    /// the window would close by accident.
    #[test]
    fn an_acknowledgement_deferred_into_a_drained_bucket_is_still_redeemed() -> anyhow::Result<()> {
        // One polynomial at threshold 2, so the two deferred acknowledgements below are exactly
        // what the cycle needs: the second one landing makes the difference between a recovered
        // SSA and a stranded one.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 1,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(1, 2))?;

        // Both shares reach the Exit ahead of the commitment, so neither has a verifier yet.
        let mut pending = Vec::new();
        for _ in 0..2 {
            let (msg, share) = next_share_for_poly(&generator, &pseudonym, 0)?;
            let ack = HalfKey::random();
            let challenge = ack.to_challenge()?;
            let enc = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                challenge,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
            )?;
            pending.push((share.id, challenge, ack));
        }

        // The first acknowledgement takes the ordinary deferral path and creates the bucket.
        let (_, _, first_ack) = pending[0];
        reconstructor.acknowledge_shares(
            *peer.public(),
            vec![VerifiedAcknowledgement::new(first_ack, &peer).leak()],
        )?;
        let bucket = reconstructor
            .pending_acks
            .get(&ssa_id)
            .ok_or_else(|| anyhow::anyhow!("the first acknowledgement must have created a bucket"))?;

        // Installing the cycle drains that bucket and invalidates its key. Our handle survives.
        commitment.process_into_reconstructor(&reconstructor)?;
        assert!(
            reconstructor.pending_acks.get(&ssa_id).is_none(),
            "the drain must have taken the cache key"
        );

        // The racing append: a `defer_ack` that looked the bucket up before the drain ran.
        let (spi, challenge, ack) = pending[1];
        reconstructor.defer_ack_into(&bucket, spi, (*peer.public(), challenge, ack));

        // Two shares interpolate the only polynomial, so the SSA is recovered — but only if the
        // second acknowledgement was redeemed rather than parked in the orphan.
        assert!(
            reconstructor
                .take_ready_resolutions()
                .iter()
                .any(|resolution| matches!(resolution, ShareResolution::RecoveredSsa(_))),
            "the acknowledgement must have been redeemed inline, not lost with the bucket"
        );

        Ok(())
    }

    /// Drives a cycle to full recovery entirely through the deferral path, so the `RecoveredSsa` it
    /// produces ends up parked in `ready_resolutions` rather than returned from `acknowledge_shares`.
    ///
    /// Both shares arrive before the commitment does, which is the ordering the emission window
    /// makes routine near a cycle boundary, so both acknowledgements defer and the drain that
    /// installs the cycle is what reconstructs.
    fn park_a_recovered_ssa() -> anyhow::Result<(SsaReconstructor<TestSpec>, SsaId<SimplePseudonym>)> {
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 1,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(1, 2))?;

        for _ in 0..2 {
            let (msg, share) = next_share_for_poly(&generator, &pseudonym, 0)?;
            let ack = HalfKey::random();
            let challenge = ack.to_challenge()?;
            let enc = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                challenge,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
            )?;
            reconstructor.acknowledge_shares(*peer.public(), vec![VerifiedAcknowledgement::new(ack, &peer).leak()])?;
        }

        commitment.process_into_reconstructor(&reconstructor)?;
        assert_ne!(
            0,
            reconstructor
                .ready_resolutions_len
                .load(std::sync::atomic::Ordering::Acquire),
            "the drain must have parked the recovery"
        );

        Ok((reconstructor, ssa_id))
    }

    /// A parked resolution must not be gated behind the peer whose shares produced it.
    ///
    /// The pipelines call `has_pending_shares` to decide whether to hand a batch to
    /// `acknowledge_shares` at all, and `acknowledge_shares` is the only thing that ever collects
    /// `ready_resolutions`. Answering purely from `awaiting_acks` therefore made delivery of a
    /// recovered deposit key depend on the producing peer sending more traffic before its own cache
    /// entry idled out — while the buffer is global, and any batch could have carried it.
    #[test]
    fn a_parked_resolution_is_collectable_through_any_peer() -> anyhow::Result<()> {
        let (reconstructor, _) = park_a_recovered_ssa()?;

        // A peer this reconstructor has never seen: no shares, no `awaiting_acks` entry.
        let bystander = *OffchainKeypair::random().public();
        assert!(
            !reconstructor.awaiting_acks.contains_key(&bystander),
            "the bystander must have no pending shares of its own"
        );
        assert!(
            reconstructor.has_pending_shares(&bystander),
            "a parked resolution must let any batch through to collect it"
        );

        // And once collected, the guard goes back to answering per-peer.
        assert!(
            reconstructor
                .take_ready_resolutions()
                .iter()
                .any(|resolution| matches!(resolution, ShareResolution::RecoveredSsa(_)))
        );
        assert!(
            !reconstructor.has_pending_shares(&bystander),
            "with nothing parked the guard must not admit an unrelated peer"
        );

        Ok(())
    }

    /// Retiring a cycle must not consume the resolution it already produced.
    ///
    /// `ready_resolutions` is global and its entries name their own `SsaId`, so a parked
    /// `RecoveredSsa` stays collectable after its cycle is torn down — and it is worth collecting:
    /// the deposit key is what pays the Exit, whether or not the Session that earned it is still
    /// open. Draining at retirement would destroy that, and would take unrelated cycles'
    /// resolutions with it, since nothing in the buffer is keyed by cycle. The only point at which
    /// delivery genuinely becomes impossible is `Drop`, which reports whatever is left.
    #[test]
    fn retiring_a_cycle_leaves_its_resolution_collectable() -> anyhow::Result<()> {
        let (reconstructor, ssa_id) = park_a_recovered_ssa()?;

        reconstructor.retire_ssa(ssa_id);

        assert!(
            reconstructor
                .take_ready_resolutions()
                .iter()
                .any(|resolution| matches!(resolution, ShareResolution::RecoveredSsa(_))),
            "retirement must not swallow a recovered deposit key"
        );

        Ok(())
    }

    /// **H8 regression.** Reclamation is scoped to the cycle, so a share for *any* polynomial keeps
    /// the whole cycle alive.
    ///
    /// This used to fail. The part builders were keyed per polynomial with an idle timer, so the
    /// clock measured "time since a share for *this* polynomial arrived". Commitments are a
    /// fraction of a percent of a cycle's bytes, so every builder was installed in the opening
    /// moments and then waited, while shares arrive polynomial-major and spread across the whole
    /// cycle. Any polynomial late in the emission order had its builder evicted before its first
    /// share landed — unrecoverably, since the commitment cannot be retransmitted and a deferred
    /// ack's only drain is an installation that had already happened. The SSA never completed and
    /// the deposit burned.
    ///
    /// The condition was `quota / line_rate > unused_verifier_lifetime`. At the deployed 1.5 Mbps
    /// per-Session cap a 519 MiB cycle runs 48.4 minutes against a 30-minute default, so every
    /// polynomial past ≈62 % of the cycle was lost.
    ///
    /// Scaled down here to two polynomials and a half-second lifetime; the shape is identical.
    ///
    /// The essential geometry is that the cycle is *continuously* busy while any single polynomial
    /// is not. Shares are spaced at half the lifetime, so the cycle never goes idle, but the four
    /// shares of polynomial 0 take twice the lifetime to arrive — long enough that polynomial 1's
    /// builder, untouched since installation, would have expired under the old per-polynomial key.
    #[test]
    fn a_cycle_stays_live_while_a_single_polynomial_goes_untouched() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 4;
        /// Bounded on **both** sides, which is why this is not simply "as large as possible":
        ///
        /// * above `SHARE_SPACING`, or the cycle idles out between two consecutive shares and the test fails for a
        ///   reason unrelated to H8. Each iteration does rather more than sleep — polynomial evaluation, encryption,
        ///   insertion, and at the threshold a Lagrange combine and a scalar multiplication — so on a contended runner
        ///   the margin needs to be several times the sleep, not the 2× it used to be;
        /// * below `(THRESHOLD + 1) × SHARE_SPACING` ≈ 1250 ms, or the assertion below stops holding and the test
        ///   exercises nothing.
        ///
        /// 1000 ms sits inside that window with the slack on the side that grows under load:
        /// cumulative elapsed time only ever overshoots, while a single iteration would have to
        /// take four times its sleep to expire the cycle.
        const VERIFIER_LIFETIME: std::time::Duration = std::time::Duration::from_millis(1000);
        /// Comfortably inside the lifetime, so no *cycle* is ever idle long enough to expire.
        const SHARE_SPACING: std::time::Duration = std::time::Duration::from_millis(250);

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            unused_verifier_lifetime: VERIFIER_LIFETIME,
            ..Default::default()
        });
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;

        // The whole commitment set lands up front, as it does in production: every part builder is
        // installed now, and the later ones then wait out most of the cycle.
        generator
            .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
            .process_into_reconstructor(&reconstructor)?;

        let installed_at = std::time::Instant::now();
        let mut recovered = false;
        for i in 0..(POLYS as usize * THRESHOLD as usize) {
            std::thread::sleep(SHARE_SPACING);

            // Shares are emitted polynomial-major, so this is the hand-over to polynomial 1 — the
            // point at which its builder has been idle for the whole of polynomial 0's run.
            if i == THRESHOLD as usize {
                assert!(
                    installed_at.elapsed() > VERIFIER_LIFETIME,
                    "the test must actually outlast the lifetime before polynomial 1's first share, otherwise it \
                     exercises nothing"
                );
            }

            let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
            let share = generator
                .next_share(&pseudonym, &msg)?
                .ok_or_else(|| anyhow::anyhow!("generator must yield a share"))?;

            let ack = HalfKey::random();
            let enc = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                ack.to_challenge()?,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
            )?;
            recovered |= reconstructor
                .acknowledge_shares(*peer.public(), vec![VerifiedAcknowledgement::new(ack, &peer).leak()])?
                .iter()
                .any(|r| matches!(r, ShareResolution::RecoveredSsa(r) if r.ssa_id == ssa_id));
        }

        if !recovered {
            // Name the mechanism, so a regression reports the cause and not just the symptom. Under
            // H8 the late polynomial's builder was gone and its shares were stranded in a bucket
            // whose only drain had already run.
            assert_eq!(
                0,
                reconstructor.deferred_ack_count(&ssa_id),
                "shares were stranded in a deferred-ack bucket — H8 has regressed"
            );
            panic!("the cycle failed to recover even though it was continuously busy");
        }

        Ok(())
    }

    /// The idle timer was not simply disabled: a cycle that receives no shares at all is still
    /// reclaimed on schedule.
    ///
    /// This is the other half of the H8 regression. Widening the reclamation scope is only correct
    /// if reclamation still happens — otherwise an abandoned cycle is pinned until session
    /// teardown.
    #[test]
    fn a_cycle_with_no_shares_at_all_still_expires() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;
        const VERIFIER_LIFETIME: std::time::Duration = std::time::Duration::from_millis(500);

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig {
            unused_verifier_lifetime: VERIFIER_LIFETIME,
            ..Default::default()
        });
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;
        generator
            .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
            .process_into_reconstructor(&reconstructor)?;

        assert_eq!(POLYS as usize, reconstructor.installed_parts(&ssa_id));

        std::thread::sleep(VERIFIER_LIFETIME * 3);

        assert_eq!(
            0,
            reconstructor.live_cycles(),
            "a cycle that never received a share must still be reclaimed"
        );

        Ok(())
    }

    /// The polynomial index travels inside a peer-supplied share, so it reaches the slot lookup as
    /// untrusted input. Once the cycle is known its dimensions are too, which makes an out-of-range
    /// index definitively invalid rather than merely early — and it must not index the slot array.
    #[test]
    fn a_share_naming_a_polynomial_outside_the_cycle_is_rejected() -> anyhow::Result<()> {
        const POLYS: u16 = 2;
        const THRESHOLD: u8 = 2;

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;
        generator
            .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
            .process_into_reconstructor(&reconstructor)?;

        // Take a real share and re-label it for a polynomial the cycle does not have.
        let msg: [u8; 20] = hopr_types::crypto_random::random_bytes();
        let mut share = generator
            .next_share(&pseudonym, &msg)?
            .ok_or_else(|| anyhow::anyhow!("generator must yield a share"))?;
        share.id = SsaPolynomialId::new(ssa_id, POLYS as PolynomialIndex + 5);

        let ack = HalfKey::random();
        let enc = share.share.encrypt(&share.id, &ack)?;
        reconstructor.insert_encrypted_share(
            peer.public(),
            ack.to_challenge()?,
            TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
        )?;

        // `acknowledge_shares` logs and swallows the error, so the observable contract is that
        // nothing resolves, nothing is deferred, and the process is still standing.
        let resolutions =
            reconstructor.acknowledge_shares(*peer.public(), vec![VerifiedAcknowledgement::new(ack, &peer).leak()])?;
        assert!(resolutions.is_empty(), "an out-of-range share must resolve to nothing");
        assert_eq!(
            0,
            reconstructor.deferred_ack_count(&ssa_id),
            "an out-of-range share must be rejected outright, not deferred forever"
        );
        assert!(
            reconstructor.cycle(&ssa_id).is_some(),
            "the cycle itself must be unharmed"
        );

        Ok(())
    }

    /// Shares for different polynomials of one cycle must be able to run concurrently.
    ///
    /// The cycle is a single cache entry, which makes collapsing it to a single mutex an easy and
    /// invisible mistake — and one that would serialise every share of a Session. This holds one
    /// polynomial's lock and asserts another's is still free.
    #[test]
    fn parts_of_one_cycle_lock_independently() -> anyhow::Result<()> {
        const POLYS: u16 = 4;
        const THRESHOLD: u8 = 2;

        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: POLYS,
            threshold: THRESHOLD,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;
        generator
            .new_ssa_commitment(&pseudonym, SsaIndex::MIN)?
            .process_into_reconstructor(&reconstructor)?;

        let cycle = reconstructor
            .cycle(&ssa_id)
            .ok_or_else(|| anyhow::anyhow!("cycle must be live"))?;

        let held = cycle.part(0).ok_or_else(|| anyhow::anyhow!("missing part 0"))?.lock();
        for poly in 1..POLYS as PolynomialIndex {
            assert!(
                cycle
                    .part(poly)
                    .ok_or_else(|| anyhow::anyhow!("missing part {poly}"))?
                    .try_lock()
                    .is_some(),
                "polynomial {poly} must not be blocked by polynomial 0 — the cycle must not share one mutex"
            );
        }
        // The accumulator is a separate lock too, so a part in flight does not block recovery
        // accounting for another part.
        assert!(
            cycle.builder().try_lock().is_some(),
            "the accumulator must lock separately"
        );
        drop(held);

        Ok(())
    }

    /// Verifies that the builder caches accept more entries than the old
    /// `MAX_POLYS_PER_SSA` size bound. After removing the hard capacity, only
    /// TTL governs eviction.  Also verifies that fully-committed IDs populate
    /// `ssa_builders` and remain cached.
    #[test]
    fn builder_caches_accept_more_entries_than_max_polys_per_ssa() -> anyhow::Result<()> {
        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        let pseudonym = SimplePseudonym::random();
        let exceed = MAX_POLYS_PER_SSA as usize + 5;
        let mut ids = Vec::with_capacity(exceed);
        for i in 0..exceed {
            let ssa_id = SsaId::new(pseudonym, (1u32 + i as u32).try_into()?);
            reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;
            ids.push(ssa_id);
        }
        reconstructor.commitment_builder.run_pending_tasks();
        for ssa_id in &ids {
            assert!(
                reconstructor.contains_builder(ssa_id),
                "commitment builder must retain every accepted SsaId ({ssa_id:?})"
            );
        }

        // Complete the first few IDs through the full commitment path to populate `ssa_cycles`.
        // Use a 2-poly 2-threshold generator so that insert_coefficient_commitments reaches the
        // completion milestone and publishes a cycle.
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 1,
        });
        for ssa_id in ids.iter().take(3) {
            let commit = generator.new_ssa_commitment(&pseudonym, ssa_id.ssa_index())?;
            commit.process_into_reconstructor(&reconstructor)?;
            reconstructor.commitment_builder.run_pending_tasks();

            assert_eq!(
                2,
                reconstructor.installed_parts(ssa_id),
                "ssa_cycles must contain completed SsaId index {} ({ssa_id:?})",
                ssa_id.ssa_index().get(),
            );
        }

        Ok(())
    }

    /// Extracts one coefficient's commitments from a generated SSA commitment, optionally narrowed
    /// to a single polynomial, in the shape `insert_coefficient_commitments` expects.
    fn coefficient_of(
        commitment: &crate::SsaCommitment<TestSpec>,
        coeff_index: CoefficientIndex,
        only_poly: Option<PolynomialIndex>,
    ) -> anyhow::Result<Vec<(PolynomialIndex, PixGroupRepr<TestSpec>)>> {
        Ok(commitment
            .verifiers
            .get(&coeff_index)
            .ok_or_else(|| anyhow::anyhow!("missing coefficient {coeff_index}"))?
            .iter()
            .filter(|(poly_index, _)| only_poly.is_none_or(|wanted| *poly_index == wanted))
            .map(|(poly_index, repr)| (*poly_index, *repr))
            .collect())
    }

    /// Retirement-frontier guard on the *first* publication point: the SSA becoming live.
    ///
    /// Since verifiers are now installed as individual polynomials complete, the part accumulator
    /// has to be published earlier — the moment the constant terms yield the SSA commitment. A
    /// `retire_ssa` racing that publication must still leave nothing behind.
    #[test]
    fn retire_ssa_frontier_prevents_builder_publication() -> anyhow::Result<()> {
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;

        // Constant term of polynomial 0 only — the SSA commitment is still unknown.
        let state = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(0))?.into_iter(),
        )?;
        assert!(
            state.ssa_deposit_address.is_none(),
            "deposit address must not be derivable from a partial constant-term set"
        );

        // Simulate `retire_ssa` winning the frontier race but pausing before cache removal.
        let registration = reconstructor
            .commitment_builder
            .get(&ssa_id)
            .context("registered commitment must exist")?;
        assert!(registration.retirement.lock().retire(ssa_id.ssa_index()));

        // Constant term of polynomial 1 completes the set, so this call would publish the builder.
        let state = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(1))?.into_iter(),
        )?;
        assert!(!state.is_verifiable, "a retired cycle is never verifiable");

        assert_eq!(
            0,
            reconstructor.live_cycles(),
            "retirement frontier must prevent cycle publication"
        );

        Ok(())
    }

    /// Retirement-frontier guard on verifier installation.
    ///
    /// Verifiers and the part accumulator are now published by the same call, so retirement racing
    /// it must withdraw both. Publication and retirement serialize on the Session frontier, so a
    /// retirement that wins the lock must prevent the installation altogether.
    #[test]
    fn retire_ssa_frontier_prevents_verifier_installation() -> anyhow::Result<()> {
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;

        // Polynomial 0's constant term: nothing is published yet, the set is incomplete.
        reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(0))?.into_iter(),
        )?;
        assert_eq!(
            0,
            reconstructor.live_cycles(),
            "no part builder may be installed while the SSA commitment is unknown"
        );

        // Retirement lands here but pauses before cache removal.
        let registration = reconstructor
            .commitment_builder
            .get(&ssa_id)
            .context("registered commitment must exist")?;
        assert!(registration.retirement.lock().retire(ssa_id.ssa_index()));

        // Polynomial 1's constant term closes the set, so this call would install both verifiers.
        let state = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            proof_of(&commitment, 0),
            coefficient_of(&commitment, 0, Some(1))?.into_iter(),
        )?;
        assert!(!state.is_verifiable, "a retired cycle is never verifiable");

        assert_eq!(
            0,
            reconstructor.live_cycles(),
            "retirement frontier must prevent installation of the accumulator and every part builder"
        );

        Ok(())
    }

    /// The commitment proof must bind everything it claims to: both of its own components, the SSA
    /// index it was issued for, and the commitment it opens.
    #[test]
    fn commitment_proof_must_bind_its_components_the_ssa_index_and_the_commitment() -> anyhow::Result<()> {
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let proof = commitment.commitment_proof;

        assert!(
            proof.verify(&ssa_id, &commitment.ssa_commitment),
            "the generator's own proof must verify"
        );

        // Flipping a bit anywhere breaks it, whether it lands in the nonce commitment or in the
        // response. Some flips make the component unparseable, which is equally a rejection.
        let bytes = proof.to_bytes();
        assert_eq!(SsaCommitmentProof::<TestSpec>::SIZE, bytes.len());
        for position in [0, bytes.len() / 2, bytes.len() - 1] {
            let mut tampered = bytes.clone();
            tampered[position] ^= 1;
            if let Ok(tampered) = SsaCommitmentProof::<TestSpec>::try_from_bytes(&tampered) {
                assert!(
                    !tampered.verify(&ssa_id, &commitment.ssa_commitment),
                    "a proof with byte {position} flipped must not verify"
                );
            }
        }

        // Bound to the SSA index, so it cannot be replayed onto another cycle even if the commitment
        // were somehow reused.
        let other_index = SsaId::new(pseudonym, SsaIndex::new(SsaIndex::MIN.get() + 1).expect("non-zero"));
        assert!(
            !proof.verify(&other_index, &commitment.ssa_commitment),
            "a proof must not verify against a different SSA index"
        );

        // And bound to the commitment: this is the property the whole thing exists for.
        let unrelated = PixGroup::<TestSpec>::mul_by_generator(&PixScalar::<TestSpec>::random(
            &mut hopr_types::crypto_random::rng(),
        ));
        assert!(
            !proof.verify(&ssa_id, &unrelated),
            "a proof must not verify against a commitment it does not open"
        );

        // A truncated or over-long encoding is refused outright.
        assert!(SsaCommitmentProof::<TestSpec>::try_from_bytes(&bytes[..bytes.len() - 1]).is_err());
        assert!(SsaCommitmentProof::<TestSpec>::try_from_bytes(&[bytes.as_slice(), &[0u8]].concat()).is_err());

        Ok(())
    }

    /// A constant-term set that carries no proof at all is refused, exactly as one carrying an
    /// invalid proof is: the Exit cannot tell the difference, and both mean the cycle is unusable.
    #[test]
    fn constant_terms_arriving_without_a_proof_are_refused() -> anyhow::Result<()> {
        let generator = SsaShareGenerator::<TestSpec>::new(SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        });
        let pseudonym = SimplePseudonym::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);
        let commitment = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;

        let reconstructor = SsaReconstructor::<TestSpec>::new(Default::default());
        reconstructor.new_exit_commitment(ssa_id, test_params(2, 2))?;

        let refused = reconstructor.insert_coefficient_commitments(
            ssa_id,
            0,
            None,
            coefficient_of(&commitment, 0, None)?.into_iter(),
        );
        assert!(
            matches!(refused, Err(PixError::UnprovenSsaCommitment)),
            "constant terms without a proof must be refused, got {refused:?}"
        );
        assert_eq!(0, reconstructor.live_cycles(), "no cycle may be published");

        Ok(())
    }

    /// A client commitment crafted so the Entry alone knows the *combined* deposit key must be
    /// refused, and must not produce a deposit address.
    ///
    /// The deposit key is `s + e`, where `s` is the sum of the Entry's polynomial constant terms and
    /// `e` is the Exit's commitment secret. Neither party is supposed to know the sum. But
    /// `SsaRequest` hands `e·G` to the Entry *before* it chooses its own constant terms
    /// (`protocols/start/src/lib.rs:305`, consumed at `transport/session/src/manager.rs:2879-2883`),
    /// so an Entry that is made to prove nothing can pick `w`, publish constant terms summing to
    /// `w·G − e·G`, and end up with a combined commitment of `w·G` — a key it can sweep alone.
    ///
    /// It cannot then produce shares for the polynomial whose constant term it does not know, so the
    /// Exit never recovers the SSA and is never paid. But the Entry controls emission order
    /// (`generator.rs` builds `poly_queue`, `next_share` drains `front_mut()`) and puts that
    /// polynomial last, so it is served nearly the whole cycle before the Exit notices — by which
    /// time it has already swept the deposit. Note this is distinct from the by-design burn
    /// semantics: there *neither* party can recover, whereas here the party that owes can.
    ///
    /// Before [`SsaCommitmentProof`] existed this construction was accepted, and the Exit published
    /// `addr(w·G)` as the address to watch — verified by asserting exactly that. The proof cannot be
    /// forged here because producing it would require `dlog(w·G − e·G)`, and knowing that together
    /// with `w` yields `e`. So the assertion is now inverted.
    #[test]
    fn exit_refuses_a_client_commitment_whose_deposit_key_the_entry_knows() -> anyhow::Result<()> {
        const POLYS: u16 = 3;
        const THRESHOLD: u8 = 2;

        let mut rng = hopr_types::crypto_random::rng();
        let ssa_id = SsaId::new(SimplePseudonym::random(), SsaIndex::MIN);
        let reconstructor = SsaReconstructor::<TestSpec>::new(SsaReconstructorConfig::default());

        // The Exit reveals its half. This is exactly what `SsaRequest` carries to the Entry.
        let exit_public = reconstructor.new_exit_commitment(ssa_id, test_params(POLYS, THRESHOLD))?;

        // The attacker picks the deposit key it wants to end up holding.
        let w = PixScalar::<TestSpec>::random(&mut rng);
        let target = PixGroup::<TestSpec>::mul_by_generator(&w);

        // Honest constant terms for every polynomial but the last.
        let honest: Vec<PixGroup<TestSpec>> = (0..POLYS - 1)
            .map(|_| PixGroup::<TestSpec>::mul_by_generator(&PixScalar::<TestSpec>::random(&mut rng)))
            .collect();
        let honest_sum: PixGroup<TestSpec> = honest.iter().copied().sum();

        // The last one is obtained by group subtraction. The attacker never learns its discrete log
        // — that would require the Exit's secret — and does not need to.
        let rogue = target - exit_public - honest_sum;

        let mut constant_terms: HashMap<PolynomialIndex, PixGroupRepr<TestSpec>> = honest
            .iter()
            .enumerate()
            .map(|(poly_index, c0)| (poly_index as PolynomialIndex, c0.to_bytes()))
            .collect();
        constant_terms.insert((POLYS - 1) as PolynomialIndex, rogue.to_bytes());

        // The best the attacker can offer is a proof over the client commitment it actually
        // published, using the only scalar it knows — which is not that commitment's discrete log.
        let bogus_proof = SsaCommitmentProof::prove(&ssa_id, &w, &(target - exit_public))?;

        let rejected =
            reconstructor.insert_coefficient_commitments(ssa_id, 0, Some(bogus_proof), constant_terms.into_iter());
        assert!(
            matches!(rejected, Err(PixError::UnprovenSsaCommitment)),
            "a commitment whose discrete logarithm the sender does not know must be refused, got {rejected:?}"
        );

        // Nothing about the cycle may have been published: with no deposit address the strategy is
        // never asked to fund an SSA the Entry could reclaim.
        assert!(
            reconstructor
                .commitment_builder
                .get(&ssa_id)
                .is_some_and(|b| b.lock().get_deposit_address().is_none()),
            "no deposit address may be derived from an unproven commitment"
        );
        // The published cycle is what makes an SSA live and able to accept recovered shares.
        // `commitment_builder` legitimately still exists — `new_exit_commitment` created it.
        assert_eq!(
            0,
            reconstructor.live_cycles(),
            "no cycle must be published for an unproven commitment"
        );

        Ok(())
    }
}