lace_tokio/

operations.rs

//! Application-facing bounded/open and porcelain operations.

use super::driver::DriverCore;
use super::policy::{all_records, private_operands};
use super::runtime::*;
use super::*;
use lace_sans_io::logic::{rx_disabled, tx_disabled};
use lace_sans_io::runtime_values::{OpenEvent, PeerDiagnostics, finish_bounded, store_outcome};

pub use bounded::BoundedInterlace;

mod bounded;
#[cfg(test)]
mod store_stream_tests;
#[cfg(test)]
mod tests;

#[must_use = "drive or drop the open interlace"]
pub struct OpenInterlace {
    pub(super) driver: DriverCore,
}

pub struct ForEachStream {
    pub(super) interlace: OpenInterlace,
    observer: Lace,
}

/// The open counterpart of [`store`]: offers records one at a time; see
/// [`store_stream`].
pub struct StoreStream {
    interlace: OpenInterlace,
    provider: Lace,
    /// The failure that ended the stream, returned by later calls; while an
    /// offer runs, the failure a cancelled offer leaves.
    failure: Option<Error>,
}

/// One route for the bare-position operand pair by participant position. A
/// `LocalPeer` position supplies its own operand instead; direct places both in
/// slot order; connected keeps the local participant's operand.
fn route(
    first: Participant,
    second: Participant,
    mut operands: [InterlaceModule; 2],
    limits: InterlaceResourceLimits,
) -> Result<OpenInterlace> {
    let local_peer = match (&first, &second) {
        (Participant::LocalPeer(_), Participant::LocalPeer(_)) => {
            return Err(Error::policy("an interlace accepts at most one LocalPeer"));
        }
        (Participant::LocalPeer(_), Participant::Connected(_))
        | (Participant::Connected(_), Participant::LocalPeer(_)) => {
            return Err(Error::policy(
                "a LocalPeer cannot interlace with a ConnectedPeer",
            ));
        }
        (Participant::LocalPeer(peer), _) => Some((0, peer.operand().clone())),
        (_, Participant::LocalPeer(peer)) => Some((1, peer.operand().clone())),
        _ => None,
    };
    if let Some((slot, operand)) = &local_peer {
        operands[*slot] = operand.clone();
    }
    let config = if first.is_local() && second.is_local() {
        OperationConfig::Direct(Box::new(
            DirectConfig::new(operands, empty_direct_authorities()).with_limits(limits),
        ))
    } else {
        let [first_operand, second_operand] = operands;
        let local_operand = if first.is_local() {
            first_operand
        } else {
            second_operand
        };
        validate_operand_blob_size(&local_operand)?;
        OperationConfig::Connected {
            local_operand,
            limits,
        }
    };
    Ok(OpenInterlace {
        driver: DriverCore::new(
            claim_participant(first)?,
            claim_participant(second)?,
            config,
            local_peer,
        )?,
    })
}

fn exhausted() -> Error {
    Error::integration("interlace exhausted before fixed point")
}

/// Drive a private capture against `target`, handing each record admitted to the
/// capture to `saved`, then close the capture.
async fn capture(
    capture: Lace,
    target: Participant,
    policy: impl IntoPolicy,
    limits: InterlaceResourceLimits,
    saved: impl FnMut(Record),
) -> Result<BoundedInterlaceOutcome> {
    let operands = private_operands(policy.into_policy()?, tx_disabled())?;
    let first = Participant::from(capture.handle());
    let mut operation = BoundedInterlace::owning(route(first, target, operands, limits)?);
    let collected = operation.collect(saved).await;
    drop(operation);
    Ok(finish_bounded(
        collected,
        capture.close().await.err(),
        PeerDiagnostics::default(),
    ))
}

pub async fn bounded_interlace(
    first: impl Into<Participant>,
    second: impl Into<Participant>,
    policy: impl IntoPolicy,
) -> Result<BoundedInterlaceOutcome> {
    bounded_interlace_with_options(first, second, policy, InterlaceOperationOptions::default())
        .await
}
pub async fn bounded_interlace_with_options(
    first: impl Into<Participant>,
    second: impl Into<Participant>,
    policy: impl IntoPolicy,
    operation: InterlaceOperationOptions,
) -> Result<BoundedInterlaceOutcome> {
    Ok(BoundedInterlace::new(first, second, policy, operation)?
        .run()
        .await)
}

