accessor.rs

//! Namespaced keeper access APIs.

use std::{
    cell::LazyCell,
    ops::{Deref, RangeInclusive},
    path::PathBuf,
    sync::Arc,
};

use accountsdb::{AccountEntry, AccountsDB, AccountsDBError};
use ledger::{
    LedgerRequestError,
    request::*,
    schema::{Event, Signed, SuperblockSeal, TransactionEntry, signature_prefix},
};
use nucleus::{
    Slot,
    ledger::{Block, BlockstorePosition},
    tls::{EncodedMessage, TlsManager},
};
use smallvec::SmallVec;
use solana_account::{AccountSharedData, CoWAccount};
use solana_hash::Hash;
use solana_pubkey::Pubkey;
use solana_sdk_ids::sysvar;
use solana_signature::Signature;
use solana_svm::{
    transaction_execution_result::ExecutedTransaction,
    transaction_processing_result::TransactionProcessingResult,
};
use solana_sysvar::{
    clock::Clock,
    slot_hashes::{SlotHashes, SysvarId},
};
use solana_transaction_error::{TransactionError, TransactionResult};
use tokio::sync::mpsc::{Receiver, UnboundedReceiver};

use crate::{
    AccountLease, ExecutionRecord, FullTransaction, Keeper, ResolvedTransaction,
    error::{KeeperError, Result},
    util::{execution_commit, request, transaction_status},
};

/// Account operations namespace.
pub struct AccountsAccessor<'a> {
    pub(crate) keeper: &'a Keeper,
}

impl<'a> AccountsAccessor<'a> {
    /// Promotes an existing recency entry without loading or pinning the account.
    pub fn touch(&self, pubkey: &Pubkey) {
        self.keeper.caches.accounts.lru.get_sync(pubkey);
    }

    /// Waits for exclusive mutation ownership of `pubkey` and records the
    /// account lifecycle observed after acquiring the lease.
    pub async fn lock(&self, pubkey: Pubkey) -> Result<AccountLease> {
        let mut lease = self.keeper.caches.accounts.lock(pubkey).await;
        lease.observed = self.loader().lifecycle(&pubkey)?;
        Ok(lease)
    }

    /// Returns recent transaction signatures that mention the account.
    pub async fn signatures(
        &self,
        params: AccountSignaturesParams,
    ) -> Result<Vec<AccountSignature>> {
        Ok(request(self.keeper, params, ReadRequest::AccountSignatures).await??)
    }

    /// Subscribes to updates for one account pubkey.
    pub fn subscribe(&self, account: Pubkey) -> Receiver<AccountSharedData> {
        self.keeper.subscriptions.accounts.subscribe(account)
    }

    /// Subscribes to account updates for accounts owned by `program`.
    pub fn subscribe_program(&self, program: Pubkey) -> Receiver<AccountEntry> {
        self.keeper.subscriptions.programs.subscribe(program)
    }

    /// Subscribes as the sole receiver of account pubkeys evicted from the recency cache.
    ///
    /// Returns an error if the process-lifetime eviction receiver was already registered.
    /// Notifications are unbounded so consumption cannot block materialization leases.
    pub fn subscribe_evictions(&self) -> Result<UnboundedReceiver<Pubkey>> {
        self.keeper.caches.accounts.evictions.subscribe()
    }

    /// Subscribes to completed accountsdb snapshot archives.
    pub fn subscribe_snapshots(&self) -> Receiver<PathBuf> {
        self.keeper.subscriptions.snapshots.subscribe(())
    }

