Struct OperationLog 

pub struct OperationLog {
    db: Database,
    oldest_entry: Arc<RwLock<Option<ChronologicalOperationLogKey>>>,
}

Fields

db: Database

oldest_entry: Arc<RwLock<Option<ChronologicalOperationLogKey>>>

Implementations

impl OperationLog

pub fn new(db: Database) -> Self

async fn get_oldest_operation_log_key( &self, ) -> Option<ChronologicalOperationLogKey>

Will return the oldest operation log key in the database and cache the result. If no entry exists yet the DB will be queried on each call till an entry is present.

pub async fn add_operation_log_entry_dbtx( &self, dbtx: &mut DatabaseTransaction<'_>, operation_id: OperationId, operation_type: &str, operation_meta: impl Serialize, )

pub async fn add_operation_log_entry_dbtx_with_creation_time( &self, dbtx: &mut DatabaseTransaction<'_>, operation_id: OperationId, operation_type: &str, operation_meta: impl Serialize, creation_time: SystemTime, )

pub async fn list_operations( &self, limit: usize, last_seen: Option<ChronologicalOperationLogKey>, ) -> Vec<(ChronologicalOperationLogKey, OperationLogEntry)>

👎Deprecated since 0.6.0: Use paginate_operations_rev instead

pub async fn paginate_operations_rev( &self, limit: usize, last_seen: Option<ChronologicalOperationLogKey>, ) -> Vec<(ChronologicalOperationLogKey, OperationLogEntry)>

Returns the last limit operations. To fetch the next page, pass the last operation’s ChronologicalOperationLogKey as start_after.

pub async fn get_operation( &self, operation_id: OperationId, ) -> Option<OperationLogEntry>

pub async fn get_operation_dbtx( dbtx: &mut DatabaseTransaction<'_>, operation_id: OperationId, ) -> Option<OperationLogEntry>

pub async fn set_operation_outcome( db: &Database, operation_id: OperationId, outcome: &(impl Serialize + Debug), ) -> Result<()>

Sets the outcome of an operation

pub fn outcome_or_updates<U, S>( db: &Database, operation_id: OperationId, operation_log_entry: &OperationLogEntry, is_terminal: impl Fn(&U) -> bool + MaybeSend + MaybeSync + 'static, stream_gen: impl FnOnce() -> S, ) -> UpdateStreamOrOutcome<U>

where U: Clone + Serialize + DeserializeOwned + Debug + MaybeSend + MaybeSync + 'static, S: Stream<Item = U> + MaybeSend + 'static,

Returns an a UpdateStreamOrOutcome enum that can be converted into an update stream for easier handling using UpdateStreamOrOutcome::into_stream but can also be matched over to shortcut the handling of final outcomes.

is_terminal reports whether a given update is a final state of the operation (no further update will ever follow it). Only terminal updates are cached as the operation’s outcome — an update stream that ends early (a consumer-side error path) must not persist a non-terminal update as the durable outcome, since later subscribers would then be short-circuited to it forever. The same predicate also validates outcomes READ from the cache: a cached outcome that no longer deserializes (e.g. written by an older version with a different update type) or that is non-terminal (frozen by a previous version that cached whatever update a stream ended on) is discarded with a warning and the outcome is rebuilt from the update stream — whose terminal end then overwrites the stale cache — instead of short-circuiting subscribers to it or panicking.

pub async fn optimistically_set_operation_outcome( db: &Database, operation_id: OperationId, outcome: &(impl Serialize + Debug), )

Tries to set the outcome of an operation, but only logs an error if it fails and does not return it. Since the outcome can always be recomputed from an update stream, failing to save it isn’t a problem in cases where we do this merely for caching.

Trait Implementations

impl Clone for OperationLog

fn clone(&self) -> OperationLog

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for OperationLog

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl IOperationLog for OperationLog

fn get_operation<'life0, 'async_trait>( &'life0 self, operation_id: OperationId, ) -> Pin<Box<dyn Future<Output = Option<OperationLogEntry>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn get_operation_dbtx<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, dbtx: &'life1 mut DatabaseTransaction<'life2>, operation_id: OperationId, ) -> Pin<Box<dyn Future<Output = Option<OperationLogEntry>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn add_operation_log_entry_dbtx<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, dbtx: &'life1 mut DatabaseTransaction<'life2>, operation_id: OperationId, operation_type: &'life3 str, operation_meta: Value, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

fn caching_operation_update_stream( &self, operation_id: OperationId, stream_gen: Box<dyn FnOnce() -> BoxStream<'static, Value> + Send>, is_terminal: Box<dyn Fn(&Value) -> bool + Send + Sync>, ) -> BoxStream<'static, Value>

Wrap a module’s operation update stream so that its last update is cached as the operation’s outcome — only when is_terminal reports that update as a final state of the operation. Outcome resolution (returning a cached outcome instead of a stream) happens in the typed caller (crate::module::ClientContext::outcome_or_updates), which can deserialize the cached value into the module’s update type and fall back to this stream when that fails.