prns_runtime_embassy/runtime/

embedded_persistence.rs

use embedded_storage_async::nor_flash::NorFlash;
use heapless::Vec as HeaplessVec;

use crate::crypto::ratchets::SeedSelfRatchetsOutcome;
use crate::engine::{EngineState, InstantMillis, Journaled, RouteSeedOutcome};
use crate::identity::Zeroizing;
use crate::interfaces::AttachedInterfaces;
use crate::persistence::{
    maximum_route_upsert_payload_len, read_routing_table_snapshot, read_self_ratchets_snapshot,
    routing_table_snapshot_len, self_ratchets_snapshot_len, write_routing_table_snapshot,
    write_self_ratchets_snapshot, FlashJournal, FlashJournalError, FlashJournalLayout,
    FlashJournalRecord, FlashJournalRecordKind, FlashJournalWarning,
    TIMEBASE_RECORD_INTERVAL_MILLIS,
};
use crate::routing::announce::emit::MAX_ANNOUNCE_APP_DATA_LEN;
use crate::routing::AnnounceIdRing;
use crate::storage::StorageLayout;
use crate::wire::{DestinationHash, TRUNCATED_HASH_BYTE_LEN};

const RECORD_SCRATCH_LEN: usize =
    (maximum_route_upsert_payload_len(MAX_ANNOUNCE_APP_DATA_LEN, 0) + 3) & !3;
const HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS: u64 = 24 * 60 * 60 * 1_000;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EmbeddedCompactionPolicy {
    minimum_interval_millis: u64,
    critical_reserve_bytes: usize,
}

impl EmbeddedCompactionPolicy {
    #[must_use]
    pub const fn new(minimum_interval_millis: u64, critical_reserve_bytes: usize) -> Self {
        Self {
            minimum_interval_millis,
            critical_reserve_bytes,
        }
    }

