Struct Service 

pub struct Service {
    services: Arc<OnceServices>,
    db: Data,
    pub mutex_insert: MutexMap<OwnedRoomId, ()>,
}

Provides persistent event lookup, insertion, and room timeline traversal.

Accepted events and outliers occupy separate maps while event IDs point to accepted timeline positions. A per-room insertion lock serializes the final mutation stage after federation and state work.

Fields

services: Arc<OnceServices>

db: Data

mutex_insert: MutexMap<OwnedRoomId, ()>

Serializes timeline insertion as the leaf per-room operation.

Acquire it after any federation or state mutex held for the same room. Never acquire either outer mutex while holding this guard.

Implementations

impl Service

pub(crate) async fn append_incoming_pdu<'a, Leafs>( &'a self, pdu: &'a PduEvent, pdu_json: CanonicalJsonObject, new_room_leafs: Leafs, state_ids_compressed: Arc<CompressedState>, soft_fail: bool, state_lock: &'a RoomMutexGuard, ) -> Result<Option<RawPduId>>

where Leafs: Iterator<Item = &'a EventId> + Send + 'a,

Appends an incoming event with its locally resolved state snapshot.

The snapshot is recorded even when the event is soft-failed. A soft-failed event is not inserted into the accepted timeline, but its predecessors are marked referenced. Only a nonempty replacement extremity band is stored; an empty calculation preserves the prior band.

impl Service

pub async fn append_pdu<'a, Leafs>( &'a self, pdu: &'a PduEvent, pdu_json: CanonicalJsonObject, leafs: Leafs, state_lock: &'a RoomMutexGuard, ) -> Result<RawPduId>

where Leafs: Iterator<Item = &'a EventId> + Send + 'a,

Persists an authenticated event and applies its timeline side effects.

This method performs no authentication. The accepted row, event mapping, outlier removal, and timestamp index are committed together, but later cache, indexing, notification, and membership work is not part of that transaction, so an error can be returned after the event is stored.

impl Service

async fn prev_state(&self, pdu: &PduEvent, state_key: &str) -> Option<PduEvent>

impl Service

async fn append_pdu_effects( &self, pdu_id: RawPduId, pdu: &PduEvent, shortroomid: ShortRoomId, count: PduCount, state_lock: &RoomMutexGuard, ) -> Result

impl Service

async fn append_member_effects(&self, pdu: &PduEvent, count: PduCount) -> Result

Record the membership transition an m.room.member event carries.

The cache is written here rather than off the resolved state so that a user who is invited or knocked and leaves immediately still leaves the earlier event on record for auth.

impl Service

async fn append_message_effects( &self, pdu_id: &RawPduId, pdu: &PduEvent, shortroomid: ShortRoomId, ) -> Result

Index an m.room.message event’s body, and queue it when it is an admin command.

The queued command carries the event’s sender, so a handler can tell who issued it, and the event’s id, which its response replies to.

impl Service

fn append_pdu_json( &self, pdu_id: &RawPduId, pdu: &PduEvent, json: &CanonicalJsonObject, )

impl Service

pub async fn backfill_if_required( &self, room_id: &RoomId, from: PduCount, ) -> Result

Attempts federation backfill when a request reaches local history’s edge.

Backfill is skipped after the create event and for effectively empty rooms that are not world-readable. No candidate, a remote fetch failure, or an empty accepted chunk is treated as a successful best-effort outcome.

impl Service

async fn backfill_candidates(&self, room_id: &RoomId) -> Candidates

impl Service

pub async fn get_event_id_near_ts_with_fallback( &self, room_id: &RoomId, ts: MilliSecondsSinceUnixEpoch, dir: Direction, ) -> Result<(MilliSecondsSinceUnixEpoch, OwnedEventId)>

Finds the nearest event to a timestamp with a federation fallback.

The local answer is retained unless a closer remote claim can be ingested. The ingested event must belong to this room and its actual timestamp must remain on the requested side of the query.

impl Service

async fn is_start_edge_hit(&self, room_id: &RoomId, event_id: &EventId) -> bool

impl Service

async fn backfill_event( &self, room_id: &RoomId, event_id: &EventId, origin: &ServerName, ) -> Result<PduEvent>

impl Service

pub async fn fetch_remote_event( &self, room_id: &RoomId, event_id: &EventId, ) -> Result

Fetches one remote event and persists it through the backfill path.

Fetcher checks are disabled only for retrieval; backfill_pdu performs signature, hash, and authorization validation. An event already accepted completes successfully without another insertion, while a stored outlier is still validated and promoted.

impl Service

