Struct Fetcher

struct Fetcher<Types, S, P>
where
    Types: NodeType,
{
    storage: Arc<S>,
    notifiers: Notifiers<Types>,
    provider: Arc<P>,
    leaf_fetcher: Arc<Fetcher<LeafRequest<Types>, LeafCallback<Types, S, P>>>,
    payload_fetcher: Option<Arc<Fetcher<PayloadRequest, PayloadCallback<Types, S, P>>>>,
    vid_common_fetcher: Option<Arc<Fetcher<VidCommonRequest, VidCommonCallback<Types, S, P>>>>,
    range_chunk_size: usize,
    active_fetch_delay: Duration,
    chunk_fetch_delay: Duration,
    backoff: ExponentialBackoff,
    retry_semaphore: Arc<Semaphore>,
    leaf_only: bool,
}

Asynchronous retrieval and storage of Fetchable resources.

Fields

storage: Arc<S>

notifiers: Notifiers<Types>

provider: Arc<P>

leaf_fetcher: Arc<Fetcher<LeafRequest<Types>, LeafCallback<Types, S, P>>>

payload_fetcher: Option<Arc<Fetcher<PayloadRequest, PayloadCallback<Types, S, P>>>>

vid_common_fetcher: Option<Arc<Fetcher<VidCommonRequest, VidCommonCallback<Types, S, P>>>>

range_chunk_size: usize

active_fetch_delay: Duration

chunk_fetch_delay: Duration

backoff: ExponentialBackoff

retry_semaphore: Arc<Semaphore>

leaf_only: bool

Implementations

impl<Types, S, P> Fetcher<Types, S, P>

where Types: NodeType, Header<Types>: QueryableHeader<Types>, S: VersionedDataSource + Sync, for<'a> S::ReadOnly<'a>: PrunedHeightStorage + NodeStorage<Types>,

pub async fn new(builder: Builder<Types, S, P>) -> Result<Self>

impl<Types, S, P> Fetcher<Types, S, P>

where Types: NodeType, Header<Types>: QueryableHeader<Types>, Payload<Types>: QueryablePayload<Types>, S: VersionedDataSource + 'static, for<'a> S::Transaction<'a>: UpdateAvailabilityStorage<Types>, for<'a> S::ReadOnly<'a>: AvailabilityStorage<Types> + NodeStorage<Types> + PrunedHeightStorage, P: AvailabilityProvider<Types>,

async fn get<T>(self: &Arc<Self>, req: impl Into<T::Request> + Send) -> Fetch<T>

where T: Fetchable<Types>,

async fn try_get<T>(self: &Arc<Self>, req: T::Request) -> Result<Option<T>>

where T: Fetchable<Types>,

Try to get an object from local storage or initialize a fetch if it is missing.

There are three possible scenarios in this function, indicated by the return type:

• Ok(Some(obj)): the requested object was available locally and successfully retrieved from the database; no fetch was spawned
• Ok(None): the requested object was not available locally, but a fetch was successfully spawned if possible (in other words, if a fetch was not spawned, it was determined that the requested object is not fetchable)
• Err(_): it could not be determined whether the object was available locally or whether it could be fetched; no fetch was spawned even though the object may be fetchable

fn get_range<R, T>(self: Arc<Self>, range: R) -> BoxStream<'static, Fetch<T>>

where R: RangeBounds<usize> + Send + 'static, T: RangedFetchable<Types>,

Get a range of objects from local storage or a provider.

Convert a finite stream of fallible local storage lookups into a (possibly infinite) stream of infallible fetches. Objects in range are loaded from local storage. Any gaps or missing objects are filled by fetching from a provider. Items in the resulting stream are futures that will never fail to produce a resource, although they may block indefinitely if the resource needs to be fetched.

Objects are loaded and fetched in chunks, which strikes a good balance of limiting the total number of storage and network requests, while also keeping the amount of simultaneous resource consumption bounded.

fn get_range_with_chunk_size<R, T>( self: Arc<Self>, chunk_size: usize, range: R, ) -> BoxStream<'static, Fetch<T>>

where R: RangeBounds<usize> + Send + 'static, T: RangedFetchable<Types>,

Same as Self::get_range, but uses the given chunk size instead of the default.

fn get_range_rev<T>( self: Arc<Self>, start: Bound<usize>, end: usize, ) -> BoxStream<'static, Fetch<T>>

where T: RangedFetchable<Types>,