    /// Updates stored `SlotHashes` and `Clock` sysvar accounts from `block`.
    ///
    /// Both sysvars must exist; fixed-capacity borrowed commits preserve their index offsets.
    pub fn update_sysvars(&self, block: Block) -> Result<()> {
        let loader = self.loader();

        // SAFETY: the sequencer is the exclusive sysvar writer and excludes compaction
        // through this block transition. The guarded loader outlives both borrowed images.
        let mut hacc = unsafe { loader.load(&SlotHashes::id()) }?
            .ok_or(KeeperError::MissingSysvar(SlotHashes::id()))?;
        let mut cacc =
            unsafe { loader.load(&Clock::id()) }?.ok_or(KeeperError::MissingSysvar(Clock::id()))?;
        let mut hashes: SlotHashes = hacc.deserialize_data().map_err(AccountsDBError::from)?;
        hashes.add(block.slot, block.hash);
        hacc.serialize_data(&hashes).map_err(AccountsDBError::from)?;
        let clock = self.keeper.clock(block);
        cacc.serialize_data(&clock).map_err(AccountsDBError::from)?;
        self.store(&[(SlotHashes::id(), hacc), (Clock::id(), cacc)]).map_err(Into::into)
    }
}

impl Deref for AccountsAccessor<'_> {
    type Target = AccountsDB;

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

/// Transaction operations namespace.
pub struct TransactionsAccessor<'a> {
    pub(crate) keeper: &'a Keeper,
}

impl<'a> TransactionsAccessor<'a> {
    /// Loads the full retained transaction for `signature`.
    pub async fn get(&self, signature: Signature) -> Result<Option<TransactionResponse>> {
        Ok(request(self.keeper, signature, ReadRequest::Transaction).await??)
    }

    /// Loads the retained execution status for `signature`.
    pub async fn status(&self, signature: Signature) -> Result<Option<TransactionStatus>> {
        if let Some(status) = self.keeper.caches.signatures.get(&signature_prefix(&signature)) {
            return Ok(status);
        }
        Ok(request(self.keeper, signature, ReadRequest::TransactionStatus).await??)
    }

    /// Subscribes as the sole receiver of all processed transactions, including failures.
    ///
    /// Returns an error if the process-lifetime transaction receiver was already registered.
    pub fn subscribe_processed(&self) -> Result<UnboundedReceiver<Arc<FullTransaction>>> {
        self.keeper.subscriptions.transactions.subscribe()
    }

    /// Subscribes to failed live transactions, including loading and execution failures.
    /// Admission rejections, simulation, and replay do not publish here.
    ///
    /// Each receiver buffers 32 transactions without blocking execution. Overflow
    /// disconnects that receiver; after draining, closure signals lost continuity.
    /// Resubscription observes future failures only, with no replay of missed events.
    pub fn subscribe_failures(&self) -> Receiver<Arc<FullTransaction>> {
        self.keeper.subscriptions.failures.subscribe(())
    }

    /// Subscribes to one execution result, including an already retained result.
    /// Admission rejections are request-specific and do not notify observers.
    pub async fn subscribe_signature(
        &self,
        signature: Signature,
    ) -> Result<oneshot::Receiver<TransactionStatus>> {
        let subscriptions = &self.keeper.subscriptions.signatures;
        let rx = subscriptions.subscribe(signature);
        // Register before reading: commit caches its result before fanout, so
        // racing publication is observed through either the channel or status.
        if let Some(status) = self.status(signature).await? {
            subscriptions.send(&signature, &status);
        }
        Ok(rx)
    }

    /// Subscribes to executed transactions mentioning `account`, including failures and their logs.
    pub async fn subscribe_logs(&self, account: Pubkey) -> Receiver<Arc<FullTransaction>> {
        self.keeper.subscriptions.logs.subscribe(account)
    }

    /// Subscribes as the sole receiver of encoded service messages.
    ///
    /// Returns an error if the process-lifetime service receiver was already registered.
    pub fn subscribe_service_messages(&self) -> Result<UnboundedReceiver<EncodedMessage>> {
        self.keeper.subscriptions.services.subscribe()
    }