pub async fn backfill_pdu( &self, room_id: &RoomId, origin: &ServerName, pdu: Box<RawJsonValue>, ) -> Result<bool>

Validates and inserts one remotely supplied event as backfilled history.

An already accepted duplicate returns false without insertion. A new event or stored outlier receives a negative backfill count, moves to accepted storage, updates its timestamp and ID indexes, and adds searchable message or topic content.

impl Service

fn prepend_backfill_pdu( &self, pdu_id: &RawPduId, room_id: &RoomId, event_id: &EventId, origin_server_ts: u64, json: &CanonicalJsonObject, )

impl Service

pub async fn build_and_append_pdu( &self, pdu_builder: PduBuilder, sender: &UserId, room_id: &RoomId, state_lock: &RoomMutexGuard, ) -> Result<OwnedEventId>

Builds, signs, validates, and persists a locally authored room event.

The caller’s state guard serializes state derivation and advancement while the event is inserted into the accepted timeline. Federation enqueue occurs after persistence, so its failure can be returned after the event and room state have already been stored.

impl Service

async fn sanitize_member_authorisation( &self, pdu_builder: &mut PduBuilder, room_id: &RoomId, ) -> Result

impl Service

async fn check_pdu_for_admin_room<Pdu>( &self, pdu: &Pdu, sender: &UserId, ) -> Result

where Pdu: Event,

impl Service

async fn check_admin_room_member<Pdu>( &self, pdu: &Pdu, sender: &UserId, ) -> Result

where Pdu: Event,

Refuses a leave or ban that would take the server user, or the last active admin, out of the admins room.

Deactivation runs the same last-admin check under this room’s state lock; see lock_admin_room.

impl Service

pub async fn create_hash_and_sign_event( &self, pdu_builder: PduBuilder, sender: &UserId, room_id: &RoomId, _mutex_lock: &RoomMutexGuard, ) -> Result<(PduEvent, CanonicalJsonObject)>

Builds a canonical signed PDU and its structured representation.

Up to twenty current forward extremities become predecessors, and a non-create event is rejected when no predecessor is available. The event is authorization-checked and formatted under the room version, including the version-specific room-ID rules, but is not persisted here.

impl Service

async fn compute_prev_events( &self, room_id: &RoomId, event_type: &TimelineEventType, ) -> Result<PrevEvents>

impl Service

pub async fn delete_pdus(&self, room_id: &RoomId) -> Result

Deletes every accepted timeline row belonging to a room.

Each event’s accepted row, ID mapping, outlier copy, and timestamp index are removed in one transaction. An error stops the scan after any prior events were already removed, and metadata owned by other services is not purged.

impl Service

pub fn pdus_near_ts( &self, user_id: Option<&UserId>, room_id: &RoomId, ts: MilliSecondsSinceUnixEpoch, dir: Direction, ) -> impl Stream<Item = Result<PdusIterItem>> + Send

Streams decoded room events from the timestamp index in one direction.

The timestamp boundary is inclusive. user_id controls sender-only transaction metadata, while event age is always updated. It does not filter which events are yielded.

impl Service

pub fn pdu_ids_near_ts( &self, room_id: &RoomId, ts: MilliSecondsSinceUnixEpoch, dir: Direction, ) -> impl Stream<Item = Result<(MilliSecondsSinceUnixEpoch, PduId)>> + Send

Streams timestamp and PDU-ID pairs from the room timestamp index.

Forward scans begin at the first row at or after the timestamp, while backward scans begin at the first row at or before it. Only rows for the requested room are yielded.

impl Service

pub fn all_pdus<'a>( &'a self, user_id: &'a UserId, room_id: &'a RoomId, ) -> impl Stream<Item = PdusIterItem> + Send + 'a

Streams all accepted PDUs in a room in forward order.

The requesting user receives sender-only transaction metadata where applicable. Unknown rooms and all per-item storage or decoding errors are suppressed, producing only successfully presented events.

impl Service

pub fn pdus<'a>( &'a self, user_id: Option<&'a UserId>, room_id: &'a RoomId, from: Option<PduCount>, ) -> impl Stream<Item = Result<PdusIterItem>> + Send + 'a

Streams accepted room events after an optional count in forward order.

The count boundary is exclusive and defaults to the minimum count. The optional user controls presentation only; stream, storage, and decoding errors remain visible to the caller.

impl Service

pub fn pdus_rev<'a>( &'a self, user_id: Option<&'a UserId>, room_id: &'a RoomId, until: Option<PduCount>, ) -> impl Stream<Item = Result<PdusIterItem>> + Send + 'a

Streams accepted room events before an optional count in reverse order.