pub async fn open_interlace(
    first: impl Into<Participant>,
    second: impl Into<Participant>,
    policy: impl IntoPolicy,
) -> Result<OpenInterlace> {
    open_interlace_with_options(first, second, policy, InterlaceOperationOptions::default()).await
}
pub async fn open_interlace_with_options(
    first: impl Into<Participant>,
    second: impl Into<Participant>,
    policy: impl IntoPolicy,
    operation: InterlaceOperationOptions,
) -> Result<OpenInterlace> {
    let module = policy.into_policy()?;
    let operands = [module.clone(), module];
    route(first.into(), second.into(), operands, operation.limits)
}
/// Open an interlace between a local Lace `selector` chooses, position First,
/// and `peer`, position Second. Every logical session takes its operand and
/// optional expected peer By from `selector`, called with the port of that
/// session's peer preface; the first call also chooses the Lace, which serves
/// the whole carrier, so a later call must return it again. `None`, or
/// another Lace, refuses the session as `PeerProtocol`. Until the first
/// selection the operation owns the transport and its prefacing deadline but
/// no Lace. `peer` must not carry an expected peer By, and the operation takes
/// no policy replacement.
pub async fn open_interlace_with_selector(
    peer: ConnectedPeer,
    selector: impl FnMut(Option<&PortText>) -> Option<(LaceHandle, SessionOperand)> + Send + 'static,
    operation: InterlaceOperationOptions,
) -> Result<OpenInterlace> {
    let peer = claim_participant(peer.into())?;
    Ok(OpenInterlace {
        driver: DriverCore::selecting(peer, Box::new(selector), operation.limits)?,
    })
}
impl OpenInterlace {
    /// The port the peer's latest preface sent; `None` on a direct pair,
    /// before the first peer preface, or when the peer sent none.
    pub fn peer_port(&self) -> Option<PortText> {
        self.driver.peer_port()
    }
    /// The By the live peer proved in the latest logical session under
    /// confidential or channel-bound assurance, for diagnostics; `None` on a
    /// direct pair, before that proof validates, or without one.
    pub fn peer_by(&self) -> Option<ByValue> {
        self.driver.peer_by()
    }
    /// The 065 encoding of the local-to-peer direction the latest preface
    /// pair selected, for diagnostics; `None` on a direct pair or before the
    /// first peer preface.
    pub fn outgoing_encoding(&self) -> Option<StreamEncoding> {
        self.driver.outgoing_encoding()
    }
    /// The 065 encoding of the peer-to-local direction, as
    /// [`Self::outgoing_encoding`].
    pub fn incoming_encoding(&self) -> Option<StreamEncoding> {
        self.driver.incoming_encoding()
    }
    /// Prepare and synchronously stage a hard policy replacement.
    pub fn replace_policy(&mut self, policy: impl IntoPolicy) -> Result<()> {
        let module = policy.into_policy()?;
        self.driver.replace_policy(module)
    }
    /// Return one observation, one terminal error, or post-terminal exhaustion.
    pub async fn next_event(&mut self) -> Result<Option<OpenEvent>> {
        self.driver.next_event().await
    }
    /// Cancel this operation, settle started work and connected transport
    /// ownership, and return its final local-close boundary.
    pub async fn close(self) -> Result<OpenBoundaryOutcome> {
        let reason = self.driver.close_cancelled().await?;
        Ok(OpenBoundaryOutcome {
            boundary: OpenBoundary::Closed(reason.clone()),
            observations: vec![OpenEvent::Closed { reason }],
        })
    }
    /// Consume `next_event` through the next fixed point or clean closure.
    pub async fn run_until_fixed_point(&mut self) -> Result<OpenBoundaryOutcome> {
        let mut observations = Vec::new();
        loop {
            match self.next_event().await? {
                Some(event @ OpenEvent::FixedPoint) => {
                    observations.push(event);
                    return Ok(OpenBoundaryOutcome {
                        boundary: OpenBoundary::FixedPoint,
                        observations,
                    });
                }
                Some(OpenEvent::Closed { reason }) => {
                    observations.push(OpenEvent::Closed {
                        reason: reason.clone(),
                    });
                    return Ok(OpenBoundaryOutcome {
                        boundary: OpenBoundary::Closed(reason),
                        observations,
                    });
                }
                Some(event) => observations.push(event),
                None => return Err(exhausted()),
            }
        }
    }
}

