Struct Data 

pub struct Data {
    servercurrentevent_data: Arc<Map>,
    servernameevent_data: Arc<Map>,
    servername_educount: Arc<Map>,
    servershortroomid_park: Arc<Map>,
    pub(super) db: Arc<Database>,
    services: Arc<OnceServices>,
}

The sending service’s column families.

Queued rows wait in servernameevent_data until a transaction claims them into servercurrentevent_data; servername_educount is the per-server EDU watermark, and servershortroomid_park holds rooms a server keeps rejecting.

Fields

servercurrentevent_data: Arc<Map>

servernameevent_data: Arc<Map>

servername_educount: Arc<Map>

servershortroomid_park: Arc<Map>

db: Arc<Database>

services: Arc<OnceServices>

Implementations

impl Data

pub(super) fn new(args: &Args<'_>) -> Self

impl Data

pub(super) fn delete_active_request(&self, key: &[u8])

impl Data

pub(super) fn delete_active_requests<'a, I>(&self, keys: I)

where I: IntoIterator<Item = &'a Vec<u8>>,

Acknowledges the active rows one transaction carried.

Empty keys mark synthetic events, which have no row.

impl Data

pub(super) async fn delete_all_requests_for(&self, destination: &Destination)

impl Data

pub(super) fn mark_as_active<'a, I>(&self, events: I)

where I: Iterator<Item = &'a (Vec<u8>, SendingEvent)>,

impl Data

pub(super) fn persist_active_edus( &self, server: &ServerName, edus: &[EduBuf], ) -> IntoIter<Vec<u8>>

Write composed EDUs straight into the active set, keyed by fresh counts.

Unlike mark_as_active there is no queue row to delete. Yields the new keys in edus order for the transaction to acknowledge.

impl Data

pub fn active_requests( &self, ) -> impl Stream<Item = (Vec<u8>, SendingEvent, Destination)> + Send + '_

Streams every active row across all destinations.

Rows are decoded as they are read; a row that fails to decode is a corrupt database and panics.

impl Data

pub fn active_requests_for( &self, destination: &Destination, ) -> impl Stream<Item = (Vec<u8>, SendingEvent)> + Send + '_ + use<'_>

Streams the active rows of one destination.

Rows are decoded as they are read; a row that fails to decode is a corrupt database and panics.

impl Data

pub(super) fn queue_requests<'a, I>(&self, requests: I) -> Vec<Vec<u8>>

where I: Iterator<Item = (&'a SendingEvent, &'a Destination)> + Clone + Debug + Send,

impl Data

pub(super) fn retain_queued<'a, I>( &'a self, events: I, ) -> impl Stream<Item = (Vec<u8>, SendingEvent)> + Send + 'a

where I: IntoIterator<Item = (Vec<u8>, SendingEvent)> + Send + 'a, I::IntoIter: Send,

Yields only pending queue items.

Empty-key payload wakes always pass because they have no durable row. A wake can outlive its row after a completed drain delivered the event.

impl Data

pub fn queued_requests( &self, destination: &Destination, ) -> impl Stream<Item = (Vec<u8>, SendingEvent)> + Send + '_ + use<'_>

Streams the queued rows of one destination.

Rows are decoded as they are read; a row that fails to decode is a corrupt database and panics.

impl Data

pub(super) fn queued_room( &self, destination: &Destination, room: ShortRoomId, ) -> impl Stream<Item = (Vec<u8>, SendingEvent)> + Send + '_ + use<'_>

Streams the queued PDU rows of one room for a destination.

An EDU row whose count equals the room’s short id has exactly the room’s key prefix and is skipped.

impl Data

pub(super) fn queued_except<'a>( &'a self, destination: &'a Destination, skip: &'a [ShortRoomId], ) -> impl Stream<Item = (Vec<u8>, SendingEvent)> + Send + 'a

Streams the queued rows of one destination, past the PDU rows of skip.

skip must be sorted. Each skipped room costs one seek rather than a visit per row, and the EDU row sharing a skipped room’s key bytes is still yielded.

impl Data

fn queued_range( &self, start: &[u8], end: Bound<Vec<u8>>, ) -> impl Stream<Item = (Vec<u8>, SendingEvent)> + Send + '_ + use<'_>

impl Data

pub(super) fn demote<'a, I>(&self, items: I, park: Option<(&ServerName, Park)>)

where I: IntoIterator<Item = &'a (Vec<u8>, SendingEvent)>,

Moves a failed transaction’s PDU rows back to the queue under their own keys.

EDU rows have no room, so they stay active and ride the next transaction. A park lands in the same batch as the rows it holds back.

impl Data

pub(super) async fn next_park( &self, server: &ServerName, room: ShortRoomId, ) -> Park

Computes the park for a room that failed again.

Each consecutive park doubles the last hold, up to a day; the caller writes it.

impl Data

pub(super) fn unpark(&self, server: &ServerName, room: ShortRoomId)

impl Data

pub(super) fn parks<'a>( &'a self, server: &'a ServerName, ) -> impl Stream<Item = Park> + Send + 'a

Streams a server’s parked rooms, expired ones included.

Rooms come in short id order, so their ids form a sorted skip list.

impl Data

pub fn parked(&self) -> impl Stream<Item = (OwnedServerName, Park)> + Send + '_

Streams every parked room with its server, expired ones included.

The server is owned, so an item may be kept across polls.

impl Data

pub(super) fn queued_badge_refresh_destinations( &self, ) -> impl Stream<Item = Destination> + Send + '_

Streams queued push destinations with a pending badge refresh.

Returned destinations are owned and may safely cross cursor advances.

impl Data

pub(super) fn queued_federation_destinations<'a, F>( &'a self, owns: F, ) -> impl Stream<Item = Result<Destination>> + Send + 'a

where F: Fn(&ServerName) -> bool + Copy + Send + 'a,

Streams distinct queued federation destinations belonging to this worker.

Each seek copies one key into the owned cursor and skips its complete destination prefix. Sigils and other shards are skipped before ownership.

impl Data

async fn seek_queued_key(&self, key: Vec<u8>) -> Result<Option<Vec<u8>>>

impl Data

pub(super) fn set_latest_educount( &self, server_name: &ServerName, last_count: u64, )

impl Data

pub async fn get_latest_educount(&self, server_name: &ServerName) -> u64

The count of the newest EDU shipped to a server.

A server never shipped to reads as zero, so its first window starts at the beginning of the log.