    /// Appends transaction bytes to the ledger, deduplicating by the first 16
    /// signature bytes. Distinct signatures with the same live prefix are
    /// treated as already processed.
    ///
    /// The inner result reports admission; the outer result reports infrastructure
    /// failure. Rejections never publish or cache an execution result. A rejected
    /// signature remains reserved until normal expiry. Execution details are
    /// appended later by `commit_execution`.
    pub async fn append(&self, transaction: &ResolvedTransaction) -> Result<TransactionResult<()>> {
        let caches = &self.keeper.caches;
        let slot = caches.blocks.latest.load().slot + 1;
        let signature = transaction.signatures()[0];
        let key = signature_prefix(&signature);
        if !caches.signatures.push(key, None, slot) {
            return Ok(Err(TransactionError::AlreadyProcessed));
        }
        if !self.keeper.blocks().is_valid(transaction.recent_blockhash()) {
            return Ok(Err(TransactionError::BlockhashNotFound));
        }
        let event = Event::Transaction(TransactionEntry {
            signature,
            payload: transaction.inner_data().clone(),
        });
        self.keeper.ledger.appender.send_async(event).await?;
        Ok(Ok(()))
    }

    /// Commits execution metadata and publishes resulting account changes.
    /// Returns the committed result for the caller to complete its request.
    pub fn commit_execution(&self, mut txn: FullTransaction) -> Result<TransactionResult<()>> {
        let subs = &self.keeper.subscriptions;
        let success =
            txn.execution.result.as_ref().is_ok_and(|execution| execution.was_successful());
        // Class-level checks are atomic. Keyed maps are accessed only for delivery.
        let logs = txn.execution.result.is_ok() && !subs.logs.is_empty();
        let failures = !success && !subs.failures.is_empty();
        let transactions = subs.transactions.sender();
        let notify = logs || failures || transactions.is_some();
        let commit = execution_commit(&mut txn, notify);
        self.keeper.ledger.appender.send(commit.event)?;

        if let Some(execution) = self.commit_state_transitions(&txn.execution.result)? {
            subs.send_accounts(&execution.loaded_transaction.accounts);
            while let Some(msg) = TlsManager::dequeue() {
                subs.services.send(|| msg);
            }
        }
        if notify {
            // Keep routing keys separate so unmatched log subscriptions never
            // force account materialization or allocate the shared payload.
            let keys: SmallVec<[Pubkey; 8]> = if logs {
                SmallVec::from_slice(txn.transaction.static_account_keys())
            } else {
                SmallVec::new()
            };
            let txn = LazyCell::new(|| {
                // Queues outlive the execution barrier. Copy borrowed images
                // once, before sharing, while account ownership protects them.
                if let Ok(execution) = &mut txn.execution.result {
                    for (_, account) in &mut execution.loaded_transaction.accounts {
                        if matches!(account.cow(), CoWAccount::Borrowed(_)) {
                            *account = account.clone();
                        }
                    }
                }
                Arc::new(txn)
            });
            for key in keys {
                subs.logs.send_with(&key, || Arc::clone(&txn));
            }
            if failures {
                subs.failures.send_with(&(), || Arc::clone(&txn));
            }
            if let Some(sender) = transactions {
                let _ = sender.send(Arc::clone(&txn));
            }
        }
        // Clear TLS unconditionally so unsent messages cannot leak into the next transaction.
        TlsManager::clear();
        // Cache before fanout so a subscriber that misses delivery can still
        // discover the committed result.
        let key = &signature_prefix(&commit.signature);
        self.keeper.caches.signatures.update(key, Some(commit.status.clone()));
        subs.signatures.send(&commit.signature, &commit.status);
        Ok(commit.status.result)
    }

    /// 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);
        let key = signature_prefix(&signature);
        self.keeper.caches.signatures.push(key, Some(status), execution.slot);
        Ok(())
    }

    /// Commits one accepted transaction to accountsdb, writing dirty accounts
    /// only for successful execution and returning it for downstream fanout.
    fn commit_state_transitions<'t>(
        &self,
        result: &'t TransactionProcessingResult,
    ) -> Result<Option<&'t ExecutedTransaction>> {
        let execution = result.as_ref().ok().filter(|e| e.was_successful());
        let accounts = execution
            .into_iter()
            .flat_map(|execution| execution.loaded_transaction.accounts.iter())
            .filter(|(id, a)| a.dirty() && !sysvar::instructions::check_id(id));
        self.keeper.accountsdb.commit(accounts)?;
        Ok(execution.map(|execution| &**execution))
    }
}