pub async fn list(target: impl Into<Participant>, policy: impl IntoPolicy) -> Result<ListOutcome> {
    list_with_options(target, policy, InterlaceOperationOptions::default()).await
}
pub async fn list_with_options(
    target: impl Into<Participant>,
    policy: impl IntoPolicy,
    operation: InterlaceOperationOptions,
) -> Result<ListOutcome> {
    let mut records = Vec::new();
    let outcome = capture(
        Lace::empty_memory(),
        target.into(),
        policy,
        operation.limits,
        |record| records.push(record),
    )
    .await?;
    Ok(ListOutcome {
        terminal: outcome.terminal,
        observations: outcome.observations,
        records,
    })
}

pub async fn get(target: impl Into<Participant>, policy: impl IntoPolicy) -> Result<GetOutcome> {
    get_with_options(target, policy, InterlaceOperationOptions::default()).await
}
pub async fn get_with_options(
    target: impl Into<Participant>,
    policy: impl IntoPolicy,
    operation: InterlaceOperationOptions,
) -> Result<GetOutcome> {
    let mut first = None;
    let outcome = capture(
        Lace::empty_memory(),
        target.into(),
        policy,
        operation.limits,
        |record| {
            first.get_or_insert(record);
        },
    )
    .await?;
    Ok(GetOutcome {
        terminal: outcome.terminal,
        observations: outcome.observations,
        record: first,
    })
}

pub async fn store(
    target: impl Into<Participant>,
    records: impl IntoRecordBatch,
) -> Result<StoreOutcome> {
    store_with_options(target, records, InterlaceOperationOptions::default()).await
}
pub async fn store_with_options(
    target: impl Into<Participant>,
    records: impl IntoRecordBatch,
    operation: InterlaceOperationOptions,
) -> Result<StoreOutcome> {
    store_inner(target.into(), records.into_records(), operation.limits).await
}
async fn store_inner(
    target: Participant,
    records: Vec<Record>,
    limits: InterlaceResourceLimits,
) -> Result<StoreOutcome> {
    if records.is_empty() {
        return Ok(StoreOutcome {
            terminal: Terminal::Complete,
            observations: Vec::new(),
            records: Vec::new(),
        });
    }
    let supplied = records
        .iter()
        .map(|record| record.hash())
        .collect::<Vec<_>>();
    let operands = private_operands(all_records().clone(), rx_disabled())?;
    let source = Lace::memory(records);
    let first = Participant::from(source.handle());
    let mut operation = BoundedInterlace::owning(route(first, target, operands, limits)?);
    let collected = operation.collect(drop).await;
    drop(operation);
    let outcome = finish_bounded(
        collected,
        source.close().await.err(),
        PeerDiagnostics::default(),
    );
    Ok(store_outcome(outcome, supplied))
}

/// Open `store`'s interlace against `target` with an empty private provider
/// and drive it to its first fixed point; then [`StoreStream::offer`] each
/// record.
pub async fn store_stream(target: impl Into<Participant>) -> Result<StoreStream> {
    store_stream_with_options(target, InterlaceOperationOptions::default()).await
}
pub async fn store_stream_with_options(
    target: impl Into<Participant>,
    operation: InterlaceOperationOptions,
) -> Result<StoreStream> {
    let target = target.into();
    let operands = private_operands(all_records().clone(), rx_disabled())?;
    let provider = Lace::empty_memory();
    let first = Participant::from(provider.handle());
    let interlace = match route(first, target, operands, operation.limits) {
        Ok(interlace) => interlace,
        Err(error) => {
            return Err(
                Error::combine([error].into_iter().chain(provider.close().await.err())).unwrap(),
            );
        }
    };
    let mut stream = StoreStream {
        interlace,
        provider,
        failure: None,
    };
    match stream.fixed_point(None).await {
        Ok(_) => Ok(stream),
        Err(error) => {
            stream.failure = Some(error.clone());
            Err(stream.close().await.err().unwrap_or(error))
        }
    }
}
impl StoreStream {
    /// Store `record` into the provider and drive to the next fixed point:
    /// `Saved` when the provider sent it, `Skipped` when it did not or the
    /// provider already held it. An error, or cancelling this future, ends
    /// the stream; later calls return the error, and records saved before it
    /// stay saved.
    pub async fn offer(&mut self, record: Record) -> Result<StoreRecordDisposition> {
        if let Some(error) = &self.failure {
            return Err(error.clone());
        }
        self.failure = Some(Error::integration(
            "a cancelled offer left the store stream unsettled",
        ));
        let result = self.offer_inner(record).await;
        self.failure = result.as_ref().err().cloned();
        result
    }