    #[must_use]
    pub const fn hopspot(critical_reserve_bytes: usize) -> Self {
        Self::new(
            HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS,
            critical_reserve_bytes,
        )
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EmbeddedPersistencePolicy {
    first_route_commit_delay_millis: u64,
    minimum_route_commit_interval_millis: u64,
    ratchet_batch_delay_millis: u64,
    retry_interval_millis: u64,
    timebase_record_interval_millis: u64,
    compaction: EmbeddedCompactionPolicy,
}

impl EmbeddedPersistencePolicy {
    #[must_use]
    pub const fn new(
        first_route_commit_delay_millis: u64,
        minimum_route_commit_interval_millis: u64,
        ratchet_batch_delay_millis: u64,
        retry_interval_millis: u64,
        timebase_record_interval_millis: u64,
        compaction: EmbeddedCompactionPolicy,
    ) -> Self {
        Self {
            first_route_commit_delay_millis,
            minimum_route_commit_interval_millis,
            ratchet_batch_delay_millis,
            retry_interval_millis,
            timebase_record_interval_millis,
            compaction,
        }
    }

    #[must_use]
    pub const fn hopspot_default(compaction: EmbeddedCompactionPolicy) -> Self {
        Self::new(
            2_000,
            5 * 60 * 1_000,
            2_000,
            5 * 60 * 1_000,
            TIMEBASE_RECORD_INTERVAL_MILLIS,
            compaction,
        )
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RouteSnapshotKeyError {
    Capacity,
}

pub trait RouteSnapshotKeys {
    fn clear(&mut self);
    fn push(&mut self, destination: DestinationHash) -> Result<(), RouteSnapshotKeyError>;
    fn get(&self, index: usize) -> Option<DestinationHash>;
}

pub struct FixedRouteSnapshotKeys<const N: usize> {
    keys: HeaplessVec<DestinationHash, N>,
}

impl<const N: usize> FixedRouteSnapshotKeys<N> {
    #[must_use]
    pub const fn new() -> Self {
        Self {
            keys: HeaplessVec::new(),
        }
    }
}

impl<const N: usize> Default for FixedRouteSnapshotKeys<N> {
    fn default() -> Self {
        Self::new()
    }
}

impl<const N: usize> RouteSnapshotKeys for FixedRouteSnapshotKeys<N> {
    fn clear(&mut self) {
        self.keys.clear();
    }

    fn push(&mut self, destination: DestinationHash) -> Result<(), RouteSnapshotKeyError> {
        self.keys
            .push(destination)
            .map_err(|_| RouteSnapshotKeyError::Capacity)
    }

    fn get(&self, index: usize) -> Option<DestinationHash> {
        self.keys.get(index).copied()
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EmbeddedPersistenceFailure {
    Flash,
    Codec,
    Capacity,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EmbeddedPersistenceTarget {
    Routes,
    CriticalState,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EmbeddedPersistenceRestoreReport {
    pub logical_start: InstantMillis,
    pub route_seeded_count: u32,
    pub route_refused_count: u32,
    pub route_dropped_count: u32,
    pub ratchet_seeded_count: u32,
    pub ratchet_refused_count: u32,
    pub warning: Option<FlashJournalWarning>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EmbeddedPersistenceDiagnostic {
    Restored(EmbeddedPersistenceRestoreReport),
    BatchPersisted {
        records: u32,
        at: InstantMillis,
        state_not_saved: bool,
    },
    CompactionStarted {
        at: InstantMillis,
        next_allowed_at: InstantMillis,
    },
    CompactionCompleted {
        records: u32,
        at: InstantMillis,
        state_not_saved: bool,
    },
    DurabilityDeferred {
        target: EmbeddedPersistenceTarget,
        until: InstantMillis,
    },
    WriteFailed {
        failure: EmbeddedPersistenceFailure,
        retry_at: InstantMillis,
    },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PendingRouteDelta {
    RouteUpsert(DestinationHash),
    RouteRemoval(DestinationHash),
}

impl PendingRouteDelta {
    fn destination(self) -> DestinationHash {
        match self {
            Self::RouteUpsert(destination) | Self::RouteRemoval(destination) => destination,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BatchKind {
    Routes,
    Ratchets,
    Compaction,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CompactionPhase {
    RecordBudget { at: InstantMillis },
    Erase { sector: usize },
    Routes { index: usize },
    Ratchets { index: usize },
    Commit,
}

struct EncodedDelta {
    kind: FlashJournalRecordKind,
    payload: Zeroizing<[u8; RECORD_SCRATCH_LEN]>,
    len: usize,
}

pub struct EmbeddedFlashPersistence<F, Keys, Observe, const PENDING: usize>
where
    F: NorFlash,
    Keys: RouteSnapshotKeys,
    Observe: FnMut(EmbeddedPersistenceDiagnostic),
{
    flash: Option<F>,
    journal: Option<FlashJournal<F>>,
    layout: FlashJournalLayout,
    policy: EmbeddedPersistencePolicy,
    observe_diagnostic: Observe,
    pending_routes: HeaplessVec<PendingRouteDelta, PENDING>,
    pending_ratchets: HeaplessVec<DestinationHash, PENDING>,
    compaction_route_keys: Keys,
    compaction_ratchet_keys: HeaplessVec<DestinationHash, PENDING>,
    route_dirty_since: Option<InstantMillis>,
    ratchet_dirty_since: Option<InstantMillis>,
    last_route_success: Option<InstantMillis>,
    last_timebase_success: Option<InstantMillis>,
    retry_not_before: Option<InstantMillis>,
    landing_batch: Option<BatchKind>,
    landing_records: u32,
    compaction: Option<CompactionPhase>,
    compaction_target: Option<EmbeddedPersistenceTarget>,
    snapshot_required: bool,
    snapshot_target: EmbeddedPersistenceTarget,
    next_compaction_not_before: Option<InstantMillis>,
    deferred_target: Option<EmbeddedPersistenceTarget>,
    deferred_until: Option<InstantMillis>,
    write_failed: bool,
}

impl<F, Keys, Observe, const PENDING: usize> EmbeddedFlashPersistence<F, Keys, Observe, PENDING>
where
    F: NorFlash,
    Keys: RouteSnapshotKeys,
    Observe: FnMut(EmbeddedPersistenceDiagnostic),
{
    #[must_use]
    pub fn new(
        flash: F,
        layout: FlashJournalLayout,
        policy: EmbeddedPersistencePolicy,
        compaction_route_keys: Keys,
        observe_diagnostic: Observe,
    ) -> Self {
        Self {
            flash: Some(flash),
            journal: None,
            layout,
            policy,
            observe_diagnostic,
            pending_routes: HeaplessVec::new(),
            pending_ratchets: HeaplessVec::new(),
            compaction_route_keys,
            compaction_ratchet_keys: HeaplessVec::new(),
            route_dirty_since: None,
            ratchet_dirty_since: None,
            last_route_success: None,
            last_timebase_success: None,
            retry_not_before: None,
            landing_batch: None,
            landing_records: 0,
            compaction: None,
            compaction_target: None,
            snapshot_required: false,
            snapshot_target: EmbeddedPersistenceTarget::Routes,
            next_compaction_not_before: None,
            deferred_target: None,
            deferred_until: None,
            write_failed: false,
        }
    }

    #[must_use]
    pub fn state_not_saved(&self) -> bool {
        self.write_failed || self.deferred_target.is_some()
    }

    pub async fn restore<S: StorageLayout>(
        &mut self,
        engine: &mut EngineState<S>,
        raw_now: InstantMillis,
    ) -> EmbeddedPersistenceRestoreReport {
        let Some(mut flash) = self.flash.take() else {
            return self.empty_restore_report(raw_now, Some(FlashJournalWarning::Corrupt));
        };
        let timebase_state = FlashJournal::inspect_timebase_state(&mut flash, self.layout)
            .await
            .ok();
        let logical_start = timebase_state
            .and_then(|state| state.high_water)
            .unwrap_or(raw_now);
        let mut scratch = Zeroizing::new([0u8; RECORD_SCRATCH_LEN]);
        let mut report = EmbeddedPersistenceRestoreReport {
            logical_start,
            route_seeded_count: 0,
            route_refused_count: 0,
            route_dropped_count: 0,
            ratchet_seeded_count: 0,
            ratchet_refused_count: 0,
            warning: None,
        };
        let opened = FlashJournal::open(flash, self.layout, &mut scratch[..], |record| {
            apply_record(engine, logical_start, record, &mut report)
        })
        .await;
        let Ok((mut journal, restored)) = opened else {
            report.warning = Some(FlashJournalWarning::Corrupt);
            (self.observe_diagnostic)(EmbeddedPersistenceDiagnostic::Restored(report));
            return report;
        };
        report.warning = restored.warning;
        let initialization_failed =
            restored.active_epoch.is_none() && journal.initialize_empty().await.is_err();
        self.next_compaction_not_before = timebase_state
            .and_then(|state| state.last_compaction_attempt)
            .map(|attempt| {
                InstantMillis(
                    attempt
                        .0
                        .saturating_add(self.policy.compaction.minimum_interval_millis),
                )
            });
        self.journal = Some(journal);
        (self.observe_diagnostic)(EmbeddedPersistenceDiagnostic::Restored(report));
        if initialization_failed {
            self.note_write_failure(raw_now, EmbeddedPersistenceFailure::Flash);
        }
        report
    }

    fn empty_restore_report(
        &mut self,
        logical_start: InstantMillis,
        warning: Option<FlashJournalWarning>,
    ) -> EmbeddedPersistenceRestoreReport {
        let report = EmbeddedPersistenceRestoreReport {
            logical_start,
            route_seeded_count: 0,
            route_refused_count: 0,
            route_dropped_count: 0,
            ratchet_seeded_count: 0,
            ratchet_refused_count: 0,
            warning,
        };
        (self.observe_diagnostic)(EmbeddedPersistenceDiagnostic::Restored(report));
        report
    }

    fn observe_journaled(&mut self, journaled: &Journaled<'_>, now: InstantMillis) {
        match journaled {
            Journaled::AnnounceHeard { observation, .. } => {
                self.queue_route(PendingRouteDelta::RouteUpsert(observation.destination), now);
                if self.route_dirty_since.is_none() {
                    self.route_dirty_since = Some(now);
                }
            }
            Journaled::RouteRemoved { destination, .. } => {
                self.queue_route(PendingRouteDelta::RouteRemoval(*destination), now);
                if self.route_dirty_since.is_none() {
                    self.route_dirty_since = Some(now);
                }
            }
            Journaled::SelfRatchetRotated { destination } => {
                self.queue_ratchet(*destination, now);
                if self.ratchet_dirty_since.is_none() {
                    self.ratchet_dirty_since = Some(now);
                }
            }
            Journaled::AnnounceHeldDropped { .. }
            | Journaled::Delivered(_)
            | Journaled::CommandSettled { .. }
            | Journaled::PersistenceFlushed { .. }
            | Journaled::PersistenceFlushFailed { .. }
            | Journaled::LinkEstablished(_)
            | Journaled::PeerIdentified { .. }
            | Journaled::RequestReceived { .. }
            | Journaled::ResponseReceived { .. }
            | Journaled::ResponseSegmentReceived { .. }
            | Journaled::ChannelMessageReceived { .. }
            | Journaled::LinkClosed { .. }
            | Journaled::LinkInterfaceMismatch { .. }
            | Journaled::ResourceReceived { .. }
            | Journaled::ResourceFailed { .. }
            | Journaled::ResourceNeedsDecompression { .. }
            | Journaled::ResourceSegmentReceived { .. }
            | Journaled::ResourceAssembled { .. } => {}
        }
    }

    fn queue_route(&mut self, delta: PendingRouteDelta, now: InstantMillis) {
        if self.snapshot_required {
            return;
        }
        if let Some(existing) = self
            .pending_routes
            .iter_mut()
            .find(|pending| pending.destination() == delta.destination())
        {
            *existing = delta;
            return;
        }
        if self.pending_routes.push(delta).is_err() {
            self.require_snapshot(EmbeddedPersistenceTarget::Routes, now);
        }
    }

    fn queue_ratchet(&mut self, destination: DestinationHash, now: InstantMillis) {
        if self.pending_ratchets.contains(&destination) {
            return;
        }
        if self.pending_ratchets.push(destination).is_err() {
            self.require_snapshot(EmbeddedPersistenceTarget::CriticalState, now);
        }
    }

    fn require_snapshot(&mut self, target: EmbeddedPersistenceTarget, now: InstantMillis) {
        self.snapshot_required = true;
        self.snapshot_target = match (self.snapshot_target, target) {
            (EmbeddedPersistenceTarget::CriticalState, _)
            | (_, EmbeddedPersistenceTarget::CriticalState) => {
                EmbeddedPersistenceTarget::CriticalState
            }
            (EmbeddedPersistenceTarget::Routes, EmbeddedPersistenceTarget::Routes) => {
                EmbeddedPersistenceTarget::Routes
            }
        };
        self.pending_routes.clear();
        if target == EmbeddedPersistenceTarget::CriticalState {
            self.pending_ratchets.clear();
        }
        let Some(until) = self.next_compaction_not_before else {
            return;
        };
        if now.0 >= until.0 {
            return;
        }
        if self.deferred_target == Some(self.snapshot_target) && self.deferred_until == Some(until)
        {
            return;
        }
        self.deferred_target = Some(self.snapshot_target);
        self.deferred_until = Some(until);
        (self.observe_diagnostic)(EmbeddedPersistenceDiagnostic::DurabilityDeferred {
            target: self.snapshot_target,
            until,
        });
    }

    fn next_deadline(&self, now: InstantMillis) -> Option<InstantMillis> {
        self.journal.as_ref()?;
        let mut deadline = if self.compaction.is_some() || self.landing_batch.is_some() {
            Some(now)
        } else {
            None
        };
        let ratchet_ready = self.ratchet_ready_at();
        let route_ready = self.route_ready_at();
        deadline = earlier(deadline, ratchet_ready);
        if self.snapshot_required {
            let requested = match self.snapshot_target {
                EmbeddedPersistenceTarget::Routes => route_ready,
                EmbeddedPersistenceTarget::CriticalState => {
                    earlier(route_ready, ratchet_ready).or(Some(now))
                }
            };
            if let Some(requested) = requested {
                let allowed = self.next_compaction_not_before.unwrap_or(requested);
                deadline = earlier(deadline, Some(InstantMillis(requested.0.max(allowed.0))));
            }
        } else {
            deadline = earlier(deadline, route_ready);
        }
        match (deadline, self.retry_not_before) {
            (Some(deadline), Some(retry)) => Some(InstantMillis(deadline.0.max(retry.0))),
            (deadline, None) => deadline,
            (None, Some(_)) => None,
        }
    }

    fn ratchet_ready_at(&self) -> Option<InstantMillis> {
        self.ratchet_dirty_since.map(|dirty| {
            InstantMillis(
                dirty
                    .0
                    .saturating_add(self.policy.ratchet_batch_delay_millis),
            )
        })
    }

    fn route_ready_at(&self) -> Option<InstantMillis> {
        self.route_dirty_since.map(|dirty| {
            let first_ready = dirty
                .0
                .saturating_add(self.policy.first_route_commit_delay_millis);
            let interval_ready = self.last_route_success.map_or(0, |last| {
                last.0
                    .saturating_add(self.policy.minimum_route_commit_interval_millis)
            });
            InstantMillis(first_ready.max(interval_ready))
        })
    }

    async fn progress<S: StorageLayout>(
        &mut self,
        engine: &mut EngineState<S>,
        now: InstantMillis,
    ) {
        if self.retry_not_before.is_some_and(|retry| now.0 < retry.0) {
            return;
        }
        if self.compaction.is_some() {
            self.progress_compaction(engine, now).await;
            return;
        }
        if let Some(batch) = self.landing_batch {
            let new_work = match batch {
                BatchKind::Routes => !self.pending_routes.is_empty() || self.snapshot_required,
                BatchKind::Ratchets => !self.pending_ratchets.is_empty(),
                BatchKind::Compaction => {
                    !self.pending_routes.is_empty()
                        || !self.pending_ratchets.is_empty()
                        || self.snapshot_required
                }
            };
            if new_work {
                self.landing_batch = None;
            } else {
                self.land_timebase(batch, now).await;
                return;
            }
        }
        let ratchet_due = self
            .ratchet_ready_at()
            .is_some_and(|ready| now.0 >= ready.0);
        let route_due = self.route_ready_at().is_some_and(|ready| now.0 >= ready.0);
        if ratchet_due {
            if let Some(index) = (!self.pending_ratchets.is_empty()).then_some(0) {
                self.append_ratchet(engine, index, now).await;
                return;
            }
            if self.snapshot_required
                && self.snapshot_target == EmbeddedPersistenceTarget::CriticalState
            {
                self.try_start_compaction(engine, now);
                return;
            }
        }
        if !route_due {
            return;
        }
        if self.snapshot_required {
            self.try_start_compaction(engine, now);
            return;
        }
        if !self.pending_routes.is_empty() {
            self.append_route(engine, 0, now).await;
        }
    }

    async fn append_route<S: StorageLayout>(
        &mut self,
        engine: &EngineState<S>,
        index: usize,
        now: InstantMillis,
    ) {
        let delta = self.pending_routes[index];
        let encoded = encode_route_delta(engine, delta);
        let Ok(payload) = encoded else {
            self.note_codec_failure(now);
            return;
        };
        let can_fit = self.journal.as_ref().is_some_and(|journal| {
            journal.active_can_fit(payload.len, self.policy.compaction.critical_reserve_bytes)
        });
        if !can_fit {
            self.require_snapshot(EmbeddedPersistenceTarget::Routes, now);
            self.try_start_compaction(engine, now);
            return;
        }
        let Some(journal) = self.journal.as_mut() else {
            return;
        };
        let result = journal
            .append(payload.kind, &payload.payload[..payload.len])
            .await;
        match result {
            Ok(()) => {
                self.pending_routes.swap_remove(index);
                self.landing_records = self.landing_records.saturating_add(1);
                if self.pending_routes.is_empty() {
                    self.landing_batch = Some(BatchKind::Routes);
                }
            }
            Err(FlashJournalError::ArenaFull) => {
                self.require_snapshot(EmbeddedPersistenceTarget::Routes, now);
                self.try_start_compaction(engine, now);
            }
            Err(error) => {
                let failure = failure_from_journal(error);
                self.note_write_failure(now, failure);
            }
        }
    }

    async fn append_ratchet<S: StorageLayout>(
        &mut self,
        engine: &EngineState<S>,
        index: usize,
        now: InstantMillis,
    ) {
        let destination = self.pending_ratchets[index];
        let Ok(payload) = encode_ratchet(engine, destination) else {
            self.note_codec_failure(now);
            return;
        };
        let Some(journal) = self.journal.as_mut() else {
            return;
        };
        let result = journal
            .append(payload.kind, &payload.payload[..payload.len])
            .await;
        match result {
            Ok(()) => {
                self.pending_ratchets.swap_remove(index);
                self.landing_records = self.landing_records.saturating_add(1);
                if self.pending_ratchets.is_empty() {
                    self.landing_batch = Some(BatchKind::Ratchets);
                }
            }
            Err(FlashJournalError::ArenaFull) => {
                self.require_snapshot(EmbeddedPersistenceTarget::CriticalState, now);
                self.try_start_compaction(engine, now);
            }
            Err(error) => self.note_write_failure(now, failure_from_journal(error)),
        }
    }

    fn try_start_compaction<S: StorageLayout>(
        &mut self,
        engine: &EngineState<S>,
        now: InstantMillis,
    ) {
        let allowed = self.next_compaction_not_before.unwrap_or(now);
        if now.0 < allowed.0 {
            self.require_snapshot(self.snapshot_target, now);
            return;
        }
        self.compaction_route_keys.clear();
        let mut route_capacity_failed = false;
        for destination in engine.persisted_route_destinations() {
            if self.compaction_route_keys.push(destination).is_err() {
                route_capacity_failed = true;
                break;
            }
        }
        if route_capacity_failed {
            self.note_write_failure(now, EmbeddedPersistenceFailure::Capacity);
            return;
        }
        self.compaction_ratchet_keys.clear();
        let mut ratchet_capacity_failed = false;
        for (destination, _, _) in engine.persisted_self_ratchet_rows() {
            if self.compaction_ratchet_keys.push(destination).is_err() {
                ratchet_capacity_failed = true;
                break;
            }
        }
        if ratchet_capacity_failed {
            self.note_write_failure(now, EmbeddedPersistenceFailure::Capacity);
            return;
        }
        let target = self.snapshot_target;
        self.pending_routes.clear();
        self.pending_ratchets.clear();
        self.route_dirty_since = None;
        self.ratchet_dirty_since = None;
        self.snapshot_required = false;
        self.snapshot_target = EmbeddedPersistenceTarget::Routes;
        self.compaction_target = Some(target);
        self.compaction = Some(CompactionPhase::RecordBudget { at: now });
        self.landing_batch = None;
        self.landing_records = 0;
    }

    async fn progress_compaction<S: StorageLayout>(
        &mut self,
        engine: &mut EngineState<S>,
        now: InstantMillis,
    ) {
        let Some(phase) = self.compaction else {
            return;
        };
        let Some(journal) = self.journal.as_mut() else {
            return;
        };
        match phase {
            CompactionPhase::RecordBudget { at } => {
                match journal.record_compaction_budget(at).await {
                    Ok(recorded_at) => {
                        self.last_timebase_success = Some(at);
                        let next_allowed_at = InstantMillis(
                            recorded_at
                                .0
                                .saturating_add(self.policy.compaction.minimum_interval_millis),
                        );
                        self.next_compaction_not_before = Some(next_allowed_at);
                        (self.observe_diagnostic)(
                            EmbeddedPersistenceDiagnostic::CompactionStarted {
                                at,
                                next_allowed_at,
                            },
                        );
                        self.compaction = Some(CompactionPhase::Erase { sector: 0 });
                    }
                    Err(error) => {
                        self.note_write_failure(now, failure_from_journal(error));
                    }
                }
            }
            CompactionPhase::Erase { sector } => {
                if sector < journal.inactive_sector_count() {
                    match journal.erase_inactive_sector(sector).await {
                        Ok(()) => {
                            let next = sector + 1;
                            if next == journal.inactive_sector_count() {
                                if journal.begin_compaction().is_err() {
                                    self.note_write_failure(
                                        now,
                                        EmbeddedPersistenceFailure::Capacity,
                                    );
                                    return;
                                }
                                self.compaction = Some(CompactionPhase::Routes { index: 0 });
                            } else {
                                self.compaction = Some(CompactionPhase::Erase { sector: next });
                            }
                        }
                        Err(error) => {
                            self.note_write_failure(now, failure_from_journal(error));
                        }
                    }
                }
            }
            CompactionPhase::Routes { index } => {
                let mut scratch = [0u8; RECORD_SCRATCH_LEN];
                let Some(destination) = self.compaction_route_keys.get(index) else {
                    self.compaction = Some(CompactionPhase::Ratchets { index: 0 });
                    return;
                };
                let Some(row) = engine.persisted_route_row(&destination) else {
                    self.compaction = Some(CompactionPhase::Routes { index: index + 1 });
                    return;
                };
                let mut durable = row.clone();
                durable.announce_id_ring = AnnounceIdRing::Table(&[]);
                let required = routing_table_snapshot_len(core::iter::once(durable.clone()));
                if required > scratch.len() {
                    self.note_codec_failure(now);
                    return;
                }
                let Ok(written) = write_routing_table_snapshot(
                    core::iter::once(durable),
                    &mut scratch[..required],
                ) else {
                    self.note_codec_failure(now);
                    return;
                };
                match journal
                    .append_compacted(FlashJournalRecordKind::RouteUpsert, &scratch[..written])
                    .await
                {
                    Ok(()) => {
                        self.landing_records = self.landing_records.saturating_add(1);
                        self.compaction = Some(CompactionPhase::Routes { index: index + 1 });
                    }
                    Err(error) => {
                        self.note_write_failure(now, failure_from_journal(error));
                    }
                }
            }
            CompactionPhase::Ratchets { index } => {
                let Some(destination) = self.compaction_ratchet_keys.get(index).copied() else {
                    self.compaction = Some(CompactionPhase::Commit);
                    return;
                };
                let Some((last_rotated, secrets)) = engine.persisted_self_ratchet_row(&destination)
                else {
                    self.compaction = Some(CompactionPhase::Ratchets { index: index + 1 });
                    return;
                };
                let mut scratch = Zeroizing::new([0u8; RECORD_SCRATCH_LEN]);
                scratch[..TRUNCATED_HASH_BYTE_LEN].copy_from_slice(destination.as_bytes());
                let required = self_ratchets_snapshot_len(secrets.len());
                let end = TRUNCATED_HASH_BYTE_LEN.saturating_add(required);
                if end > scratch.len() {
                    self.note_codec_failure(now);
                    return;
                }
                let Ok(written) = write_self_ratchets_snapshot(
                    last_rotated,
                    secrets,
                    &mut scratch[TRUNCATED_HASH_BYTE_LEN..end],
                ) else {
                    self.note_codec_failure(now);
                    return;
                };
                match journal
                    .append_compacted(
                        FlashJournalRecordKind::SelfRatchet,
                        &scratch[..TRUNCATED_HASH_BYTE_LEN + written],
                    )
                    .await
                {
                    Ok(()) => {
                        self.landing_records = self.landing_records.saturating_add(1);
                        self.compaction = Some(CompactionPhase::Ratchets { index: index + 1 });
                    }
                    Err(error) => {
                        self.note_write_failure(now, failure_from_journal(error));
                    }
                }
            }
            CompactionPhase::Commit => match journal.commit_compaction().await {
                Ok(()) => {
                    self.compaction = None;
                    self.compaction_target = None;
                    let records = core::mem::take(&mut self.landing_records);
                    self.retry_not_before = None;
                    self.write_failed = false;
                    if self.snapshot_required {
                        self.require_snapshot(self.snapshot_target, now);
                    } else {
                        self.deferred_target = None;
                        self.deferred_until = None;
                    }
                    let state_not_saved = self.state_not_saved();
                    (self.observe_diagnostic)(EmbeddedPersistenceDiagnostic::CompactionCompleted {
                        records,
                        at: now,
                        state_not_saved,
                    });
                    if !self.snapshot_required
                        && self.pending_routes.is_empty()
                        && self.pending_ratchets.is_empty()
                    {
                        self.landing_batch = Some(BatchKind::Compaction);
                    }
                }
                Err(error) => {
                    self.note_write_failure(now, failure_from_journal(error));
                }
            },
        }
    }

    async fn land_timebase(&mut self, batch: BatchKind, now: InstantMillis) {
        let should_record = self.last_timebase_success.is_none_or(|last| {
            now.0.saturating_sub(last.0) >= self.policy.timebase_record_interval_millis
        });
        if should_record {
            let Some(journal) = self.journal.as_mut() else {
                return;
            };
            if let Err(error) = journal.record_timebase(now).await {
                self.note_write_failure(now, failure_from_journal(error));
                return;
            }
            self.last_timebase_success = Some(now);
        }
        match batch {
            BatchKind::Routes => {
                self.route_dirty_since = None;
                self.last_route_success = Some(now);
            }
            BatchKind::Ratchets => {
                self.ratchet_dirty_since = None;
            }
            BatchKind::Compaction => {
                self.route_dirty_since = None;
                self.ratchet_dirty_since = None;
                self.last_route_success = Some(now);
            }
        }
        self.retry_not_before = None;
        self.write_failed = false;
        let records = core::mem::take(&mut self.landing_records);
        self.landing_batch = None;
        if batch != BatchKind::Compaction {
            let state_not_saved = self.state_not_saved();
            (self.observe_diagnostic)(EmbeddedPersistenceDiagnostic::BatchPersisted {
                records,
                at: now,
                state_not_saved,
            });
        }
    }

    fn note_codec_failure(&mut self, now: InstantMillis) {
        self.note_write_failure(now, EmbeddedPersistenceFailure::Codec);
    }

    fn note_write_failure(&mut self, now: InstantMillis, failure: EmbeddedPersistenceFailure) {
        let retry_at = InstantMillis(now.0.saturating_add(self.policy.retry_interval_millis));
        self.retry_not_before = Some(retry_at);
        self.write_failed = true;
        if self.compaction.is_some() {
            let target = self
                .compaction_target
                .unwrap_or(EmbeddedPersistenceTarget::Routes);
            if let Some(journal) = self.journal.as_mut() {
                journal.abort_compaction();
            }
            self.compaction = None;
            self.compaction_target = None;
            self.require_snapshot(target, now);
            match target {
                EmbeddedPersistenceTarget::Routes => {
                    self.route_dirty_since.get_or_insert(now);
                }
                EmbeddedPersistenceTarget::CriticalState => {
                    self.ratchet_dirty_since.get_or_insert(now);
                }
            }
        }
        (self.observe_diagnostic)(EmbeddedPersistenceDiagnostic::WriteFailed { failure, retry_at });
    }
}

pub(crate) trait ManifoldPersistence<S: StorageLayout> {
    fn observe(&mut self, journaled: &Journaled<'_>, now: InstantMillis);
    fn deadline(&self, now: InstantMillis) -> Option<InstantMillis>;
    async fn progress(&mut self, engine: &mut EngineState<S>, now: InstantMillis);
}

impl<S, F, Keys, Observe, const PENDING: usize> ManifoldPersistence<S>
    for EmbeddedFlashPersistence<F, Keys, Observe, PENDING>
where
    S: StorageLayout,
    F: NorFlash,
    Keys: RouteSnapshotKeys,
    Observe: FnMut(EmbeddedPersistenceDiagnostic),
{
    fn observe(&mut self, journaled: &Journaled<'_>, now: InstantMillis) {
        self.observe_journaled(journaled, now);
    }

    fn deadline(&self, now: InstantMillis) -> Option<InstantMillis> {
        self.next_deadline(now)
    }

    async fn progress(&mut self, engine: &mut EngineState<S>, now: InstantMillis) {
        self.progress(engine, now).await;
    }
}

pub(crate) struct NoManifoldPersistence;

impl<S: StorageLayout> ManifoldPersistence<S> for NoManifoldPersistence {
    fn observe(&mut self, _journaled: &Journaled<'_>, _now: InstantMillis) {}

    fn deadline(&self, _now: InstantMillis) -> Option<InstantMillis> {
        None
    }

    async fn progress(&mut self, _engine: &mut EngineState<S>, _now: InstantMillis) {}
}

fn encode_route_delta<S: StorageLayout>(
    engine: &EngineState<S>,
    delta: PendingRouteDelta,
) -> Result<EncodedDelta, ()> {
    match delta {
        PendingRouteDelta::RouteUpsert(destination) => {
            let Some(row) = engine.persisted_route_row(&destination) else {
                return encode_tombstone(destination);
            };
            let mut durable = row.clone();
            durable.announce_id_ring = AnnounceIdRing::Table(&[]);
            let required = routing_table_snapshot_len(core::iter::once(durable.clone()));
            if required > RECORD_SCRATCH_LEN {
                return Err(());
            }
            let mut scratch = Zeroizing::new([0u8; RECORD_SCRATCH_LEN]);
            let written =
                write_routing_table_snapshot(core::iter::once(durable), &mut scratch[..required])
                    .map_err(|_| ())?;
            Ok(EncodedDelta {
                kind: FlashJournalRecordKind::RouteUpsert,
                payload: scratch,
                len: written,
            })
        }
        PendingRouteDelta::RouteRemoval(destination) => encode_tombstone(destination),
    }
}

fn encode_ratchet<S: StorageLayout>(
    engine: &EngineState<S>,
    destination: DestinationHash,
) -> Result<EncodedDelta, ()> {
    let Some((last_rotated, secrets)) = engine.persisted_self_ratchet_row(&destination) else {
        return Err(());
    };
    let required = self_ratchets_snapshot_len(secrets.len());
    if TRUNCATED_HASH_BYTE_LEN + required > RECORD_SCRATCH_LEN {
        return Err(());
    }
    let mut scratch = Zeroizing::new([0u8; RECORD_SCRATCH_LEN]);
    scratch[..TRUNCATED_HASH_BYTE_LEN].copy_from_slice(destination.as_bytes());
    let written = write_self_ratchets_snapshot(
        last_rotated,
        secrets,
        &mut scratch[TRUNCATED_HASH_BYTE_LEN..TRUNCATED_HASH_BYTE_LEN + required],
    )
    .map_err(|_| ())?;
    Ok(EncodedDelta {
        kind: FlashJournalRecordKind::SelfRatchet,
        payload: scratch,
        len: TRUNCATED_HASH_BYTE_LEN + written,
    })
}

fn encode_tombstone(destination: DestinationHash) -> Result<EncodedDelta, ()> {
    let mut payload = Zeroizing::new([0u8; RECORD_SCRATCH_LEN]);
    payload[..TRUNCATED_HASH_BYTE_LEN].copy_from_slice(destination.as_bytes());
    Ok(EncodedDelta {
        kind: FlashJournalRecordKind::RouteRemoval,
        payload,
        len: TRUNCATED_HASH_BYTE_LEN,
    })
}

fn apply_record<S: StorageLayout>(
    engine: &mut EngineState<S>,
    now: InstantMillis,
    record: FlashJournalRecord<'_>,
    report: &mut EmbeddedPersistenceRestoreReport,
) {
    match record.kind {
        FlashJournalRecordKind::ArenaCommit => {}
        FlashJournalRecordKind::RouteUpsert => {
            let Ok(mut rows) = read_routing_table_snapshot(record.payload) else {
                report.route_refused_count = report.route_refused_count.saturating_add(1);
                return;
            };
            let Some(Ok(row)) = rows.next() else {
                report.route_refused_count = report.route_refused_count.saturating_add(1);
                return;
            };
            if rows.next().is_some() {
                report.route_refused_count = report.route_refused_count.saturating_add(1);
                return;
            }
            let destination = row.destination;
            let Ok(pending) = engine.prepare_persisted_route(row) else {
                report.route_refused_count = report.route_refused_count.saturating_add(1);
                return;
            };
            let Ok(verified) = pending.verify() else {
                report.route_refused_count = report.route_refused_count.saturating_add(1);
                return;
            };
            let _ = engine.drop_route(&destination, AttachedInterfaces::new(&[]));
            match engine.seed_verified_route(verified, now) {
                RouteSeedOutcome::Seeded => {
                    report.route_seeded_count = report.route_seeded_count.saturating_add(1);
                }
                RouteSeedOutcome::RefusedDestinationMismatch
                | RouteSeedOutcome::RefusedBlackholedIdentity
                | RouteSeedOutcome::RefusedInvalidSignature => {
                    report.route_refused_count = report.route_refused_count.saturating_add(1);
                }
                RouteSeedOutcome::AlreadyPresent
                | RouteSeedOutcome::TableFull
                | RouteSeedOutcome::AppDataArenaFull => {
                    report.route_dropped_count = report.route_dropped_count.saturating_add(1);
                }
            }
        }
        FlashJournalRecordKind::RouteRemoval => {
            let Ok(bytes) = <[u8; TRUNCATED_HASH_BYTE_LEN]>::try_from(record.payload) else {
                report.route_refused_count = report.route_refused_count.saturating_add(1);
                return;
            };
            let destination = DestinationHash::new(bytes);
            let _ = engine.drop_route(&destination, AttachedInterfaces::new(&[]));
        }
        FlashJournalRecordKind::SelfRatchet => {
            let Some((destination, sealed)) = record
                .payload
                .split_first_chunk::<TRUNCATED_HASH_BYTE_LEN>()
            else {
                report.ratchet_refused_count = report.ratchet_refused_count.saturating_add(1);
                return;
            };
            let Ok(restored) = read_self_ratchets_snapshot(sealed) else {
                report.ratchet_refused_count = report.ratchet_refused_count.saturating_add(1);
                return;
            };
            match engine.replace_persisted_self_ratchets(
                &DestinationHash::new(*destination),
                restored.last_rotated,
                restored.secrets_newest_first(),
            ) {
                SeedSelfRatchetsOutcome::Seeded => {
                    report.ratchet_seeded_count = report.ratchet_seeded_count.saturating_add(1);
                }
                SeedSelfRatchetsOutcome::AlreadyMinted | SeedSelfRatchetsOutcome::Untracked => {
                    report.ratchet_refused_count = report.ratchet_refused_count.saturating_add(1);
                }
            }
        }
    }
}

fn failure_from_journal<E>(error: FlashJournalError<E>) -> EmbeddedPersistenceFailure {
    match error {
        FlashJournalError::Flash(_) => EmbeddedPersistenceFailure::Flash,
        FlashJournalError::ArenaFull
        | FlashJournalError::OutOfBounds
        | FlashJournalError::Misaligned
        | FlashJournalError::Uninitialized
        | FlashJournalError::CompactionInProgress
        | FlashJournalError::NoCompaction
        | FlashJournalError::PayloadTooLarge
        | FlashJournalError::ScratchTooShort => EmbeddedPersistenceFailure::Capacity,
    }
}

fn earlier(first: Option<InstantMillis>, second: Option<InstantMillis>) -> Option<InstantMillis> {
    match (first, second) {
        (Some(first), Some(second)) => Some(InstantMillis(first.0.min(second.0))),
        (Some(value), None) | (None, Some(value)) => Some(value),
        (None, None) => None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use embedded_storage::nor_flash::{ErrorType, NorFlashError, NorFlashErrorKind};
    use embedded_storage_async::nor_flash::ReadNorFlash;
    use std::cell::{Cell, RefCell};
    use std::rc::Rc;
    use std::vec::Vec;

    const ERASE: usize = 512;
    const CAPACITY: usize = ERASE * 6;
    const LAYOUT: FlashJournalLayout = FlashJournalLayout::new(
        [0, ERASE as u32],
        [
            crate::persistence::FlashArenaRange::new((ERASE * 2) as u32, (ERASE * 4) as u32),
            crate::persistence::FlashArenaRange::new((ERASE * 4) as u32, (ERASE * 6) as u32),
        ],
    );

    #[derive(Debug)]
    struct TestFlash {
        bytes: [u8; CAPACITY],
        sector_erases: [u32; CAPACITY / ERASE],
        fail_next_write: Rc<Cell<bool>>,
    }

    impl TestFlash {
        fn new() -> Self {
            Self {
                bytes: [0xFF; CAPACITY],
                sector_erases: [0; CAPACITY / ERASE],
                fail_next_write: Rc::new(Cell::new(false)),
            }
        }

        fn controlled() -> (Self, Rc<Cell<bool>>) {
            let flash = Self::new();
            let control = Rc::clone(&flash.fail_next_write);
            (flash, control)
        }
    }

    #[derive(Debug)]
    struct TestFlashError;

    impl NorFlashError for TestFlashError {
        fn kind(&self) -> NorFlashErrorKind {
            NorFlashErrorKind::Other
        }
    }

    impl ErrorType for TestFlash {
        type Error = TestFlashError;
    }

    impl ReadNorFlash for TestFlash {
        const READ_SIZE: usize = 4;

        async fn read(&mut self, offset: u32, bytes: &mut [u8]) -> Result<(), Self::Error> {
            let start = offset as usize;
            let end = start + bytes.len();
            bytes.copy_from_slice(&self.bytes[start..end]);
            Ok(())
        }

        fn capacity(&self) -> usize {
            CAPACITY
        }
    }

    impl NorFlash for TestFlash {
        const WRITE_SIZE: usize = 4;
        const ERASE_SIZE: usize = ERASE;

        async fn write(&mut self, offset: u32, bytes: &[u8]) -> Result<(), Self::Error> {
            if self.fail_next_write.replace(false) {
                return Err(TestFlashError);
            }
            let start = offset as usize;
            for (stored, written) in self.bytes[start..start + bytes.len()].iter_mut().zip(bytes) {
                *stored &= *written;
            }
            Ok(())
        }

        async fn erase(&mut self, from: u32, to: u32) -> Result<(), Self::Error> {
            self.bytes[from as usize..to as usize].fill(0xFF);
            for sector in from as usize / ERASE..to as usize / ERASE {
                self.sector_erases[sector] = self.sector_erases[sector].saturating_add(1);
            }
            Ok(())
        }
    }

    fn ready_with_observer<Observe>(
        observe: Observe,
    ) -> EmbeddedFlashPersistence<TestFlash, FixedRouteSnapshotKeys<8>, Observe, 4>
    where
        Observe: FnMut(EmbeddedPersistenceDiagnostic),
    {
        embassy_futures::block_on(async {
            let flash = TestFlash::new();
            let mut scratch = [0u8; RECORD_SCRATCH_LEN];
            let (mut journal, _) = FlashJournal::open(flash, LAYOUT, &mut scratch, |_| {})
                .await
                .unwrap();
            journal.initialize_empty().await.unwrap();
            journal
                .record_compaction_budget(InstantMillis(0))
                .await
                .unwrap();
            let mut persistence = EmbeddedFlashPersistence::new(
                TestFlash::new(),
                LAYOUT,
                EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(0)),
                FixedRouteSnapshotKeys::new(),
                observe,
            );
            persistence.flash = None;
            persistence.journal = Some(journal);
            persistence.next_compaction_not_before =
                Some(InstantMillis(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS));
            persistence
        })
    }

    fn ready() -> EmbeddedFlashPersistence<
        TestFlash,
        FixedRouteSnapshotKeys<8>,
        fn(EmbeddedPersistenceDiagnostic),
        4,
    > {
        ready_with_observer((|_| {}) as fn(EmbeddedPersistenceDiagnostic))
    }

    fn signed_route(secret: u8, app_data: &[u8]) -> crate::routing::PersistedRouteRow<'_> {
        use crate::identity::in_memory::InMemoryNodeIdentity;
        use crate::interfaces::InterfaceId;
        use crate::routing::announce::{Announce, AnnounceId, DottedNameHash};
        use crate::routing::routes::RouteEntry;
        use crate::routing::{AnnounceIdRing, NextHop, RouteResponsiveness};

        let signer = InMemoryNodeIdentity::from_secret_key_bytes(&[secret; 64]);
        let announce = Announce::build_signed(
            &signer,
            DottedNameHash::new([secret; 10]),
            AnnounceId::from_wire([secret.wrapping_add(1); 10]),
            None,
            app_data,
        )
        .unwrap();
        crate::routing::PersistedRouteRow {
            destination: announce.destination,
            entry: RouteEntry {
                hops: secret,
                learned_at: InstantMillis(500),
                last_relayed_at: InstantMillis(700),
                responsiveness: RouteResponsiveness::Responsive,
                receiving_interface: InterfaceId::new([secret; 8]),
                next_hop: NextHop::Direct,
            },
            public_keys: announce.public_keys,
            dotted_name_hash: announce.dotted_name_hash,
            announce_id: announce.announce_id,
            ratchet: announce.ratchet,
            signature: announce.signature,
            app_data,
            announce_id_ring: AnnounceIdRing::Wire(
                &[0; crate::routing::announce::ANNOUNCE_ID_WIRE_LEN],
            ),
        }
    }

    #[test]
    fn exact_route_and_ratchet_deadlines_are_distinct() {
        let policy =
            EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(64));
        assert_eq!(policy.first_route_commit_delay_millis, 2_000);
        assert_eq!(policy.minimum_route_commit_interval_millis, 300_000);
        assert_eq!(policy.ratchet_batch_delay_millis, 2_000);
        assert_eq!(policy.retry_interval_millis, 300_000);
        assert_eq!(
            policy.timebase_record_interval_millis,
            TIMEBASE_RECORD_INTERVAL_MILLIS
        );
        assert_eq!(
            policy.compaction.minimum_interval_millis,
            HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS
        );
        assert_eq!(policy.compaction.critical_reserve_bytes, 64);
    }

    #[test]
    fn deadline_formula_batches_first_write_then_honors_five_minutes() {
        let mut persistence = ready();
        persistence.route_dirty_since = Some(InstantMillis(1_000));
        assert_eq!(
            persistence.next_deadline(InstantMillis(1_500)),
            Some(InstantMillis(3_000))
        );
        persistence.last_route_success = Some(InstantMillis(10_000));
        persistence.route_dirty_since = Some(InstantMillis(11_000));
        assert_eq!(
            persistence.next_deadline(InstantMillis(12_000)),
            Some(InstantMillis(310_000))
        );
        persistence.ratchet_dirty_since = Some(InstantMillis(12_000));
        assert_eq!(
            persistence.next_deadline(InstantMillis(12_000)),
            Some(InstantMillis(14_000))
        );
        persistence.retry_not_before = Some(InstantMillis(400_000));
        assert_eq!(
            persistence.next_deadline(InstantMillis(12_000)),
            Some(InstantMillis(400_000))
        );
    }

    #[test]
    fn repeated_route_and_ratchet_changes_coalesce_by_destination() {
        let mut persistence = ready();
        let destination = DestinationHash::new([0x11; TRUNCATED_HASH_BYTE_LEN]);
        persistence.queue_route(
            PendingRouteDelta::RouteUpsert(destination),
            InstantMillis(0),
        );
        persistence.queue_route(
            PendingRouteDelta::RouteUpsert(destination),
            InstantMillis(0),
        );
        persistence.queue_route(
            PendingRouteDelta::RouteRemoval(destination),
            InstantMillis(0),
        );
        persistence.queue_ratchet(destination, InstantMillis(0));
        persistence.queue_ratchet(destination, InstantMillis(0));
        assert_eq!(
            persistence.pending_routes.as_slice(),
            &[PendingRouteDelta::RouteRemoval(destination)]
        );
        assert_eq!(persistence.pending_ratchets.as_slice(), &[destination]);
    }

    #[test]
    fn pending_overflow_waits_for_the_batch_deadline_before_compacting() {
        let mut persistence = ready();
        for byte in 0..5 {
            persistence.queue_route(
                PendingRouteDelta::RouteUpsert(DestinationHash::new(
                    [byte; TRUNCATED_HASH_BYTE_LEN],
                )),
                InstantMillis(1_000),
            );
        }
        persistence.route_dirty_since = Some(InstantMillis(1_000));
        assert!(persistence.snapshot_required);
        assert_eq!(
            persistence.deferred_target,
            Some(EmbeddedPersistenceTarget::Routes)
        );
        assert_eq!(
            persistence.next_deadline(InstantMillis(1_500)),
            Some(InstantMillis(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS))
        );

        let mut engine = EngineState::<crate::storage::GrowableHeap>::default();
        embassy_futures::block_on(persistence.progress(&mut engine, InstantMillis(2_999)));
        assert_eq!(persistence.compaction, None);
        assert!(persistence.snapshot_required);

        embassy_futures::block_on(persistence.progress(
            &mut engine,
            InstantMillis(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS),
        ));
        assert_eq!(
            persistence.compaction,
            Some(CompactionPhase::RecordBudget {
                at: InstantMillis(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS)
            })
        );
    }

    #[test]
    fn failures_keep_dirty_state_and_raise_the_notice() {
        let mut persistence = ready();
        let destination = DestinationHash::new([0x22; TRUNCATED_HASH_BYTE_LEN]);
        persistence.queue_route(
            PendingRouteDelta::RouteUpsert(destination),
            InstantMillis(0),
        );
        persistence.note_codec_failure(InstantMillis(1_000));
        assert_eq!(persistence.pending_routes.len(), 1);
        assert_eq!(persistence.retry_not_before, Some(InstantMillis(301_000)));
        assert!(persistence.state_not_saved());
        assert!(!persistence.snapshot_required);

        persistence.retry_not_before = None;
        persistence.compaction = Some(CompactionPhase::Commit);
        persistence.compaction_target = Some(EmbeddedPersistenceTarget::Routes);
        persistence.note_write_failure(InstantMillis(2_000), EmbeddedPersistenceFailure::Flash);
        assert_eq!(persistence.pending_routes.len(), 0);
        assert_eq!(persistence.retry_not_before, Some(InstantMillis(302_000)));
        assert!(persistence.state_not_saved());
        assert!(persistence.snapshot_required);
        assert_eq!(persistence.compaction, None);
    }

    #[test]
    fn legacy_timebase_allows_one_needed_compaction_then_adopts_the_budget_marker() {
        embassy_futures::block_on(async {
            let (mut journal, _) = {
                let flash = TestFlash::new();
                let mut scratch = [0u8; RECORD_SCRATCH_LEN];
                FlashJournal::open(flash, LAYOUT, &mut scratch, |_| {})
                    .await
                    .unwrap()
            };
            journal.initialize_empty().await.unwrap();
            journal
                .record_timebase(InstantMillis(10_000))
                .await
                .unwrap();
            let mut persistence =
                EmbeddedFlashPersistence::<_, FixedRouteSnapshotKeys<8>, _, 4>::new(
                    journal.release(),
                    LAYOUT,
                    EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(
                        0,
                    )),
                    FixedRouteSnapshotKeys::new(),
                    (|_| {}) as fn(EmbeddedPersistenceDiagnostic),
                );
            let mut engine = EngineState::<crate::storage::GrowableHeap>::default();
            let report = persistence.restore(&mut engine, InstantMillis(0)).await;
            assert_eq!(persistence.next_compaction_not_before, None);
            persistence.require_snapshot(EmbeddedPersistenceTarget::Routes, report.logical_start);
            persistence.route_dirty_since = Some(InstantMillis(report.logical_start.0 - 2_000));
            persistence
                .progress(&mut engine, report.logical_start)
                .await;
            persistence
                .progress(&mut engine, report.logical_start)
                .await;
            assert!(matches!(
                persistence.compaction,
                Some(CompactionPhase::Erase { sector: 0 })
            ));

            let flash = persistence.journal.take().unwrap().release();
            let mut restored = EmbeddedFlashPersistence::<_, FixedRouteSnapshotKeys<8>, _, 4>::new(
                flash,
                LAYOUT,
                EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(0)),
                FixedRouteSnapshotKeys::new(),
                (|_| {}) as fn(EmbeddedPersistenceDiagnostic),
            );
            restored.restore(&mut engine, InstantMillis(0)).await;
            assert!(restored.next_compaction_not_before.is_some());
        });
    }

    #[test]
    fn failed_compaction_attempt_consumes_the_daily_budget() {
        let diagnostics = Rc::new(RefCell::new(Vec::new()));
        let observed = Rc::clone(&diagnostics);
        let mut persistence = ready_with_observer(move |diagnostic| {
            observed.borrow_mut().push(diagnostic);
        });
        let mut engine = EngineState::<crate::storage::GrowableHeap>::default();
        let first = InstantMillis(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS);
        persistence.require_snapshot(EmbeddedPersistenceTarget::Routes, first);
        persistence.route_dirty_since = Some(InstantMillis(first.0 - 2_000));
        embassy_futures::block_on(persistence.progress(&mut engine, first));
        let second = InstantMillis(first.0 + HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS);
        assert_eq!(persistence.next_compaction_not_before, Some(first));
        embassy_futures::block_on(persistence.progress(&mut engine, first));
        assert_eq!(persistence.next_compaction_not_before, Some(second));
        assert_eq!(
            persistence.compaction,
            Some(CompactionPhase::Erase { sector: 0 })
        );

        persistence.note_write_failure(first, EmbeddedPersistenceFailure::Flash);
        assert_eq!(persistence.compaction, None);
        assert!(persistence.snapshot_required);
        assert_eq!(persistence.next_deadline(first), Some(second));
        embassy_futures::block_on(persistence.progress(&mut engine, InstantMillis(second.0 - 1)));
        assert_eq!(persistence.compaction, None);
        embassy_futures::block_on(persistence.progress(&mut engine, second));
        assert_eq!(
            persistence.compaction,
            Some(CompactionPhase::RecordBudget { at: second })
        );
        embassy_futures::block_on(persistence.progress(&mut engine, second));
        assert_eq!(
            persistence.compaction,
            Some(CompactionPhase::Erase { sector: 0 })
        );

        let starts = diagnostics
            .borrow()
            .iter()
            .filter(|diagnostic| {
                matches!(
                    diagnostic,
                    EmbeddedPersistenceDiagnostic::CompactionStarted { .. }
                )
            })
            .count();
        assert_eq!(starts, 2);
    }

    #[test]
    fn recorded_compaction_budget_survives_reboot() {
        embassy_futures::block_on(async {
            let mut persistence = ready();
            let mut engine = EngineState::<crate::storage::GrowableHeap>::default();
            let attempt = InstantMillis(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS);
            persistence.require_snapshot(EmbeddedPersistenceTarget::Routes, attempt);
            persistence.route_dirty_since = Some(InstantMillis(attempt.0 - 2_000));
            persistence.progress(&mut engine, attempt).await;
            persistence.progress(&mut engine, attempt).await;
            assert_eq!(
                persistence.compaction,
                Some(CompactionPhase::Erase { sector: 0 })
            );

            let flash = persistence.journal.take().unwrap().release();
            let mut restored = EmbeddedFlashPersistence::<_, FixedRouteSnapshotKeys<8>, _, 4>::new(
                flash,
                LAYOUT,
                EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(0)),
                FixedRouteSnapshotKeys::new(),
                (|_| {}) as fn(EmbeddedPersistenceDiagnostic),
            );
            let report = restored.restore(&mut engine, InstantMillis(0)).await;
            assert!(report.logical_start.0 >= attempt.0);
            assert_eq!(
                restored.next_compaction_not_before,
                Some(InstantMillis(
                    attempt
                        .0
                        .saturating_add(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS)
                ))
            );
        });
    }

    #[test]
    fn marker_write_failure_does_not_consume_the_compaction_budget() {
        embassy_futures::block_on(async {
            let (flash, fail_next_write) = TestFlash::controlled();
            let mut scratch = [0u8; RECORD_SCRATCH_LEN];
            let (mut journal, _) = FlashJournal::open(flash, LAYOUT, &mut scratch, |_| {})
                .await
                .unwrap();
            journal.initialize_empty().await.unwrap();
            let mut persistence =
                EmbeddedFlashPersistence::<_, FixedRouteSnapshotKeys<8>, _, 4>::new(
                    TestFlash::new(),
                    LAYOUT,
                    EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(
                        0,
                    )),
                    FixedRouteSnapshotKeys::new(),
                    (|_| {}) as fn(EmbeddedPersistenceDiagnostic),
                );
            persistence.flash = None;
            persistence.journal = Some(journal);
            let mut engine = EngineState::<crate::storage::GrowableHeap>::default();
            let attempt = InstantMillis(2_000);
            persistence.require_snapshot(EmbeddedPersistenceTarget::Routes, attempt);
            persistence.route_dirty_since = Some(InstantMillis(0));
            persistence.progress(&mut engine, attempt).await;
            fail_next_write.set(true);
            persistence.progress(&mut engine, attempt).await;
            assert_eq!(persistence.next_compaction_not_before, None);
            assert_eq!(persistence.compaction, None);

            let flash = persistence.journal.take().unwrap().release();
            let mut restored = EmbeddedFlashPersistence::<_, FixedRouteSnapshotKeys<8>, _, 4>::new(
                flash,
                LAYOUT,
                EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(0)),
                FixedRouteSnapshotKeys::new(),
                (|_| {}) as fn(EmbeddedPersistenceDiagnostic),
            );
            restored.restore(&mut engine, InstantMillis(0)).await;
            assert_eq!(restored.next_compaction_not_before, None);
        });
    }

    #[test]
    fn timebase_writes_and_repeated_reboots_do_not_move_the_compaction_deadline() {
        embassy_futures::block_on(async {
            let mut persistence = ready();
            let deadline = persistence.next_compaction_not_before;
            persistence
                .journal
                .as_mut()
                .unwrap()
                .record_timebase(InstantMillis(3 * 60 * 60 * 1_000))
                .await
                .unwrap();
            let flash = persistence.journal.take().unwrap().release();
            let mut engine = EngineState::<crate::storage::GrowableHeap>::default();

            let mut first = EmbeddedFlashPersistence::<_, FixedRouteSnapshotKeys<8>, _, 4>::new(
                flash,
                LAYOUT,
                EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(0)),
                FixedRouteSnapshotKeys::new(),
                (|_| {}) as fn(EmbeddedPersistenceDiagnostic),
            );
            first.restore(&mut engine, InstantMillis(0)).await;
            assert_eq!(first.next_compaction_not_before, deadline);

            let flash = first.journal.take().unwrap().release();
            let mut second = EmbeddedFlashPersistence::<_, FixedRouteSnapshotKeys<8>, _, 4>::new(
                flash,
                LAYOUT,
                EmbeddedPersistencePolicy::hopspot_default(EmbeddedCompactionPolicy::hopspot(0)),
                FixedRouteSnapshotKeys::new(),
                (|_| {}) as fn(EmbeddedPersistenceDiagnostic),
            );
            second.restore(&mut engine, InstantMillis(0)).await;
            assert_eq!(second.next_compaction_not_before, deadline);
        });
    }

    #[test]
    fn overflow_during_compaction_commits_once_and_defers_the_next_snapshot() {
        let diagnostics = Rc::new(RefCell::new(Vec::new()));
        let observed = Rc::clone(&diagnostics);
        let mut persistence = ready_with_observer(move |diagnostic| {
            observed.borrow_mut().push(diagnostic);
        });
        let mut engine = EngineState::<crate::storage::GrowableHeap>::default();
        let now = InstantMillis(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS);
        persistence.require_snapshot(EmbeddedPersistenceTarget::Routes, now);
        persistence.route_dirty_since = Some(InstantMillis(now.0 - 2_000));
        embassy_futures::block_on(persistence.progress(&mut engine, now));

        for byte in 0..6 {
            persistence.queue_route(
                PendingRouteDelta::RouteUpsert(DestinationHash::new(
                    [byte; TRUNCATED_HASH_BYTE_LEN],
                )),
                now,
            );
        }
        persistence.route_dirty_since = Some(now);
        for _ in 0..8 {
            embassy_futures::block_on(persistence.progress(&mut engine, now));
        }

        assert_eq!(persistence.compaction, None);
        assert!(persistence.snapshot_required);
        assert!(persistence.state_not_saved());
        assert_eq!(
            persistence.next_deadline(now),
            Some(InstantMillis(
                now.0 + HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS
            ))
        );
        let diagnostics = diagnostics.borrow();
        assert_eq!(
            diagnostics
                .iter()
                .filter(|diagnostic| matches!(
                    diagnostic,
                    EmbeddedPersistenceDiagnostic::CompactionStarted { .. }
                ))
                .count(),
            1
        );
        assert_eq!(
            diagnostics
                .iter()
                .filter(|diagnostic| matches!(
                    diagnostic,
                    EmbeddedPersistenceDiagnostic::CompactionCompleted { .. }
                ))
                .count(),
            1
        );
        assert_eq!(
            diagnostics
                .iter()
                .filter(|diagnostic| matches!(
                    diagnostic,
                    EmbeddedPersistenceDiagnostic::DurabilityDeferred { .. }
                ))
                .count(),
            1
        );
    }

    #[test]
    fn captured_route_keys_survive_slot_shifts_and_new_routes_land_after_compaction() {
        embassy_futures::block_on(async {
            let mut persistence = ready();
            let mut engine = EngineState::<crate::storage::GrowableHeap>::default();
            let rows = [signed_route(0x31, &[0xA1]), signed_route(0x32, &[0xA2])];
            for row in &rows {
                assert_eq!(
                    engine.seed_route(row, InstantMillis(1_000)),
                    RouteSeedOutcome::Seeded
                );
            }
            let now = InstantMillis(HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS);
            persistence.require_snapshot(EmbeddedPersistenceTarget::Routes, now);
            persistence.route_dirty_since = Some(InstantMillis(now.0 - 2_000));
            persistence.progress(&mut engine, now).await;
            persistence.progress(&mut engine, now).await;
            persistence.progress(&mut engine, now).await;

            let removed = rows[0].destination;
            let retained = rows[1].destination;
            let _ = engine.drop_route(&removed, AttachedInterfaces::new(&[]));
            let added = signed_route(0x34, &[0xA4]);
            assert_eq!(engine.seed_route(&added, now), RouteSeedOutcome::Seeded);
            persistence.queue_route(PendingRouteDelta::RouteUpsert(added.destination), now);
            persistence.route_dirty_since = Some(now);

            for _ in 0..8 {
                persistence.progress(&mut engine, now).await;
            }
            assert_eq!(persistence.compaction, None);
            assert_eq!(
                (
                    persistence.pending_routes.len(),
                    persistence.snapshot_required,
                    persistence.write_failed,
                    persistence.route_dirty_since,
                    persistence.landing_batch,
                ),
                (1, false, false, Some(now), None)
            );
            let correction_at = InstantMillis(now.0 + 2_000);
            persistence.progress(&mut engine, correction_at).await;
            persistence.progress(&mut engine, correction_at).await;

            let flash = persistence.journal.take().unwrap().release();
            let mut scratch = [0u8; RECORD_SCRATCH_LEN];
            let mut restored = Vec::new();
            let _ = FlashJournal::open(flash, LAYOUT, &mut scratch, |record| {
                if record.kind != FlashJournalRecordKind::RouteUpsert {
                    return;
                }
                let mut rows = read_routing_table_snapshot(record.payload).unwrap();
                restored.push(rows.next().unwrap().unwrap().destination);
            })
            .await
            .unwrap();
            assert_eq!(restored, vec![retained, added.destination]);
        });
    }

    #[test]
    fn sixteen_route_records_restore_eight_and_report_capacity_drops() {
        type EightRouteStorage =
            crate::storage::TestFixedStorage<8, 8, 256, 2, 2, 16, 4, 4, 4, 4, 4, 16>;
        let now = InstantMillis(1_000);
        let mut engine = EngineState::<EightRouteStorage>::default();
        let mut report = EmbeddedPersistenceRestoreReport {
            logical_start: now,
            route_seeded_count: 0,
            route_refused_count: 0,
            route_dropped_count: 0,
            ratchet_seeded_count: 0,
            ratchet_refused_count: 0,
            warning: None,
        };

        for secret in 1..=16 {
            let row = signed_route(secret, &[]);
            let required = routing_table_snapshot_len(core::iter::once(row.clone()));
            let mut scratch = [0u8; RECORD_SCRATCH_LEN];
            let written =
                write_routing_table_snapshot(core::iter::once(row), &mut scratch[..required])
                    .unwrap();
            apply_record(
                &mut engine,
                now,
                FlashJournalRecord {
                    epoch: 1,
                    kind: FlashJournalRecordKind::RouteUpsert,
                    payload: &scratch[..written],
                },
                &mut report,
            );
        }

        assert_eq!(
            report,
            EmbeddedPersistenceRestoreReport {
                logical_start: now,
                route_seeded_count: 8,
                route_refused_count: 0,
                route_dropped_count: 8,
                ratchet_seeded_count: 0,
                ratchet_refused_count: 0,
                warning: None,
            }
        );
    }

    #[test]
    fn thirty_days_of_pressure_erase_each_arena_sector_at_most_fifteen_times() {
        let diagnostics = Rc::new(RefCell::new(Vec::new()));
        let observed = Rc::clone(&diagnostics);
        let mut persistence = ready_with_observer(move |diagnostic| {
            observed.borrow_mut().push(diagnostic);
        });
        let mut engine = EngineState::<crate::storage::GrowableHeap>::default();
        embassy_futures::block_on(async {
            for day in 1..=30 {
                let now = InstantMillis(day * HOPSPOT_MINIMUM_COMPACTION_INTERVAL_MILLIS);
                persistence.require_snapshot(EmbeddedPersistenceTarget::Routes, now);
                persistence.route_dirty_since = Some(InstantMillis(now.0 - 2_000));
                for _ in 0..8 {
                    persistence.progress(&mut engine, now).await;
                }
                assert_eq!(persistence.compaction, None);
            }
        });
        assert_eq!(
            diagnostics
                .borrow()
                .iter()
                .filter(|diagnostic| matches!(
                    diagnostic,
                    EmbeddedPersistenceDiagnostic::CompactionStarted { .. }
                ))
                .count(),
            30
        );
        let flash = persistence.journal.take().unwrap().release();
        assert_eq!(
            [
                flash.sector_erases[2] - 1,
                flash.sector_erases[3] - 1,
                flash.sector_erases[4],
                flash.sector_erases[5],
            ],
            [15, 15, 15, 15]
        );
    }
}