uqa_storage/mvcc/serializable/

liveness.rs

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Participant lifetime and orphan recovery never infer an uncertain durable outcome from process death.

use std::sync::{Arc, Weak};

use parking_lot::Mutex;
use uqa_core::memory::{BudgetedVec, MemoryBudget, MemoryReservation};

use super::{
    DatabaseId, ParticipantOwner, SerializableGraph, SerializableTransactionId, VersionError,
    VersionResult,
};
use crate::{
    mvcc::{CommitStatus, StorageTransactionId},
    read_control::StorageReadControl,
};

struct Participant<T> {
    id: SerializableTransactionId,
    _retained: T,
    _memory: MemoryReservation,
}

trait ParticipantLease: Send + Sync {
    fn id(&self) -> SerializableTransactionId;
}

impl<T: Send + Sync> ParticipantLease for Participant<T> {
    fn id(&self) -> SerializableTransactionId {
        self.id
    }
}

/// All clones retain the original participant's liveness through nested execution and pinned data views. The provider exposes this handle only after admission is retained. Lease release never acquires SSI admission or resolves a logical transaction. A final retained physical owner may subsequently perform its provider's ordinary close.
#[derive(Clone)]
pub struct SerializableParticipant(Arc<dyn ParticipantLease>);

impl SerializableParticipant {
    pub fn retain(
        id: SerializableTransactionId,
        lease: Box<dyn Send + Sync>,
        control: &StorageReadControl,
    ) -> VersionResult<Self> {
        Self::retain_value(id, lease, control)
    }

    fn retain_value<T: Send + Sync + 'static>(
        id: SerializableTransactionId,
        lease: T,
        control: &StorageReadControl,
    ) -> VersionResult<Self> {
        let memory = control
            .memory()
            .reserve(std::mem::size_of::<Participant<T>>() + 2 * std::mem::size_of::<usize>())?;
        Ok(Self(Arc::new(Participant {
            id,
            _retained: lease,
            _memory: memory,
        })))
    }

    pub fn id(&self) -> SerializableTransactionId {
        self.0.id()
    }
}

struct LocalEntry {
    id: SerializableTransactionId,
    lease: Weak<dyn ParticipantLease>,
}

struct LocalLeaseOwner<T> {
    registry: Arc<LocalSerializableLeases>,
    _retained: T,
}

impl<T> Drop for LocalLeaseOwner<T> {
    fn drop(&mut self) {
        // This private list lock never enters SSI/provider admission or closes physical owners. Participant construction occurs outside it, so failed publication can release its lease here too.
        self.registry.reclaim();
    }
}

/// Liveness for an exclusively owned database or one browser/in-memory owner. File-backed multiprocess providers must use native leases instead. The entries and their capacity share the owner's allowance; live handles retain the registry through their lease payload.
pub struct LocalSerializableLeases(Mutex<BudgetedVec<LocalEntry>>);

impl LocalSerializableLeases {
    pub fn new(memory: &MemoryBudget) -> Self {
        Self(Mutex::new(BudgetedVec::new(memory)))
    }

    pub fn retain(
        self: &Arc<Self>,
        id: SerializableTransactionId,
        control: &StorageReadControl,
    ) -> VersionResult<SerializableParticipant> {
        self.retain_with(id, (), control)
    }

    /// Retain an additional physical/coordinator owner with the original participant, charging the complete lease payload before allocation. Destruction must not reenter SSI admission or resolve a logical transaction. Admission operations must retain the physical owner separately so its final close cannot run while their gate is held.
    pub fn retain_with<T: Send + Sync + 'static>(
        self: &Arc<Self>,
        id: SerializableTransactionId,
        retained: T,
        control: &StorageReadControl,
    ) -> VersionResult<SerializableParticipant> {
        control.check()?;
        let participant = SerializableParticipant::retain_value(
            id,
            LocalLeaseOwner {
                registry: Arc::clone(self),
                _retained: retained,
            },
            control,
        )?;
        let mut entries = self.0.lock();
        prune(&mut entries);
        if entries.is_empty() {
            *entries = BudgetedVec::new(control.memory());
        }
        let position = entries
            .binary_search_by_key(&key(id), |entry| key(entry.id))
            .err()
            .ok_or(VersionError::InvalidEncoding(
                "duplicate serializable participant lease",
            ))?;
        entries.reserve(1)?;
        entries.push(LocalEntry {
            id,
            lease: Arc::downgrade(&participant.0),
        })?;
        entries[position..].rotate_right(1);
        Ok(participant)
    }

    pub fn is_alive(&self, id: SerializableTransactionId) -> bool {
        let entries = self.0.lock();
        entries
            .binary_search_by_key(&key(id), |entry| key(entry.id))
            .is_ok_and(|position| entries[position].lease.strong_count() != 0)
    }

    pub fn reclaim(&self) {
        prune(&mut self.0.lock());
    }
}

