Struct Service 

pub struct Service {
    services: Arc<OnceServices>,
    db: Data,
    locks: MutexMap<DelayId, ()>,
    user_locks: MutexMap<OwnedUserId, ()>,
    wake: Notify,
}

Schedules delayed events and sends them when they fall due.

The lock order is the user’s lock, then an event’s lock, then the room’s lock, never the reverse.

Fields

services: Arc<OnceServices>

db: Data

locks: MutexMap<DelayId, ()>

Serializes the operations on one delay_id.

An entry lives only while held, so its size is bounded by concurrent operations.

user_locks: MutexMap<OwnedUserId, ()>

Serializes admission and deactivation per user; an entry lives only while held.

wake: Notify

Wakes the worker when a new event may fall due before its next pass.

Implementations

impl Service

pub fn limits(&self) -> Option<Limits>

Returns the limits in force, or None while delayed events are disabled.

Either configured limit at zero disables the feature.

impl Service

pub async fn check_capacity( &self, user_id: &UserId, max_scheduled: usize, ) -> Result

Refuses a user who already has max_scheduled events scheduled.

Callers hold the user’s lock through the insert, so the count cannot be overshot. At the cap, entries left by dropped records are reclaimed before refusing. The refusal’s Retry-After is the time until the user’s next event falls due, rounded up to a second.

impl Service

