Struct Service 

pub struct Service {
    pub mutex_federation: MutexMap<OwnedRoomId, ()>,
    services: Arc<OnceServices>,
    bad_event_ratelimiter: Arc<RwLock<HashMap<OwnedEventId, (Instant, u32)>>>,
    db: Data,
}

Fields

mutex_federation: MutexMap<OwnedRoomId, ()>

services: Arc<OnceServices>

bad_event_ratelimiter: Arc<RwLock<HashMap<OwnedEventId, (Instant, u32)>>>

db: Data

Implementations

impl Service

pub async fn acl_check( &self, server_name: &ServerName, room_id: &RoomId, ) -> Result

Returns Ok if the acl allows the server

impl Service

pub(super) async fn fetch_auth<'a, Events>( &self, origin: &ServerName, room_id: &RoomId, events: Events, room_version: &RoomVersionId, recursion_level: usize, ) -> Vec<(PduEvent, Option<CanonicalJsonObject>)>

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

Find the event and auth it. Once the event is validated (steps 1 - 8) it is appended to the outliers Tree.

Returns pdu and if we fetched it over federation the raw json.

a. Look in the main timeline (pduid_pdu tree) b. Look at outlier pdu tree c. Ask origin server over federation d. TODO: Ask other servers over federation?

impl Service

async fn fetch_auth_chain( &self, origin: &ServerName, _room_id: &RoomId, event_id: &EventId, room_version: &RoomVersionId, ) -> (OwnedEventId, Option<PduEvent>, Vec<(OwnedEventId, CanonicalJsonObject)>)

impl Service

pub(super) async fn fetch_prev<'a, Events>( &self, origin: &ServerName, room_id: &RoomId, initial_set: Events, room_version: &RoomVersionId, recursion_level: usize, first_ts_in_room: MilliSecondsSinceUnixEpoch, ) -> Result<(Vec<OwnedEventId>, HashMap<OwnedEventId, (PduEvent, CanonicalJsonObject)>)>

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

impl Service

pub(super) async fn fetch_state( &self, origin: &ServerName, room_id: &RoomId, event_id: &EventId, room_version: &RoomVersionId, recursion_level: usize, create_event_id: &EventId, ) -> Result<Option<HashMap<u64, OwnedEventId>>>

Call /state_ids to find out what the state at this pdu is. We trust the server’s response to some extend (sic), but we still do a lot of checks on the events

impl Service

pub async fn handle_incoming_pdu<'a>( &'a self, origin: &'a ServerName, room_id: &'a RoomId, event_id: &'a EventId, pdu: CanonicalJsonObject, is_timeline_event: bool, ) -> Result<Option<(RawPduId, bool)>>

When receiving an event one needs to: 0. Check the server is in the room

1. Skip the PDU if we already know about it 1.1. Remove unsigned field
2. Check signatures, otherwise drop
3. Check content hash, redact if doesn’t match
4. Fetch any missing auth events doing all checks listed here starting at 1. These are not timeline events
5. Reject “due to auth events” if can’t get all the auth events or some of the auth events are also rejected “due to auth events”
6. Reject “due to auth events” if the event doesn’t pass auth based on the auth events
7. Persist this event as an outlier
8. If not timeline event: stop
9. Fetch any missing prev events doing all checks listed here starting at 1. These are timeline events
10. Fetch missing state and auth chain events by calling /state_ids at backwards extremities doing all the checks in this list starting at
    1. These are not timeline events
11. Check the auth of the event passes based on the state of the event
12. Ensure that the state is derived from the previous current state (i.e. we calculated by doing state res where one of the inputs was a previously trusted set of state, don’t just trust a set of state we got from a remote)
13. Use state resolution to find new room state
14. Check if the event passes auth based on the “current state” of the room, if not soft fail it

impl Service

pub(super) async fn handle_outlier_pdu( &self, origin: &ServerName, room_id: &RoomId, event_id: &EventId, pdu_json: CanonicalJsonObject, room_version: &RoomVersionId, recursion_level: usize, auth_events_known: bool, ) -> Result<(PduEvent, CanonicalJsonObject)>

impl Service

pub(super) async fn handle_prev_pdu( &self, origin: &ServerName, room_id: &RoomId, event_id: &EventId, eventid_info: Option<(PduEvent, CanonicalJsonObject)>, room_version: &RoomVersionId, recursion_level: usize, first_ts_in_room: MilliSecondsSinceUnixEpoch, prev_id: &EventId, create_event_id: &EventId, ) -> Result<Option<(RawPduId, bool)>>

impl Service

pub async fn parse_incoming_pdu( &self, pdu: &RawJsonValue, ) -> Result<(OwnedRoomId, OwnedEventId, CanonicalJsonObject)>

impl Service

fn cache_policy_refused(&self, event_id: &EventId)

impl Service

fn cache_policy_backoff(&self, event_id: &EventId, until_secs: u64)

impl Service