The count boundary is exclusive and defaults to the maximum count. The optional user controls presentation only; stream, storage, and decoding errors remain visible to the caller.

impl Service

pub fn pdus_raw(&self) -> impl Stream<Item = Result<Val<'_>>> + Send

Streams raw JSON values from all accepted timeline rows.

Values borrow the database cursor and must be owned before they are retained across another poll. Storage errors remain in the stream.

impl Service

pub fn outlier_pdus_raw(&self) -> impl Stream<Item = Result<Val<'_>>> + Send

Streams raw JSON values from all outlier rows.

Values borrow the database cursor and must be owned before they are retained across another poll. Storage errors remain in the stream.

impl Service

fn each_slice( (pdu_id, pdu): KeyVal<'_>, user_id: Option<&UserId>, ) -> Result<PdusIterItem>

impl Service

fn each_pdu( (pdu_id, pdu): (RawPduId, PduEvent), user_id: Option<&UserId>, ) -> Result<PdusIterItem>

impl Service

pub async fn purge_history( &self, room_id: &RoomId, until: PduCount, delete_local_events: bool, ) -> Result<usize>

Purges eligible room history strictly before a timeline count.

State events are preserved, and locally sent events remain unless delete_local_events is set. Forward extremities are untouched, while search, relation, and retained-original data are removed for each deleted event; an error can stop the operation after partial progress.

impl Service

pub async fn redact_pdu<Pdu: Event + Send + Sync>( &self, event_id: &EventId, reason: &Pdu, shortroomid: ShortRoomId, state_lock: &RoomMutexGuard, ) -> Result

Replaces an accepted PDU with its room-version redacted form.

Failure to resolve the event’s accepted PDU ID is treated as a successful no-op. Original retention, search removal, and relation deletion occur before the accepted row is replaced, so the operation is not atomic if a later step fails.

impl Service

pub async fn replace_pdu( &self, pdu_id: &RawPduId, pdu_json: &CanonicalJsonObject, ) -> Result

Replaces the stored JSON of an accepted PDU without changing its ID.

A definite missing-row result is returned as NotFound; otherwise the accepted timeline row is overwritten in place.

impl Service

pub(super) fn stage_replace_pdu( &self, txn: &mut Txn, pdu_id: &RawPduId, pdu_json: &CanonicalJsonObject, )

Stage replacement of a PDU already loaded under its room’s guard.

The caller retains the guard until the transaction commits.

impl Service

pub fn add_pdu_outlier(&self, event_id: &EventId, pdu: &CanonicalJsonObject)

Stores an event as an outlier outside the accepted room timeline.

The event is keyed directly by event ID and no accepted-timeline mapping or stream position is created.

impl Service

pub async fn first_pdu_in_room(&self, room_id: &RoomId) -> Result<PduEvent>

Returns the earliest accepted PDU in a room.

Unknown or empty rooms report the underlying stream’s not-found result.

impl Service

pub async fn latest_pdu_in_room(&self, room_id: &RoomId) -> Result<PduEvent>

Returns the latest accepted PDU in a room.

Presentation removes sender-only transaction metadata because no requesting user is supplied.

impl Service

pub async fn first_item_in_room( &self, room_id: &RoomId, ) -> Result<(PduCount, PduEvent)>

Returns the earliest accepted PDU and its room-local stream count.

The forward room stream provides the first available item after applying ordinary presentation transformations.

impl Service

pub async fn latest_item_in_room( &self, sender_user: Option<&UserId>, room_id: &RoomId, ) -> Result<PduEvent>

Returns the latest accepted PDU in a room.

sender_user controls presentation of sender-only transaction metadata; it does not filter events by sender.

impl Service

pub async fn prev_shortstatehash( &self, room_id: &RoomId, before: PduCount, ) -> Result<ShortStateHash>

Returns the state snapshot at the room event directly before a count.

The before boundary is exclusive and need not identify an existing event or even belong to the room.

impl Service

pub async fn next_shortstatehash( &self, room_id: &RoomId, after: PduCount, ) -> Result<ShortStateHash>

Returns the state snapshot at the room event directly after a count.

The after boundary is exclusive and need not identify an existing event or even belong to the room.

impl Service

pub async fn shortstatehash_after( &self, room_id: &RoomId, count: PduCount, ) -> Result<ShortStateHash>

Returns the state snapshot after every room event at or before a count.

The boundary is inclusive because a client’s sync position is often the count of the newest event it received, and need not belong to this room. The snapshot precedes the first event strictly after the count, falling back to current state only when no event follows; a count before the room’s first recorded snapshot is not found.

impl Service

pub async fn get_shortstatehash( &self, room_id: &RoomId, count: PduCount, ) -> Result<ShortStateHash>

Returns the state snapshot recorded at a room timeline count.

The count is resolved through the room’s accepted timeline row and then its event-to-state association.

impl Service

pub async fn prev_timeline_count(&self, before: &PduId) -> Result<PduCount>

Returns the room timeline count directly before an encoded PDU ID.

The boundary is exclusive and need not identify an existing row. Its room component selects the timeline prefix to scan.

impl Service

pub async fn next_timeline_count(&self, after: &PduId) -> Result<PduCount>

Returns the room timeline count directly after an encoded PDU ID.

The boundary is exclusive and need not identify an existing row. Its room component selects the timeline prefix to scan.

impl Service

pub async fn last_timeline_count( &self, sender_user: Option<&UserId>, room_id: &RoomId, upper_bound: Option<PduCount>, ) -> Result<PduCount>

Returns the latest normal timeline count at or below an optional bound.

Backfilled counts are not returned. When no normal event qualifies, the sentinel PduCount::max() is returned instead of a not-found error; sender_user affects presentation only.

impl Service

pub async fn get_event_id_near_ts( &self, room_id: &RoomId, ts: MilliSecondsSinceUnixEpoch, dir: Direction, ) -> Result<(MilliSecondsSinceUnixEpoch, OwnedEventId)>

Returns the indexed event ID nearest a timestamp in one direction.

Forward lookup selects the first event at or after the timestamp; backward lookup selects the first event at or before it. The stored event timestamp is returned with the ID.

impl Service

pub async fn get_pdu_id_near_ts( &self, room_id: &RoomId, ts: MilliSecondsSinceUnixEpoch, dir: Direction, ) -> Result<(MilliSecondsSinceUnixEpoch, PduId)>

Returns the indexed PDU ID nearest a timestamp in one direction.

Forward lookup selects the first event at or after the timestamp; backward lookup selects the first event at or before it. A room with no qualifying event returns NotFound.

impl Service

pub async fn get_pdu_near_ts( &self, _user_id: Option<&UserId>, room_id: &RoomId, ts: MilliSecondsSinceUnixEpoch, dir: Direction, ) -> Result<PdusIterItem>

Returns the accepted PDU nearest a timestamp in one direction.

The result contains the selected room-local count and decoded PDU. The current _user_id parameter has no effect and no presentation or visibility filtering is applied.

impl Service

async fn count_to_id( &self, room_id: &RoomId, count: PduCount, dir: Direction, ) -> Result<RawPduId>

impl Service

fn pdu_count_to_id( shortroomid: ShortRoomId, count: PduCount, dir: Direction, ) -> RawPduId

impl Service

pub async fn get_pdu_from_shorteventid( &self, shorteventid: ShortEventId, ) -> Result<PduEvent>

Returns a decoded PDU resolved from a short event ID.

The short ID is expanded to an event ID before accepted and outlier storage are queried through Service::get_pdu.

impl Service

pub async fn get_pdu(&self, event_id: &EventId) -> Result<PduEvent>

Returns a decoded PDU from accepted or outlier storage.

Both lookups are polled concurrently, so if duplicate rows exist the first successful lookup determines the returned value.

impl Service

pub async fn get_outlier_pdu(&self, event_id: &EventId) -> Result<PduEvent>

Returns a decoded PDU from outlier storage.

Accepted timeline storage is not consulted, and storage or decoding errors propagate to the caller.

impl Service

pub async fn get_non_outlier_pdu(&self, event_id: &EventId) -> Result<PduEvent>

Returns a PDU from the accepted timeline.

Looks up the accepted record only, without consulting outliers. Storage and decoding errors propagate to the caller.

impl Service

pub async fn get_pdu_from_id(&self, pdu_id: &RawPduId) -> Result<PduEvent>

Returns a decoded PDU by its accepted timeline ID.

The accepted row is read directly without consulting the event-ID mapping or outlier storage.

impl Service

pub async fn get_pdu_json( &self, event_id: &EventId, ) -> Result<CanonicalJsonObject>

Returns canonical PDU JSON from accepted or outlier storage.

Both lookups are polled concurrently, so if duplicate rows exist the first successful lookup determines the returned value.

impl Service

pub async fn get_outlier_pdu_json( &self, event_id: &EventId, ) -> Result<CanonicalJsonObject>

Returns canonical PDU JSON from outlier storage.

Accepted timeline storage is not consulted, and storage or decoding errors propagate to the caller.

impl Service

pub async fn get_non_outlier_pdu_json( &self, event_id: &EventId, ) -> Result<CanonicalJsonObject>

Returns the JSON of a PDU from the accepted timeline.

Looks up the accepted record only, without consulting outliers. Storage and decoding errors propagate to the caller.

impl Service

pub async fn get_pdu_json_from_id( &self, pdu_id: &RawPduId, ) -> Result<CanonicalJsonObject>

Returns canonical PDU JSON by its accepted timeline ID.

The accepted row is read directly without consulting the event-ID mapping or outlier storage.

impl Service

pub async fn get<T>(&self, event_id: &EventId) -> Result<T>

where T: for<'de> Deserialize<'de>,

Deserializes an event from accepted or outlier storage into T.

Both lookups are polled concurrently, so if duplicate rows exist the first successful lookup determines the returned value.

impl Service

pub async fn get_outlier<T>(&self, event_id: &EventId) -> Result<T>

where T: for<'de> Deserialize<'de>,

Deserializes an event from outlier storage into T.

Accepted timeline storage is not consulted, and storage or decoding errors propagate to the caller.

impl Service

pub async fn get_non_outlier<T>(&self, event_id: &EventId) -> Result<T>

where T: for<'de> Deserialize<'de>,

Deserializes a PDU from the accepted timeline into T.

Resolves the event ID through eventid_pduid and reads pduid_pdu, without consulting outliers. Storage and decoding errors propagate to the caller.

impl Service

pub async fn get_from_id<T>(&self, pdu_id: &RawPduId) -> Result<T>

where T: for<'de> Deserialize<'de>,

Deserializes an accepted timeline row into T by PDU ID.

The row is read directly without consulting the event-ID mapping or outlier storage.

impl Service

pub async fn pdu_exists<'a>(&'a self, event_id: &'a EventId) -> bool

Reports whether an event exists in accepted or outlier storage.

The two existence checks run concurrently, and any lookup errors are treated as absence.

impl Service

pub async fn non_outlier_pdus_exist<'a, I>(&self, event_ids: I) -> bool

where I: Iterator<Item = &'a EventId> + Send,

Reports whether every event has an accepted timeline row.

The probe stops at the first event without one, and an empty input reports true; a lookup error counts as absence. The rows are read one at a time, so callers pass a small set such as one event’s references, never a bulk population.

impl Service

pub fn watch_event<'a>( &'a self, event_id: &EventId, ) -> impl Future<Output = ()> + Send + 'a