Same as Self::get_range, but yields objects in reverse order by height.

Note that unlike Self::get_range, which accepts any range and yields an infinite stream if the range has no upper bound, this function requires there to be a defined upper bound, otherwise we don’t know where the reversed stream should start. The end bound given here is inclusive; i.e. the first item yielded by the stream will have height end.

fn get_range_with_chunk_size_rev<T>( self: Arc<Self>, chunk_size: usize, start: Bound<usize>, end: usize, ) -> BoxStream<'static, Fetch<T>>

where T: RangedFetchable<Types>,

Same as Self::get_range_rev, but uses the given chunk size instead of the default.

async fn get_chunk<T>(self: &Arc<Self>, chunk: Range<usize>) -> Vec<Fetch<T>>

where T: RangedFetchable<Types>,

Get a range of objects from local storage or a provider.

This method is similar to get_range, except that:

• It fetches all desired objects together, as a single chunk
• It loads the object or triggers fetches right now rather than providing a lazy stream which only fetches objects when polled.

async fn try_get_chunk<T>( self: &Arc<Self>, chunk: &Range<usize>, ) -> Result<Vec<Option<T>>>

where T: RangedFetchable<Types>,

Try to get a range of objects from local storage, initializing fetches if any are missing.

If this function succeeded, then for each object in the requested range, either:

• the object was available locally, and corresponds to Some(_) object in the result
• the object was not available locally (and corresponds to None in the result), but a fetch was successfully spawned if possible (in other words, if a fetch was not spawned, it was determined that the requested object is not fetchable)

This function will fail if it could not be determined which objects in the requested range are available locally, or if, for any missing object, it could not be determined whether that object is fetchable. In this case, there may be no fetch spawned for certain objects in the requested range, even if those objects are actually fetchable.

async fn fetch<T>( self: &Arc<Self>, tx: &mut <Self as VersionedDataSource>::ReadOnly<'_>, req: T::Request, ) -> Result<()>

where T: Fetchable<Types>,

Spawn an active fetch for the requested object, if possible.

On success, either an active fetch for req has been spawned, or it has been determined that req is not fetchable. Fails if it cannot be determined (e.g. due to errors in the local database) whether req is fetchable or not.

async fn proactive_scan( self: Arc<Self>, minor_interval: Duration, major_interval: usize, major_offset: usize, chunk_size: usize, metrics: ScannerMetrics, )

Proactively search for and retrieve missing objects.

This function will proactively identify and retrieve blocks and leaves which are missing from storage. It will run until cancelled, thus, it is meant to be spawned as a background task rather than called synchronously.

impl<Types, S, P> Fetcher<Types, S, P>

where Types: NodeType, Header<Types>: QueryableHeader<Types>, Payload<Types>: QueryablePayload<Types>, S: VersionedDataSource + 'static, for<'a> S::Transaction<'a>: UpdateAvailabilityStorage<Types> + UpdateAggregatesStorage<Types>, for<'a> S::ReadOnly<'a>: AvailabilityStorage<Types> + NodeStorage<Types> + PrunedHeightStorage + AggregatesStorage<Types>, P: AvailabilityProvider<Types>,

async fn aggregate( self: Arc<Self>, chunk_size: usize, metrics: AggregatorMetrics, )

impl<Types, S, P> Fetcher<Types, S, P>

where Types: NodeType, S: VersionedDataSource, for<'a> S::Transaction<'a>: UpdateAvailabilityStorage<Types>,

async fn store_and_notify<T>(&self, obj: T)

where T: Storable<Types>,

Store an object and notify anyone waiting on this object that it is available.

Trait Implementations

impl<Types, S: Debug, P: Debug> Debug for Fetcher<Types, S, P>

where Types: NodeType + Debug,

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

Formats the value using the given formatter.

impl<Types, S, P> VersionedDataSource for Fetcher<Types, S, P>

where Types: NodeType, S: VersionedDataSource + Send + Sync, P: Send + Sync,

type Transaction<'a> = <S as VersionedDataSource>::Transaction<'a> where Self: 'a

A transaction which can read and modify the data source.

type ReadOnly<'a> = <S as VersionedDataSource>::ReadOnly<'a> where Self: 'a

async fn write(&self) -> Result<Self::Transaction<'_>>

Start an atomic transaction on the data source.

async fn read(&self) -> Result<Self::ReadOnly<'_>>

Start a read-only transaction on the data source.