1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
pub mod buffered;
pub mod errors;
pub mod futures;
use std::fmt::Debug;
pub use buffered::{Buffer, BufferConfig, BufferedStream, PrefetchHandle, SharedBufferConfig};
pub use errors::CorrelatedStreamError;
pub use futures::{CompletionHandle, Next, NextVec, NextVecIterator, ResyncHandle};
use crate::correlated_randomness::Correlation;
/// Main abstraction interface to obtain preprocessing items.
///
/// To keep consistency, requests must be issued in the same order across all parties.
pub trait CorrelatedStream<P: Correlation>: Send {
/// The error type returned by the generator.
type Error: Debug + Clone + From<CorrelatedStreamError> + Send + 'static;
/// Returns a single future that resolves to all `n_elements` items at once.
/// Can fail if request is too large/small or if the request cannot
/// be fulfilled by the stream for some reason.
fn next_n(&self, n_elements: usize) -> Result<NextVec<P, Self::Error>, CorrelatedStreamError>;
/// Proactively generates `n_elements` items into the buffer so later [`next_n`](Self::next_n)
/// calls need not wait on generation. Fire-and-forget (proceeds whether or not the handle is
/// awaited); does not advance [`position`](Self::position). `Ok` means the items were generated
/// or were already buffered.
///
/// Completion is best-effort across a [`resync`](Self::resync): items generated for a prefetch
/// that land while a resync is pending are burned against its deficit rather than buffered, so
/// the handle can resolve `Ok` with fewer (even zero) items actually buffered. The guarantee is
/// on progress, not on occupancy — a later `next_n` may still have to wait on generation.
fn prefetch_n(&self, n_elements: usize) -> PrefetchHandle<Self::Error>;
/// The cumulative number of items *delivered* through [`next_n`](Self::next_n) since creation
/// (prefetching doesn't count). Stays in sync across parties, so it can be exchanged to agree
/// on a common resync target.
fn position(&self) -> u64;
/// The number of already-generated elements currently sitting in the stream's buffer, ready to
/// be delivered without further generation. Purely a runtime metric — unlike
/// [`position`](Self::position) it need not stay in sync across parties. The default `0` suits
/// streams that do not buffer.
fn buffered(&self) -> u64;
/// Advances to the absolute logical `target` by discarding intervening elements (draining the
/// buffer, then skipping at the generator if supported) — the recovery primitive for
/// non-fatal desyncs, where every party resyncs to the agreed `max(position)`.
///
/// Forward-only: `target == position()` is a no-op, `target < position()` must error with
/// [`ResyncRewind`](CorrelatedStreamError::ResyncRewind), and a `target` beyond what the stream
/// can reach must error with [`ResyncUnsupported`](CorrelatedStreamError::ResyncUnsupported).
///
/// A resync is a quiescent barrier: while one is pending, `next_n` and further `resync` calls
/// are rejected with [`ResyncInProgress`](CorrelatedStreamError::ResyncInProgress) — elements
/// drawn mid-resync could come from below the target. [`prefetch_n`](Self::prefetch_n) is the
/// exception: it delivers nothing, so it cannot draw pre-target elements and is still
/// accepted, though it is starved of new generation orders until the resync clears and items
/// already in flight for it may be burned against the resync deficit (see its docs).
///
/// Callers must therefore serialize `next_n` against `resync` — the rejection is a race
/// detector, not the ordering mechanism. `target` is absolute, so a request issued
/// concurrently with a resync has no well-defined side of the barrier, and nothing here can
/// make the side it lands on agree with the one the other parties chose.
fn resync(&self, target: u64) -> ResyncHandle<Self::Error>;
}