/// Block operations namespace.
pub struct BlocksAccessor<'a> {
    pub(crate) keeper: &'a Keeper,
}

impl<'a> BlocksAccessor<'a> {
    /// Loads a retained block at the requested detail level.
    pub async fn get(&self, params: BlockParams) -> Result<Option<BlockResponse>> {
        Ok(request(self.keeper, params, ReadRequest::Block).await??)
    }

    /// Returns published retained bounds, assuming consecutive block slots.
    pub fn retained_range(&self) -> Option<RangeInclusive<Slot>> {
        self.keeper.ledger.retained_range()
    }

    /// Returns the latest block boundary known to keeper.
    pub fn latest(&self) -> Block {
        **self.keeper.caches.blocks.latest.load()
    }

    /// Returns the slot currently being built (one past the latest block).
    pub fn current_slot(&self) -> Slot {
        self.keeper.caches.blocks.latest.load().slot + 1
    }

    /// Returns whether `hash` remains in the recent blockhash cache.
    /// Expired entries remain accepted until a cache push sweeps them.
    pub fn is_valid(&self, hash: &Hash) -> bool {
        self.keeper.caches.blocks.history.contains(hash)
    }

    /// Subscribes to newly committed slots.
    pub fn subscribe(&self) -> Receiver<Block> {
        self.keeper.subscriptions.blocks.subscribe(())
    }

    /// Publishes a completed block and advances block-derived account state.
    ///
    /// Replay skips the ledger append because the block is already stored.
    pub fn append(&self, block: Signed<Block>, replay: bool) -> Result<()> {
        let event = Event::Block(block);
        if !replay {
            self.keeper.ledger.appender.send(event)?;
            self.keeper.subscriptions.blocks.send(&(), &block);
        }
        self.keeper.caches.blocks.push(*block);
        self.keeper.accounts().update_sysvars(*block)?;
        self.keeper.accounts().set_slot(block.slot)?;
        Ok(())
    }
}

/// Superblock operations namespace
pub struct SuperblockAccessor<'a> {
    pub(crate) keeper: &'a Keeper,
}

impl SuperblockAccessor<'_> {
    /// Reconstructed accountsdb seal state, without a producer signature.
    pub fn sealed(&self) -> SuperblockSeal {
        SuperblockSeal {
            id: self.keeper.accountsdb.superblock(),
            checksum: self.keeper.accountsdb.checksum(),
            transactions: self.keeper.accountsdb.transactions(),
        }
    }

    /// Returns the ledger root containing retained superblock directories.
    pub fn directory(&self) -> &PathBuf {
        &self.keeper.ledger.directory
    }

    /// Follower's current durable blockstore position, reported to the leader at handshake.
    pub fn position(&self) -> BlockstorePosition {
        self.keeper.ledger.position()
    }

    /// Enqueues a seal onto the append stream and returns its completion signal.
    ///
    /// The signal resolves after the appender durably seals the current
    /// superblock, adopts the seal's transaction count, and rotates to its successor.
    pub fn append(&self, seal: Signed<SuperblockSeal>) -> Result<oneshot::Receiver<()>> {
        let (response, completion) = oneshot::channel();
        let event = Event::Superblock { seal, response };
        self.keeper.ledger.appender.send(event)?;
        Ok(completion)
    }

    /// Blocks until every queued append event has been flushed and made durable.
    pub fn sync(&self, is_final: bool) -> Result<()> {
        let (response, ack) = oneshot::channel();
        let event = Event::Sync { response, is_final };
        self.keeper.ledger.appender.send(event)?;
        ack.recv().map_err(LedgerRequestError::from).map_err(Into::into)
    }
}