fix: restore processed signatures on restart

engine/README.md

@@ -45,7 +45,8 @@ quiesces at superblock seals and compares the reconstructed checksum with the
 recorded seal. A mismatch returns `ReplayError::StateMismatch`. Current state
 opens without replay when its slot and transaction count are each at least the
 ledger values. After replay actually runs, the final transaction counts must be
-equal or startup returns `ReplayError::StateMismatch`.
+equal or startup returns `ReplayError::StateMismatch`. Replay caches each
+re-executed terminal transaction result without appending it to the ledger.
 
 Internal pacing appends one reset marker at the current slot and clears
 chain-mirrored volatile accounts before the pacemaker task starts. Internal

engine/tests/recovery.rs

@@ -11,10 +11,12 @@
 use std::{path::PathBuf, time::Duration};
 
 use engine::{EngineError, ReplayError, testkit::TestEngine};
-use keeper::testkit::{corrupt, load_v42_data, store_v42};
+use keeper::testkit::{corrupt, load_v42_data, signed_view, store_v42};
 use nucleus::ledger::ACCOUNTSDB_SNAPSHOT_FILE;
 use solana_account::AccountMode;
 use solana_pubkey::Pubkey;
+use solana_signature::Signature;
+use solana_transaction::TransactionError;
 use tokio::time;
 use v42_calculator_interface::builder::Expr as E;
 
@@ -25,10 +27,44 @@ async fn commit_and_seal(te: &mut TestEngine, key: Pubkey, value: i64) -> PathBu
     te.seal_and_archive().await
 }
 
