Struct Service 

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

Fields

mutex_federation: MutexMap<OwnedRoomId, ()>

services: Arc<OnceServices>

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) fn record_attempt(&self, ctx: Context, event_id: &EventId)

Record a federation attempt before a cancellable await, so a premature cancellation still leaves a Pending row behind to rate-gate against.

impl Service

pub(super) fn record_outcome( &self, ctx: Context, event_id: &EventId, disposition: Disposition, )

impl Service

pub(super) async fn record_success(&self, ctx: Context, event_id: &EventId)

impl Service

pub(super) async fn is_suppressed( &self, ctx: Context, event_id: &EventId, range: Range<Duration>, ) -> Suppression

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, incoming_event_id: &EventId, 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

async fn await_prev_gap<'a, Events>( &self, initial_set: Events, wait: Duration, ) -> bool

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

impl Service

async fn prefetch_missing_events( &self, origin: &ServerName, room_id: &RoomId, incoming_event_id: &EventId, room_version: &RoomVersionId, recursion_level: usize, )

Fill the prev gap below incoming_event_id with one /get_missing_events batch, landing each returned event as a local outlier so the per-event walk resolves it without a federation fetch. latest_events is the held event the server walks back from, bounded by our forward extremities so it returns only the gap; best effort, so a failed batch or rejected event just leaves that id for the walk.

impl Service

async fn land_missing_event( &self, origin: &ServerName, room_id: &RoomId, pdu: &RawJsonValue, room_version: &RoomVersionId, recursion_level: usize, ) -> Result

Authenticate and persist one event from the missing-events batch as an outlier, deriving its id from content rather than trusting a requested id.

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

async fn handle_rescinded_invite( &self, room_id: &RoomId, pdu: &CanonicalJsonObject, ) -> Result<bool>

Apply a federated leave that rescinds an out-of-band invite for a local user.

We are not resident in the room, so the kick cannot be processed as a normal timeline event for lack of room state; but it must still clear the invite so the invited user’s /sync reflects the rescission. Mirrors Synapse’s out-of-band membership handling: only a kick from the original inviter is honored, since without the room state we cannot judge any other sender’s authority. Returns true when a rescission was applied.

impl Service

async fn handle_prev_events( &self, origin: &ServerName, room_id: &RoomId, event_id: &EventId, sorted_prev_events: Vec<OwnedEventId>, eventid_info: HashMap<OwnedEventId, (PduEvent, CanonicalJsonObject)>, room_version: &RoomVersionId, recursion_level: usize, first_ts_in_room: MilliSecondsSinceUnixEpoch, create_event_id: &EventId, ) -> Result<()>

Upgrade an incoming PDU’s previous events, walking interior events after their parents so each derives state locally instead of refetching it.

impl Service

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

Upgrade one previous event, folding a transient failure into a no-op so a single bad prev does not abort the batch; a shutdown still propagates.

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(super) async fn cached_resolved_state( &self, event_id: &EventId, ) -> Option<HashMap<u64, OwnedEventId>>

State resolved for an event not in our timeline, keyed by event id, held only to spare a repeat /state_ids fetch on the next walk. The value is a shortstatehash into the shared compressor tables, never the authoritative shorteventid_shortstatehash, so no authoritative state read observes it.

impl Service

pub(super) async fn cache_resolved_state( &self, room_id: &RoomId, event_id: &EventId, state: Arc<CompressedState>, )

Persist the state resolved for event_id over federation so a later walk of the same event resolves without another fetch. Best effort: a failed compressor write leaves the next walk to refetch.

impl Service

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

impl Service

async fn invited_room_version(&self, room_id: &RoomId) -> Option<RoomVersionId>

Recover a room’s version from a locally-invited member’s stored stripped state, for a room we are not resident in (e.g. a rescinded out-of-band invite). The create event in the stripped state carries the version.

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

async fn resolve_state_at_incoming_event( &self, origin: &ServerName, room_id: &RoomId, incoming_pdu: &PduEvent, room_version: &RoomVersionId, recursion_level: usize, create_event_id: &EventId, ) -> Result<(HashMap<u64, OwnedEventId>, bool)>

impl Service

async fn auth_check_outlier_pdu( &self, room_id: &RoomId, incoming_pdu: &PduEvent, room_rules: &RoomVersionRules, state_at_incoming_event: &HashMap<u64, OwnedEventId>, ) -> Result

impl Service

async fn compute_soft_fail( &self, incoming_pdu: &PduEvent, room_rules: &RoomVersionRules, pdu_json: &mut CanonicalJsonObject, ) -> Result<bool>

impl Service

async fn compute_remaining_extremities( &self, room_id: &RoomId, incoming_pdu: &PduEvent, ) -> Vec<OwnedEventId>

impl Service

async fn resolve_and_force_state_after( &self, room_id: &RoomId, room_version: &RoomVersionId, incoming_pdu: &PduEvent, state_at_incoming_event: &HashMap<u64, OwnedEventId>, state_lock: &RoomMutexGuard, ) -> Result

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.