pub async fn existing_txn( &self, txn: RoomTxn<'_>, event_type: &str, ) -> Result<Option<DelayId>>

Returns the delay_id an earlier scheduling with this transaction stored.

The stored response is the delay_id itself, so a retried request answers with the event it already scheduled.

impl Service

pub fn schedule(&self, event: &Event, txn: Option<RoomTxn<'_>>) -> DelayId

Stores a new scheduled event and returns its delay_id.

The client’s transaction response, when given, commits in the same batch as the event’s rows. The worker is woken in case the event falls due before its next pass.

impl Service

pub async fn get(&self, delay_id: &str, owner: Option<&UserId>) -> Result<Event>

Reads a scheduled or finalised event, as missing unless owner scheduled it.

A tokenless management request passes no owner; anything that names one is held to it. The finalised record is read only when no scheduled one is found, since one batch writes it and deletes the scheduled rows.

impl Service

pub fn scheduled<'a>( &'a self, user_id: &'a UserId, ) -> impl Stream<Item = (DelayId, Event)> + Send + 'a

Streams the events a user has scheduled, with their delay_ids.

An index entry whose record is missing or unreadable is skipped.

impl Service

pub async fn restart(&self, delay_id: &str, owner: Option<&UserId>) -> Result

Counts a scheduled event’s delay again from now.

The event keeps its delay_id and content; only its due time and its due-time index entry move. A finalised event conflicts (MSC4140).

impl Service

pub async fn cancel(&self, delay_id: &str, owner: Option<&UserId>) -> Result

Cancels a scheduled event, so it can no longer be sent.

An event already cancelled, or whose send failed, cancels again without effect; one already sent conflicts (MSC4140).

impl Service

pub async fn send( &self, delay_id: &str, owner: Option<&UserId>, ) -> Result<OwnedEventId>

Sends a scheduled event now and finalises it whatever the outcome.

A failed send is never kept for another attempt, and its error is the one an ordinary send returns; an event stored before a later step failed counts as sent. An event already sent answers with its event ID again; one cancelled, or whose send failed, conflicts (MSC4140).

impl Service

pub async fn remove_user(&self, user_id: &UserId)

Removes every event a user has scheduled.

Called on deactivation, after the account’s devices are removed; an index entry whose record is already gone is dropped as well. A schedule holding the user’s lock inserts before the sweep starts; one taking it after the sweep can still schedule. A send already holding its event’s lock completes.

impl Service

pub async fn lock_user( &self, user_id: &UserId, ) -> MutexMapGuard<OwnedUserId, ()>

Locks scheduling admission and deactivation for one user.

A schedule holds this lock from its capacity check through its insert, and deactivation holds it through its sweep.

impl Service

fn due_events(&self) -> impl Stream<Item = (u64, DelayId)> + Send + '_

Streams events as they fall due, until shutdown.

Each pass reads the due-time index up to now, passing over events whose send is still in flight, then waits for the next deadline, a newly scheduled event, or shutdown. A send that hangs holds one of the worker’s slots, never the passes after it.

impl Service

async fn wait_for_due(&self, last: u64) -> Option<()>

Waits for the first deadline after last, a newly scheduled event, or shutdown, returning None on shutdown.

The wait is capped, so a lost wake costs at most one more minute.

impl Service

fn due_at(&self, now: u64) -> impl Stream<Item = (u64, DelayId)> + Send + '_

Streams the events due at now whose send is not already in flight.

One unreadable index entry is logged and passed over; nothing a single row holds can end the worker.

impl Service

async fn send_scheduled(&self, delay_id: &DelayId, due: u64)

Sends one due event, unless it moved or went since the index was read.

A due entry whose record is gone is deleted, and an unreadable record is dropped with its due entry, so no row can stall the worker. A refusal by the room’s rules has no remedy, so only a server fault logs in a release build.

impl Service

async fn send_and_finalise( &self, delay_id: &str, event: Event, ) -> Result<OwnedEventId>

Sends an event, then finalises it with the outcome.

Sending first means a crash between the two steps repeats the event rather than losing it.

impl Service

async fn append(&self, delay_id: &str, event: &Event) -> Result<OwnedEventId>

Appends an event through the ordinary local send path.

A suspended sender is refused here, since the suspension may postdate the scheduling. The event names its delay ID in its unsigned data, which the serving paths show to its sender alone (MSC4140). Once the event is stored, the append succeeds even if a later step, such as thread indexing, fails.

impl Service

async fn remove_owned(&self, user_id: &UserId, delay_id: &DelayId)

Removes one of a user’s events during deactivation.

An owner index entry whose record is gone or unreadable is deleted with that record; the worker drops its due entry when it falls due.

impl Service

async fn reclaim_orphans(&self, user_id: &UserId) -> usize

Deletes a user’s owner index entries whose record is gone or unreadable, and counts the events left.

It runs only at the cap, so an entry the worker could not clean up after a dropped record cannot hold one of the user’s slots for good.

impl Service

async fn next_due_after(&self, now: u64) -> Option<u64>

Returns when the first event due after now falls due.

The worker sleeps until then, so an unreadable entry is logged and read as no event at all.

impl Service

async fn next_due_of(&self, user_id: &UserId) -> Option<u64>

Returns when the user’s next event falls due.

It sets the Retry-After of a refusal at capacity, so it reads only that user’s events.

impl Service

fn owned_ids<'a>( &'a self, user_id: &'a UserId, ) -> impl Stream<Item = DelayId> + Send + 'a

impl Service

async fn get_event(&self, delay_id: &str) -> Result<Event>

impl Service

fn insert( &self, delay_id: &str, event: &Event, response: Option<(RoomTxn<'_>, RoomScope<'_>)>, )

Writes a new event’s rows and its transaction response in one batch.

Committing the response with the rows means a retried request finds both or neither.

impl Service

fn reschedule(&self, delay_id: &str, event: &Event, old_due: u64)

impl Service

fn finalise(&self, delay_id: &str, event: &Event)

Replaces a scheduled event’s rows with its finalised record in one batch.

A reader that misses the scheduled record therefore finds the finalised one, and the worker, which reads scheduled records alone, never sends it.

impl Service

fn remove(&self, delay_id: &str, user_id: &UserId, due: u64)

impl Service

fn drop_due(&self, delay_id: &str, due: u64)

impl Service

fn drop_owned(&self, user_id: &UserId, delay_id: &str)

Trait Implementations

impl Service for Service

fn build(args: &Args<'_>) -> Result<Arc<Self>>

Implement the construction of the service instance. Services are generally singletons so expect this to only be called once for a service type. Note that it may be called again after a server reload, but the prior instance will have been dropped first. Failure will shutdown the server with an error.

fn worker<'async_trait>( self: Arc<Self>, ) -> Pin<Box<dyn Future<Output = Result> + Send + 'async_trait>>

where Self: 'async_trait,

Implement the service’s worker loop. The service manager spawns a task and calls this function after all services have been built.

fn name(&self) -> &str

Return the name of the service. i.e. crate::service::make_name(std::module_path!())

fn interrupt<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

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

Interrupt the service. This is sent to initiate a graceful shutdown. The service worker should return from its work loop.

fn clear_cache<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

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

Clear any caches or similar runtime state.

fn memory_usage<'life0, 'life1, 'async_trait>( &'life0 self, _out: &'life1 mut (dyn Write + Send), ) -> Pin<Box<dyn Future<Output = Result> + Send + 'async_trait>>

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

Memory usage report in a markdown string.

fn unconstrained(&self) -> bool

Return true if the service worker opts out of the tokio cooperative budgeting. This can reduce tail latency at the risk of event loop starvation.