Returns a future that resolves on the next event-to-PDU mapping mutation.

Registration against the event-to-PDU mapping is eager, so the watcher is installed before the returned future is first awaited. Accepted insertion is the normal wakeup, but any mutation under the event-ID prefix can resolve the future.

impl Service

pub async fn non_outlier_pdu_exists(&self, event_id: &EventId) -> Result

Checks whether an event has an accepted timeline row.

The event-to-PDU mapping is resolved first and the target row is then tested without fetching or decoding the PDU.

impl Service

pub async fn outlier_pdu_exists(&self, event_id: &EventId) -> Result

Checks whether an event has an outlier row.

Accepted timeline storage is not consulted and the PDU is not fetched or decoded.

impl Service

pub async fn get_pdu_count(&self, event_id: &EventId) -> Result<PduCount>

Returns the room-local timeline count assigned to an accepted event.

The event ID is resolved through the accepted event-to-PDU mapping.

impl Service

pub async fn get_shorteventid_from_pdu_id( &self, pdu_id: &PduId, ) -> Result<ShortEventId>

Returns the short event ID represented by an accepted PDU ID.

The accepted row supplies the full event ID, which is then resolved through the short-ID service.

impl Service

pub async fn get_event_id_from_pdu_id( &self, pdu_id: &PduId, ) -> Result<OwnedEventId>

Returns the event ID stored at an accepted PDU ID.

The accepted row is decoded as a PDU to recover its event ID.

impl Service

pub async fn get_pdu_id_from_shorteventid( &self, shorteventid: ShortEventId, ) -> Result<RawPduId>

Returns the accepted PDU ID associated with a short event ID.

The short ID is first expanded to its full event ID before the timeline mapping is read.

impl Service

pub async fn get_pdu_id(&self, event_id: &EventId) -> Result<RawPduId>

Returns the accepted timeline ID associated with an event.

Outlier storage is not consulted because outliers have no room timeline position.

Trait Implementations

impl FetchEvent for &Service

async fn get<T>(self, event_id: &EventId) -> Result<T>

where T: for<'de> Deserialize<'de> + Send,

Decodes an event into the requested representation.

async fn exists(self, event_id: &EventId) -> Result<bool>

Checks whether an event exists without requiring a full decode.

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 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 name(&self) -> &str

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

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 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 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.