-// Replay must rebuild everything between the restored snapshot and the ledger
-// tip: dropping superblock 2's archive forces the restore back onto snapshot 1,
-// so re-executing B crosses superblock 2's sealed checksum (the verification
-// arm's happy path) before C is rebuilt from the unsealed head.
+/// Verifies a startup-restored terminal status rejects the original bytes.
+async fn assert_restored_signature(
+    te: &TestEngine,
+    key: Pubkey,
+    signature: Signature,
+    transaction: Vec<u8>,
+    value: i64,
+) {
+    let status = te
+        .transactions()
+        .status(signature)
+        .await
+        .expect("status lookup succeeds")
+        .expect("terminal status is restored");
+    assert!(
+        status.result.is_ok(),
+        "successful terminal status is available"
+    );
+
+    let result = te
+        .transaction(transaction)
+        .expect("persisted transaction remains valid")
+        .execute()
+        .await
+        .expect("duplicate submission returns a terminal status");
+    assert_eq!(result, Err(TransactionError::AlreadyProcessed));
+    assert_eq!(
+        load_v42_data(te, key),
+        Some(value),
+        "duplicate transaction was not executed again"
+    );
+}
+
+/// Proves snapshot-tail replay rebuilds state and refreshes processed signatures.
+///
+/// Dropping superblock 2's archive forces the restore back onto snapshot 1, so
+/// re-executing B crosses superblock 2's sealed checksum before C is rebuilt
+/// from the unsealed head. C must then remain deduplicated after replay.
 #[tokio::test(flavor = "multi_thread")]
 async fn replay_rebuilds_state_after_counter_lag() {
     let mut te = TestEngine::new().await;
@@ -41,10 +77,13 @@ async fn replay_rebuilds_state_after_counter_lag() {
         s1.ends_with(ACCOUNTSDB_SNAPSHOT_FILE),
         "archive is the compressed accountsdb tarball"
     );
-    // B: K = 20 sealed into superblock 2; C: K = 30 lives only in the ledger's
-    // unsealed head, past every archived snapshot.
+    // B: K = 20 sealed into superblock 2; C increments K and lives only in the
+    // ledger's unsealed head, past every archived snapshot.
     let s2 = commit_and_seal(&mut te, key, 20).await;
-    te.execute(&[E::lit(30).compose(key, &[])]).await.expect("C commits");
+    let (signature, transaction) =
+        signed_view(&te, None, (E::acc(0) + E::lit(1)).compose(key, &[]));
+    let transaction = transaction.inner_data().as_ref().clone();
+    te.execute(transaction.clone()).await.expect("C commits");
     te.advance(2).await;
     let (dirs, authority) = te.close().await;
 
@@ -59,9 +98,10 @@ async fn replay_rebuilds_state_after_counter_lag() {
     let te2 = TestEngine::with(dirs, authority).await;
     assert_eq!(
         load_v42_data(&te2, key),
-        Some(30),
+        Some(21),
         "both post-snapshot mutations were rebuilt purely from ledger replay"
     );
+    assert_restored_signature(&te2, key, signature, transaction, 21).await;
     // The temporary replay sequencer must hand off to a working live one.
     te2.execute(&[E::lit(1).compose(key, &[])])
         .await
@@ -100,11 +140,11 @@ async fn replay_aborts_on_checksum_mismatch() {
     );
 }
 
-// A healthy restart opens persisted state as-is and restores the clean-shutdown
-// volatile dump. A failed execution still counts on both durable sides without
-// writing accounts. The direct-stored delegated account exists in neither
-// snapshots nor ledger, while the read-only account exists only in the volatile
-// dump; the post-seal transaction write pins the persisted tip alongside them.
+/// Proves a clean restart restores processed signatures without re-execution.
+///
+/// Persisted state reopens as-is with the clean-shutdown volatile dump. A failed
+/// execution still counts on both durable sides without writing accounts. The
+/// exact successful transaction remains terminal and is rejected on resubmission.
 #[tokio::test(flavor = "multi_thread")]
 async fn clean_restart_reopens_persisted_and_volatile_state() {
     let mut te = TestEngine::new().await;
@@ -121,16 +161,22 @@ async fn clean_restart_reopens_persisted_and_volatile_state() {
         Some(20),
         "failed execution writes no state"
     );
+    let (signature, transaction) =
+        signed_view(&te, None, (E::acc(0) + E::lit(1)).compose(key, &[]));
+    let transaction = transaction.inner_data().as_ref().clone();
+    te.execute(transaction.clone()).await.expect("increment commits");
+    te.advance(1).await;
     let direct = store_v42(&te, 7, AccountMode::Delegated);
     let volatile = store_v42(&te, 8, AccountMode::ReadOnly);
     let (dirs, authority) = te.close().await;
 
     let te2 = TestEngine::with(dirs, authority).await;
     assert_eq!(
         load_v42_data(&te2, key),
-        Some(20),
+        Some(21),
         "persisted tip state reopened as-is"
     );
+    assert_restored_signature(&te2, key, signature, transaction, 21).await;
     assert_eq!(
         load_v42_data(&te2, direct),
         Some(7),

keeper/README.md

@@ -24,11 +24,16 @@ non-zero superblock interval used by pacing and cache TTL calculation. The
 shared accountsdb, blockstore, and ledger parameters are defined by nucleus;
 keeper consumes them when opening its durable stores and caches.
 
-Startup reconstructs the recent-blockhash cache from retained ledger blocks for
-the configured 60-second validity window, bounded by the accountsdb slot so an
-unreplayed ledger tail cannot advance startup state. Persisted `SlotHashes`
-supplies the newest entries, extended with older ledger hashes when available;
-snapshot bootstrap without ledger history uses `SlotHashes` alone.
+Startup issues one retained-history read, ending at the accountsdb slot and
+covering the largest of the block cache, signature cache, and `SlotHashes`
+windows. An unreplayed ledger tail therefore cannot advance startup state.
+Persisted `SlotHashes` supplies the newest block entries, extended with older
+ledger hashes when available; snapshot bootstrap without ledger history uses
+`SlotHashes` alone. Indexed terminal transaction statuses seed the processed
+signature cache at their original block slots. If Engine startup replay advances
+the authoritative block boundary, each re-executed transaction caches its
+terminal result without appending another ledger record or publishing live
+subscriptions.
 
 ## Authority
 

keeper/src/accessor.rs

@@ -30,11 +30,11 @@ use solana_transaction_error::TransactionError;
 use tokio::sync::mpsc::Receiver;
 
 use crate::{
-    FullTransaction, Keeper, ResolvedTransaction,
+    ExecutionRecord, FullTransaction, Keeper, ResolvedTransaction,
     cache::{AccountCache, MissingAccount},
     error::Result,
     subscriptions::TransactionLogs,
-    util::{execution_commit, request},
+    util::{execution_commit, request, transaction_status},
 };
 
 /// Account operations namespace.
@@ -259,9 +259,24 @@ impl<'a> TransactionsAccessor<'a> {
         Ok(())
     }
 
+    /// Commits replayed state and caches its re-executed terminal status.
+    ///
+    /// Replay does not append ledger records or publish live subscriptions.
+    pub fn commit_replay(
+        &self,
+        transaction: &ResolvedTransaction,
+        execution: &ExecutionRecord,
+    ) -> Result<()> {
+        self.commit_state_transitions(&execution.result)?;
+        let signature = transaction.signatures()[0];
+        let status = transaction_status(&execution.result, execution.slot);
+        self.keeper.caches.signatures.push(signature, Some(status), execution.slot);
+        Ok(())
+    }
+
     /// Commits one accepted transaction to accountsdb, writing dirty accounts
     /// only for successful execution and returning it for downstream fanout.
-    pub fn commit_state_transitions<'t>(
+    fn commit_state_transitions<'t>(
         &self,
         result: &'t TransactionProcessingResult,
     ) -> Result<Option<&'t ExecutedTransaction>> {

keeper/src/builder.rs

@@ -3,7 +3,6 @@
 use std::{
     collections::HashMap,
     fs::{self, File},
-    sync::Arc,
     time::Duration,
 };
 
@@ -16,7 +15,7 @@ use ledger::{
 use nucleus::{
     Slot,
     config::{AccountsDBParams, Authority, BlockstoreParams, LedgerParams},
-    ledger::{ACCOUNTSDB_SNAPSHOT_FILE, Block},
+    ledger::ACCOUNTSDB_SNAPSHOT_FILE,
     shutdown::ShutdownManager,
 };
 use serde::Serialize;
@@ -39,7 +38,7 @@ use tracing::{error, info, warn};
 
 use crate::{
     Keeper,
-    cache::{AccountCache, BlockSeed, BlocksCache, Caches, ExpiringCache},
+    cache::{CacheSeed, Caches},
     error::Result,
     metrics,
     subscriptions::Subscriptions,
@@ -80,8 +79,8 @@ impl KeeperBuilder {
     pub async fn build(mut self, shutdown: &mut ShutdownManager) -> Result<Keeper> {
         let ledger = Ledger::init(&self.ledger.directory, self.ledger.size_limit, shutdown)?;
         let accountsdb = self.accountsdb(&ledger)?;
-        let (blocks, featureset) = self.prepopulate(&accountsdb, &ledger).await?;
-        let caches = self.caches(blocks);
+        let (seed, featureset) = self.prepopulate(&accountsdb, &ledger).await?;
+        let caches = self.caches(seed);
         metrics::init();
         Ok(Keeper {
             authority: self.authority,
@@ -99,11 +98,11 @@ impl KeeperBuilder {
         &mut self,
         accountsdb: &AccountsDB,
         ledger: &LedgerHandle,
-    ) -> Result<(BlockSeed, FeatureSet)> {
+    ) -> Result<(CacheSeed, FeatureSet)> {
         let mut accounts = Vec::new();
         let featureset = self.seed_featureset(&mut accounts)?;
         self.seed_programs(&mut accounts)?;
-        let blocks = self.seed_sysvars(accountsdb, ledger, &mut accounts).await?;
+        let seed = self.seed_sysvars(accountsdb, ledger, &mut accounts).await?;
         let authority = self.authority.pubkey();
         if accountsdb.loader().load(&authority)?.is_none() {
             let sponsor = AccountBuilder::default()
@@ -113,16 +112,17 @@ impl KeeperBuilder {
         }
         accounts.extend(self.accounts.drain());
         accountsdb.store(&accounts)?;
-        Ok((blocks, featureset))
+        Ok((seed, featureset))
     }
 
     /// Builds read-side caches using blocktime-derived slot TTLs.
-    fn caches(&self, blocks: BlockSeed) -> Caches {
-        let blocks = BlocksCache::new(blocks, self.ttl(BLOCK_CACHE_WINDOW));
-        let signatures = ExpiringCache::new(self.ttl(SIGNATURE_CACHE_WINDOW));
-        let accounts = Arc::new(AccountCache::new(self.accountsdb.lru_capacity));
-
-        Caches { signatures, blocks, accounts }
+    fn caches(&self, seed: CacheSeed) -> Caches {
+        Caches::new(
+            seed,
+            self.ttl(BLOCK_CACHE_WINDOW),
+            self.ttl(SIGNATURE_CACHE_WINDOW),
+            self.accountsdb.lru_capacity,
+        )
     }
 
     /// Converts a wall-clock cache window into whole configured block slots.
@@ -185,37 +185,41 @@ impl KeeperBuilder {
         accountsdb: &AccountsDB,
         ledger: &LedgerHandle,
         accounts: &mut Vec<AccountEntry>,
-    ) -> Result<BlockSeed> {
+    ) -> Result<CacheSeed> {
         let slot = accountsdb.slot();
         let loader = accountsdb.loader();
         let slothashes = loader
             .load(&SlotHashes::id())?
             .map(|account| account.deserialize_data::<SlotHashes>().map_err(AccountsDBError::from))
             .transpose()?;
 
-        let retained = self.ttl(BLOCK_CACHE_WINDOW).max(SLOTHASH_ENTRIES as Slot);
+        let retained = self
+            .ttl(BLOCK_CACHE_WINDOW)
+            .max(self.ttl(SIGNATURE_CACHE_WINDOW))
+            .max(SLOTHASH_ENTRIES as Slot);
         let start = slot.saturating_sub(retained - 1);
         let (payload, handle) = RequestPayload::new(start..slot.saturating_add(1));
         ledger.reader.send(ReadRequest::BlockRange(payload))?;
-        let blocks = handle.recv_timeout().await??;
+        let history = handle.recv_timeout().await??;
 
         if slothashes.is_none() {
             // Keep the sysvar account at its fixed serialized capacity so live
             // updates can replace entries without resizing the account.
             let mut hashes = SlotHashes::new(&[Default::default(); SLOTHASH_ENTRIES]);
-            for block in blocks.iter().take(SLOTHASH_ENTRIES) {
-                hashes.add(block.slot, block.hash);
+            for block in history.iter().take(SLOTHASH_ENTRIES) {
+                hashes.add(block.block.slot, block.block.hash);
             }
             let acc = self.account(&hashes, &sysvar::ID)?;
             accounts.push((SlotHashes::id(), acc.build()));
         }
 
-        let blocks = Self::block_seed(blocks, slothashes.as_ref());
+        let seed = CacheSeed::new(history, slothashes.as_ref());
 
         // Set the clock slot one ahead from the last
+        let latest = seed.latest();
         let clock = Clock {
-            slot: blocks.latest.slot + 1,
-            unix_timestamp: blocks.latest.time,
+            slot: latest.slot + 1,
+            unix_timestamp: latest.time,
             ..Default::default()
         };
         accounts.push((Clock::id(), self.account(&clock, &sysvar::ID)?.build()));
@@ -241,26 +245,7 @@ impl KeeperBuilder {
             EpochRewards::id(),
             self.account(&EpochRewards::default(), &sysvar::ID)?.build(),
         ));
-        Ok(blocks)
-    }
-
-    /// Extends persisted SlotHashes with older retained ledger history.
-    fn block_seed(blocks: Vec<Block>, slothashes: Option<&SlotHashes>) -> BlockSeed {
-        let Some(hashes) = slothashes.map(SlotHashes::slot_hashes) else {
-            let latest = blocks.first().copied().unwrap_or_default();
-            let history = blocks.iter().rev().map(|b| (b.slot, b.hash)).collect();
-            return BlockSeed { latest, history };
-        };
-        let mut history = hashes.to_vec();
-        history.extend(blocks.iter().skip(history.len()).map(|b| (b.slot, b.hash)));
-        let latest = history.first().map_or(Block::default(), |&(slot, hash)| Block {
-            slot,
-            hash,
-            time: blocks.first().map_or(0, |b| b.time),
-            parent: history.get(1).map(|(_, hash)| *hash).unwrap_or_default(),
-        });
-        history.reverse();
-        BlockSeed { latest, history }
+        Ok(seed)
     }
 
     /// Builds a rent-exempt system account containing a serialized sysvar-like state.