fn prune(entries: &mut BudgetedVec<LocalEntry>) {
    let mut kept = 0;
    for index in 0..entries.len() {
        if entries[index].lease.strong_count() != 0 {
            entries.swap(kept, index);
            kept += 1;
        }
    }
    entries.truncate(kept);
    if entries.is_empty() {
        *entries = BudgetedVec::new(entries.budget());
    }
}

fn key(id: SerializableTransactionId) -> ([u8; 16], [u8; 16], u64) {
    (id.database().as_bytes(), id.coordinator(), id.allocation())
}

impl SerializableGraph {
    pub fn database(&self) -> DatabaseId {
        self.database
    }
    pub fn coordinator(&self) -> [u8; 16] {
        self.coordinator
    }

    /// Bind admission to a retained provider lease before publishing the actor. Failed lease allocation aborts only the new actor and still consumes its allocation when the owner retains this state. Legacy manually owned actors have no lease binding and are never inferred dead by recovery.
    pub fn admit_with_lease(
        &mut self,
        read_only: bool,
        control: &StorageReadControl,
        retain: impl FnOnce(SerializableTransactionId) -> VersionResult<SerializableParticipant>,
    ) -> VersionResult<SerializableParticipant> {
        let id = self.admit(read_only, control)?;
        let retained = retain(id).and_then(|participant| {
            if participant.id() != id {
                return Err(VersionError::InvalidEncoding(
                    "serializable lease identity mismatch",
                ));
            }
            Ok(participant)
        });
        match retained {
            Ok(participant) => {
                let position = self.position(id)?;
                self.transactions[position].owner = ParticipantOwner::Leased;
                self.checkpoint_changed = true;
                Ok(participant)
            }
            Err(error) => {
                self.rollback(id)?;
                Err(error)
            }
        }
    }

    /// Retire only participants whose liveness owner is authoritatively gone. Completed participants release their outcome-retention lease without changing their confirmed state. A live unbound participant can abort immediately. A prepared physical publication must be resolved or aborted through its exact durable receipt identity; a missing receipt stops recovery and never proves rollback. Earlier confirmed resolutions survive a later error. The caller retains the resulting state before releasing shared admission, including on failure.
    pub fn recover_abandoned(
        &mut self,
        control: &StorageReadControl,
        mut is_alive: impl FnMut(SerializableTransactionId) -> VersionResult<bool>,
        mut finish: impl FnMut(StorageTransactionId) -> VersionResult<CommitStatus>,
    ) -> VersionResult<()> {
        for index in 0..self.transactions.len() {
            control.check()?;
            let entry = self.transactions[index];
            if entry.owner != ParticipantOwner::Leased {
                continue;
            }
            let participant =
                SerializableTransactionId::new(self.database, self.coordinator, entry.id)?;
            if is_alive(participant)? {
                continue;
            }
            if !entry.live() {
                self.transactions[index].owner = ParticipantOwner::Manual;
                self.checkpoint_changed = true;
                continue;
            }
            let Some(publication) = self.publication(participant)? else {
                self.rollback(participant)?;
                self.transactions[index].owner = ParticipantOwner::Manual;
                self.checkpoint_changed = true;
                continue;
            };
            let status = match finish(publication.transaction()) {
                Ok(status) => status,
                Err(VersionError::AlreadyCommitted(receipt)) => CommitStatus::Committed(receipt),
                Err(VersionError::AlreadyAborted(transaction))
                    if transaction == publication.transaction() =>
                {
                    CommitStatus::Aborted
                }
                Err(error) => return Err(error),
            };
            match self.resolve_publication(publication, status)? {
                CommitStatus::Unknown => return Err(VersionError::UnknownTransaction),
                CommitStatus::Pending | CommitStatus::Committed(_) | CommitStatus::Aborted => {}
            }
            if !self.transactions[index].live() {
                self.transactions[index].owner = ParticipantOwner::Manual;
                self.checkpoint_changed = true;
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests;