Struct Service 

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

Fields

mutex_federation: MutexMap<OwnedRoomId, ()>

Serializes room federation as the outermost per-room operation.

Acquire it before the state or timeline insertion mutex for the same room. The canonical order is federation, state, then insertion.

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 async fn clear_upgrade_backoff(&self, event_id: &EventId)

Clears the upgrade backoff recorded against an event.

The soft-fail marker and this backoff gate the same retry, so operator recovery has to drop both for the next delivery to evaluate the event without waiting out the window.

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 fn handle_incoming_pdu<'a>( &'a self, __arg1: &'a ServerName, __arg2: &'a RoomId, __arg3: &'a EventId, __arg4: CanonicalJsonObject, __arg5: bool, ) -> Pin<Box<dyn Future<Output = Result<Option<(RawPduId, bool)>>> + Send + 'a>>

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

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

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

pub fn clear_policy_signature_state(&self, event_id: &EventId)

Clears the cached policy server decision for an event.

A later validation can contact the policy server again.

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.

Missing or invalid signatures are fetched again because the policy server’s key may have rotated. A fetched signature is folded into the event.

impl Service

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

where E: Event,

MSC4284: fetch a missing or invalid inbound policy server signature.

The signature is folded into pdu_json so it persists and federates onward. Refusals map to Invalid; transient failures map to Pass.

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 fork_state<'a, State>( &'a self, state: State, ) -> StateMap<OwnedEventId>

where State: Iterator<Item = (ShortStateKey, &'a OwnedEventId)> + Send + 'a,

Converts one fork branch from short state keys to typed state keys.

A later duplicate state key replaces an earlier entry. Short state key lookup failures are omitted.

impl Service

pub(super) async fn fork_chain<'a, Events>( &'a self, room_id: &'a RoomId, room_version: &'a RoomVersionId, starting_events: Events, ) -> Result<AuthSet<OwnedEventId>>

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

Collects the full auth chain for one fork branch.

The ids are distinct as AuthSet::from_distinct requires: the chain walk dedups short ids and the short-to-event-id mapping is injective. Iteration order is arbitrary.

impl Service

pub(super) async fn state_at_incoming_local<Pdu>( &self, room_id: &RoomId, incoming_pdu: &Pdu, room_version: &RoomVersionId, create_event_id: &EventId, mode: WalkMode, ) -> Result<Option<HashMap<ShortStateKey, OwnedEventId>>>

where Pdu: Event,

Build the state before incoming_pdu from events we already hold, walking locally held uncommitted ancestry down to committed or memoized ancestors with an auth gate on every folded state event. Some(map) is a complete gated build in the shape the sibling builders return; None falls back to the federation state fetch, for any reason. Err propagates only server shutdown and room-version failures.

impl Service

async fn walk_task( &self, room_id: OwnedRoomId, room_version: RoomVersionId, create_event_id: OwnedEventId, mode: WalkMode, top_prevs: PrevEvents, parent: Span, ) -> Result<Option<HashMap<ShortStateKey, OwnedEventId>>>

Walk body on its own task: a poll descends every combinator layer from the task root, and under /send intake, already the server’s deepest stack, the walk’s auth-gate subtree overflows the worker stack in debug builds.

impl Service

pub async fn local_state_report( &self, event_id: &EventId, ) -> Result<LocalBuildReport>

Run a read-only (shadow-mode) walk for one stored event and describe the outcome, for the admin debug command.

impl Service

async fn walk_state( &self, walk: &mut Walk<'_>, ) -> Result<Option<HashMap<ShortStateKey, OwnedEventId>>>

Drive discovery then the post-order build; any abnormality sets walk.fallback and yields None.

impl Service

async fn walk_discover(&self, walk: &mut Walk<'_>) -> Result

Classify the uncommitted ancestry below the incoming event with point reads only, emitting held nodes in post-order; every condition the build cannot survive sets walk.fallback here, before any state materializes.

impl Service

async fn walk_auth_present(&self, walk: &Walk<'_>, pdu: &PduEvent) -> bool

The auth gate must stay evaluable: every auth event of a held node has to be present locally before the walk commits to building through it. Hydra rooms chain the create event implied by the room id.

impl Service

async fn walk_build( &self, walk: &mut Walk<'_>, ) -> Result<Option<HashMap<ShortStateKey, OwnedEventId>>>

Compute state through the walk sub-DAG in post-order, so every node’s prevs resolve before it, then combine at the incoming event’s own prevs.

impl Service

async fn walk_node(&self, walk: &mut Walk<'_>, index: usize) -> bool

Resolve one held node: state-before from its prevs, its own gated fold on top, retained until the last consumer releases it.

impl Service

async fn state_after( &self, walk: &mut Walk<'_>, event_id: &EventId, ) -> Option<Arc<HashMap<ShortStateKey, OwnedEventId>>>

State after one prev: an already-resolved node or materialized frontier entry shares its map; otherwise the frontier materializes here.

impl Service

async fn committed_state_after( &self, walk: &mut Walk<'_>, event_id: &EventId, shortstatehash: ShortStateHash, ) -> Option<Arc<HashMap<ShortStateKey, OwnedEventId>>>

State after a committed frontier event: its stored state plus its own key folded unguarded, exactly the degree-one builder’s shape; a committed event passed full state-dependent auth at its own upgrade.

impl Service

async fn memoized_state_after( &self, walk: &mut Walk<'_>, event_id: &EventId, ) -> Option<Arc<HashMap<ShortStateKey, OwnedEventId>>>

State after a memoized frontier event: the memo row is its state-before (the column’s uniform meaning), so its own gated fold recomputes on top.

impl Service

async fn gated_fold( &self, room_rules: &RoomVersionRules, gate_drops: &mut usize, pdu: &PduEvent, before: &Arc<HashMap<ShortStateKey, OwnedEventId>>, ) -> Arc<HashMap<ShortStateKey, OwnedEventId>>

Fold the event’s own state key over its state-before, only when the position-correct auth gate passes; a rejection leaves state unchanged. Discovery pre-verified the auth events exist locally, so a gate error is a deterministic auth verdict, not an unevaluable input.

impl Service

async fn fork_resolve( &self, walk: &mut Walk<'_>, prevs: &[OwnedEventId], memo_event_id: Option<&EventId>, ) -> Option<Arc<HashMap<ShortStateKey, OwnedEventId>>>

State before a fork node, resolving the state after each of its prevs exactly as the committed-prev fork resolves today. Fork outputs are the artifacts worth memoizing; chain nodes are cheap to re-derive.

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 current_state_auth_passes( &self, room_id: &RoomId, incoming_pdu: &PduEvent, room_rules: &RoomVersionRules, ) -> bool

impl Service

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

Re-examines an event that already carries a soft-fail marker.

The marker is a standing verdict rather than a permanent rejection, so it lapses on the upgrade backoff and the event is weighed again. Asking before state resolution keeps a still-refused event cheap to decline, since a cached policy answer needs no round trip.

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>, ResolvedVia)>

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.