    async fn offer_inner(&mut self, record: Record) -> Result<StoreRecordDisposition> {
        let hash = record.hash();
        let outcome = self
            .provider
            .owner
            .start_store_execution_at(
                StoreOp::StoreRecords {
                    records: vec![record],
                },
                Some(ParticipantPosition::First),
            )
            .receive()
            .await?;
        if let Some(error) = outcome.failure {
            return Err(error);
        }
        match outcome.response {
            Some(Ok(StoreResult::StoredRecords { outcomes, .. })) => match outcomes.as_slice() {
                [StoreRecordOutcome::Duplicate { .. }] => Ok(StoreRecordDisposition::Skipped),
                [StoreRecordOutcome::Stored { .. }] => Ok(if self.fixed_point(Some(hash)).await? {
                    StoreRecordDisposition::Saved
                } else {
                    StoreRecordDisposition::Skipped
                }),
                _ => Err(Error::integration(
                    "provider store returned the wrong outcomes",
                )),
            },
            _ => Err(Error::integration(
                "provider store returned the wrong result",
            )),
        }
    }

    /// Drive to the next fixed point; whether the provider sent `hash`
    /// before it.
    async fn fixed_point(&mut self, hash: Option<RecordHash>) -> Result<bool> {
        let mut sent = false;
        loop {
            match self.interlace.next_event().await? {
                Some(OpenEvent::FixedPoint) => return Ok(sent),
                Some(OpenEvent::RecordSent {
                    position: ParticipantPosition::First,
                    record_hash,
                }) => sent |= Some(record_hash) == hash,
                Some(OpenEvent::Closed { reason }) => {
                    return Err(Error::peer_protocol(format!(
                        "store stream closed before an offer's fixed point: {reason:?}"
                    )));
                }
                Some(_) => {}
                None => return Err(exhausted()),
            }
        }
    }

    /// Cancel the interlace and close the provider, returning the failure
    /// that ended the stream with any cleanup failure.
    pub async fn close(self) -> Result<()> {
        let StoreStream {
            interlace,
            provider,
            failure,
        } = self;
        let interlace = interlace.close().await.err();
        let provider = provider.close().await.err();
        Error::combine(failure.into_iter().chain(interlace).chain(provider)).map_or(Ok(()), Err)
    }
}

pub async fn for_each(
    target: impl Into<Participant>,
    policy: impl IntoPolicy,
) -> Result<ForEachStream> {
    for_each_with_options(target, policy, InterlaceOperationOptions::default()).await
}
pub async fn for_each_with_options(
    target: impl Into<Participant>,
    policy: impl IntoPolicy,
    operation: InterlaceOperationOptions,
) -> Result<ForEachStream> {
    let target = target.into();
    let observer = Lace::empty_memory();
    let operands = private_operands(policy.into_policy()?, tx_disabled())?;
    let first = Participant::from(observer.handle());
    Ok(ForEachStream {
        interlace: route(first, target, operands, operation.limits)?,
        observer,
    })
}
impl ForEachStream {
    /// Cancel the observer operation and return its final local-close event.
    pub async fn close(self) -> Result<OpenEvent> {
        let ForEachStream {
            interlace,
            observer,
        } = self;
        let interlace_result = interlace.driver.close_cancelled().await;
        let observer_result = observer.close().await;
        if let Some(error) = Error::combine(
            interlace_result
                .as_ref()
                .err()
                .cloned()
                .into_iter()
                .chain(observer_result.err()),
        ) {
            return Err(error);
        }
        Ok(OpenEvent::Closed {
            reason: interlace_result?,
        })
    }

    pub async fn next_event(&mut self) -> Result<Option<OpenEvent>> {
        loop {
            match self.interlace.driver.next_event().await? {
                Some(
                    event @ (OpenEvent::RecordSaved {
                        position: ParticipantPosition::First,
                        ..
                    }
                    | OpenEvent::FixedPoint
                    | OpenEvent::Closed { .. }),
                ) => return Ok(Some(event)),
                Some(_) => {}
                None => return Ok(None),
            }
        }
    }
}