async fn cached_policy_state( &self, event_id: &EventId, ) -> Option<PolicySigState>

impl Service

pub async fn lookup_policy_server( &self, room_id: &RoomId, ) -> Option<RoomPolicyEventContent>

Returns the room’s policy event content when a policy server is in effect: state event present (stable m.room.policy, falling back to MSC4284’s unstable org.matrix.msc4284.policy), parses cleanly under either the stable public_keys map or the unstable singular public_key field, and the via server has a joined user in the room. Any failure returns None, signalling “no policy server configured” so the caller skips the gate entirely.

impl Service

pub async fn sign_outgoing_pdu<E>( &self, pdu_json: &mut CanonicalJsonObject, pdu: &E, ) -> Result

where E: Event,

MSC4284: ask the room’s policy server to sign an outgoing event. The signature is folded into pdu_json["signatures"] so it persists with the event and federates transitively to other servers in the room. Returns Forbidden when the policy server explicitly refuses; network errors and timeouts fail open with a warn log.

impl Service

async fn fetch_policy_signature( &self, policy: &RoomPolicyEventContent, pdu_json: &CanonicalJsonObject, room_version: &RoomVersionId, ) -> FetchOutcome

Calls the policy server’s /sign endpoint. The classification of the response (Signed / Refused / RateLimited / FailOpen) lets each caller choose its own reaction.

impl Service

pub async fn verify_or_fetch_inbound_policy_signature<E>( &self, pdu_json: &mut CanonicalJsonObject, pdu: &E, ) -> PolicyCheck

where E: Event,

MSC4284: verify the inbound PDU’s policy server signature; if missing, ask the policy server to sign and fold the result in. Mirrors check_inbound_policy_signature but upgrades Missing to Pass (fetched and verified, or fail-open on network error/timeout) or Invalid (refused).

impl Service

async fn fetch_inbound_policy_signature<E>( &self, pdu_json: &mut CanonicalJsonObject, pdu: &E, ) -> PolicyCheck

where E: Event,

MSC4284: when an inbound PDU has no policy server signature, ask the policy server to sign on the originator’s behalf; fold the returned signature into pdu_json so it persists with the event and federates onward. Cached refusals short-circuit to Invalid; cached backoffs (or fresh 429s) fail open as Pass until the deadline. Forbidden from the policy server maps to Invalid. Network errors and timeouts fail open with a warn log, mapped to Pass since the next server in the room is likely to retry.

impl Service

pub async fn check_inbound_policy_signature<E>( &self, pdu_json: &CanonicalJsonObject, pdu: &E, ) -> PolicyCheck

where E: Event,

MSC4284: verify the policy server signature on an inbound PDU. Returns NotApplicable for rooms without a configured policy server (the gate is skipped); Pass when the signature verifies; Missing when no signature is present for the configured server; Invalid when the signature is present but cryptographic verification fails.

impl Service

pub async fn resolve_state( &self, room_id: &RoomId, room_version: &RoomVersionId, incoming_state: HashMap<u64, OwnedEventId>, ) -> Result<Arc<CompressedState>>

impl Service

pub(super) async fn state_resolution<StateSets, AuthSets>( &self, _room_id: &RoomId, room_version: &RoomVersionId, state_sets: StateSets, auth_chains: AuthSets, ) -> Result<StateMap<OwnedEventId>>

where StateSets: Stream<Item = StateMap<OwnedEventId>> + Send, AuthSets: Stream<Item = AuthSet<OwnedEventId>> + Send,

impl Service

pub(super) async fn state_at_incoming_degree_one<Pdu>( &self, incoming_pdu: &Pdu, ) -> Result<Option<HashMap<u64, OwnedEventId>>>

where Pdu: Event,

impl Service

pub(super) async fn state_at_incoming_resolved<Pdu>( &self, incoming_pdu: &Pdu, room_id: &RoomId, room_version: &RoomVersionId, ) -> Result<Option<HashMap<u64, OwnedEventId>>>

where Pdu: Event,

impl Service

async fn state_at_incoming_fork<Pdu>( &self, room_id: &RoomId, room_version: &RoomVersionId, sstatehash: ShortStateHash, prev_event: Pdu, ) -> Result<(StateMap<OwnedEventId>, AuthSet<OwnedEventId>)>

where Pdu: Event,

impl Service

pub(super) async fn upgrade_outlier_to_timeline_pdu( &self, origin: &ServerName, room_id: &RoomId, incoming_pdu: PduEvent, pdu_json: CanonicalJsonObject, room_version: &RoomVersionId, recursion_level: usize, create_event_id: &EventId, ) -> Result<Option<(RawPduId, bool)>>

impl Service

fn cancel_back_off(&self, event_id: &EventId) -> bool

impl Service

fn back_off(&self, event_id: &EventId) -> bool

impl Service

fn is_backed_off(&self, event_id: &EventId, range: Range<Duration>) -> bool

impl Service

async fn event_exists(&self, event_id: &EventId) -> bool

impl Service

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

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