dandri/synapse
1
0 Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-10-24 02:03:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Partial Counter update

Browse Source
This commit is contained in:
Eric Eastwood committed 2025-05-16 17:10:22 -05:00
1 parent 72b3becb89
commit 86f39b04a6
9 files changed
+256 -172
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

synapse/appservice/api.py
+40 -32
View File
@@ -56,33 +56,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
sent_transactions_counter = Counter(
"synapse_appservice_api_sent_transactions",
"Number of /transactions/ requests sent",
["service"],
)
failed_transactions_counter = Counter(
"synapse_appservice_api_failed_transactions",
"Number of /transactions/ requests that failed to send",
["service"],
)
sent_events_counter = Counter(
"synapse_appservice_api_sent_events", "Number of events sent to the AS", ["service"]
)
sent_ephemeral_counter = Counter(
"synapse_appservice_api_sent_ephemeral",
"Number of ephemeral events sent to the AS",
["service"],
)
sent_todevice_counter = Counter(
"synapse_appservice_api_sent_todevice",
"Number of todevice messages sent to the AS",
["service"],
)
HOUR_IN_MS = 60 * 60 * 1000
@@ -133,6 +106,41 @@ class ApplicationServiceApi(SimpleHttpClient):
hs.get_clock(), "as_protocol_meta", timeout_ms=HOUR_IN_MS
)
self.sent_transactions_counter = Counter(
"synapse_appservice_api_sent_transactions",
"Number of /transactions/ requests sent",
["service"],
registry=hs.metrics_collector_registry,
)
self.failed_transactions_counter = Counter(
"synapse_appservice_api_failed_transactions",
"Number of /transactions/ requests that failed to send",
["service"],
registry=hs.metrics_collector_registry,
)
self.sent_events_counter = Counter(
"synapse_appservice_api_sent_events",
"Number of events sent to the AS",
["service"],
registry=hs.metrics_collector_registry,
)
self.sent_ephemeral_counter = Counter(
"synapse_appservice_api_sent_ephemeral",
"Number of ephemeral events sent to the AS",
["service"],
registry=hs.metrics_collector_registry,
)
self.sent_todevice_counter = Counter(
"synapse_appservice_api_sent_todevice",
"Number of todevice messages sent to the AS",
["service"],
registry=hs.metrics_collector_registry,
)
def _get_headers(self, service: "ApplicationService") -> Dict[bytes, List[bytes]]:
"""This makes sure we have always the auth header and opentracing headers set."""
@@ -395,10 +403,10 @@ class ApplicationServiceApi(SimpleHttpClient):
service.url,
[event.get("event_id") for event in events],
)
sent_transactions_counter.labels(service.id).inc()
sent_events_counter.labels(service.id).inc(len(serialized_events))
sent_ephemeral_counter.labels(service.id).inc(len(ephemeral))
sent_todevice_counter.labels(service.id).inc(len(to_device_messages))
self.sent_transactions_counter.labels(service.id).inc()
self.sent_events_counter.labels(service.id).inc(len(serialized_events))
self.sent_ephemeral_counter.labels(service.id).inc(len(ephemeral))
self.sent_todevice_counter.labels(service.id).inc(len(to_device_messages))
return True
except CodeMessageException as e:
logger.warning(
@@ -417,7 +425,7 @@ class ApplicationServiceApi(SimpleHttpClient):
ex.args,
exc_info=logger.isEnabledFor(logging.DEBUG),
)
failed_transactions_counter.labels(service.id).inc()
self.failed_transactions_counter.labels(service.id).inc()
return False
async def claim_client_keys(
synapse/federation/federation_client.py
+11 -6
View File
@@ -84,8 +84,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
sent_queries_counter = Counter("synapse_federation_client_sent_queries", "", ["type"])
PDU_RETRY_TIME_MS = 1 * 60 * 1000
@@ -167,6 +165,13 @@ class FederationClient(FederationBase):
reset_expiry_on_get=False,
)
self.sent_queries_counter = Counter(
"synapse_federation_client_sent_queries",
"",
["type"],
registry=hs.metrics_collector_registry,
)
def _clear_tried_cache(self) -> None:
"""Clear pdu_destination_tried cache"""
now = self._clock.time_msec()
@@ -206,7 +211,7 @@ class FederationClient(FederationBase):
Returns:
The JSON object from the response
"""
sent_queries_counter.labels(query_type).inc()
self.sent_queries_counter.labels(query_type).inc()
return await self.transport_layer.make_query(
destination,
@@ -228,7 +233,7 @@ class FederationClient(FederationBase):
Returns:
The JSON object from the response
"""
sent_queries_counter.labels("client_device_keys").inc()
self.sent_queries_counter.labels("client_device_keys").inc()
return await self.transport_layer.query_client_keys(
destination, content, timeout
)
@@ -239,7 +244,7 @@ class FederationClient(FederationBase):
"""Query the device keys for a list of user ids hosted on a remote
server.
"""
sent_queries_counter.labels("user_devices").inc()
self.sent_queries_counter.labels("user_devices").inc()
return await self.transport_layer.query_user_devices(
destination, user_id, timeout
)
@@ -261,7 +266,7 @@ class FederationClient(FederationBase):
Returns:
The JSON object from the response
"""
sent_queries_counter.labels("client_one_time_keys").inc()
self.sent_queries_counter.labels("client_one_time_keys").inc()
# Convert the query with counts into a stable and unstable query and check
# if attempting to claim more than 1 OTK.
synapse/federation/federation_server.py
+37 -24
View File
@@ -105,25 +105,6 @@ TRANSACTION_CONCURRENCY_LIMIT = 10
logger = logging.getLogger(__name__)
received_pdus_counter = Counter("synapse_federation_server_received_pdus", "")
received_edus_counter = Counter("synapse_federation_server_received_edus", "")
received_queries_counter = Counter(
"synapse_federation_server_received_queries", "", ["type"]
)
pdu_process_time = Histogram(
"synapse_federation_server_pdu_process_time",
"Time taken to process an event",
)
last_pdu_ts_metric = Gauge(
"synapse_federation_last_received_pdu_time",
"The timestamp of the last PDU which was successfully received from the given domain",
labelnames=("server_name",),
)
# The name of the lock to use when process events in a room received over
# federation.
@@ -185,6 +166,38 @@ class FederationServer(FederationBase):
# Whether we have started handling old events in the staging area.
self._started_handling_of_staged_events = False
self.received_pdus_counter = Counter(
"synapse_federation_server_received_pdus",
"",
registry=hs.metrics_collector_registry,
)
self.received_edus_counter = Counter(
"synapse_federation_server_received_edus",
"",
registry=hs.metrics_collector_registry,
)
self.received_queries_counter = Counter(
"synapse_federation_server_received_queries",
"",
["type"],
registry=hs.metrics_collector_registry,
)
self.pdu_process_time = Histogram(
"synapse_federation_server_pdu_process_time",
"Time taken to process an event",
registry=hs.metrics_collector_registry,
)
self.last_pdu_ts_metric = Gauge(
"synapse_federation_last_received_pdu_time",
"The timestamp of the last PDU which was successfully received from the given domain",
labelnames=("server_name",),
registry=hs.metrics_collector_registry,
)
@wrap_as_background_process("_handle_old_staged_events")
async def _handle_old_staged_events(self) -> None:
"""Handle old staged events by fetching all rooms that have staged
@@ -424,7 +437,7 @@ class FederationServer(FederationBase):
report back to the sending server.
"""
received_pdus_counter.inc(len(transaction.pdus))
self.received_pdus_counter.inc(len(transaction.pdus))
origin_host, _ = parse_server_name(origin)
@@ -535,7 +548,7 @@ class FederationServer(FederationBase):
)
if newest_pdu_ts and origin in self._federation_metrics_domains:
last_pdu_ts_metric.labels(server_name=origin).set(newest_pdu_ts / 1000)
self.last_pdu_ts_metric.labels(server_name=origin).set(newest_pdu_ts / 1000)
return pdu_results
@@ -543,7 +556,7 @@ class FederationServer(FederationBase):
"""Process the EDUs in a received transaction."""
async def _process_edu(edu_dict: JsonDict) -> None:
received_edus_counter.inc()
self.received_edus_counter.inc()
edu = Edu(
origin=origin,
@@ -658,7 +671,7 @@ class FederationServer(FederationBase):
async def on_query_request(
self, query_type: str, args: Dict[str, str]
) -> Tuple[int, Dict[str, Any]]:
received_queries_counter.labels(query_type).inc()
self.received_queries_counter.labels(query_type).inc()
resp = await self.registry.on_query(query_type, args)
return 200, resp
@@ -1299,7 +1312,7 @@ class FederationServer(FederationBase):
origin, event.event_id
)
if received_ts is not None:
pdu_process_time.observe(
self.pdu_process_time.observe(
(self._clock.time_msec() - received_ts) / 1000
)
synapse/federation/sender/__init__.py
+17 -11
View File
@@ -186,15 +186,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
sent_pdus_destination_dist_count = Counter(
"synapse_federation_client_sent_pdu_destinations_count",
"Number of PDUs queued for sending to one or more destinations",
)
sent_pdus_destination_dist_total = Counter(
"synapse_federation_client_sent_pdu_destinations",
"Total number of PDUs queued for sending across all destinations",
)
# Time (in s) to wait before trying to wake up destinations that have
# catch-up outstanding.
@@ -397,6 +388,7 @@ class FederationSender(AbstractFederationSender):
for d in self._per_destination_queues.values()
if d.transmission_loop_running
),
registry=hs.metrics_collector_registry,
)
LaterGauge(
@@ -406,6 +398,7 @@ class FederationSender(AbstractFederationSender):
lambda: sum(
d.pending_pdu_count() for d in self._per_destination_queues.values()
),
registry=hs.metrics_collector_registry,
)
LaterGauge(
"synapse_federation_transaction_queue_pending_edus",
@@ -414,6 +407,19 @@ class FederationSender(AbstractFederationSender):
lambda: sum(
d.pending_edu_count() for d in self._per_destination_queues.values()
),
registry=hs.metrics_collector_registry,
)
self.sent_pdus_destination_dist_count = Counter(
"synapse_federation_client_sent_pdu_destinations_count",
"Number of PDUs queued for sending to one or more destinations",
registry=hs.metrics_collector_registry,
)
self.sent_pdus_destination_dist_total = Counter(
"synapse_federation_client_sent_pdu_destinations",
"Total number of PDUs queued for sending across all destinations",
registry=hs.metrics_collector_registry,
)
self._is_processing = False
@@ -710,8 +716,8 @@ class FederationSender(AbstractFederationSender):
if not destinations:
return
sent_pdus_destination_dist_total.inc(len(destinations))
sent_pdus_destination_dist_count.inc()
self.sent_pdus_destination_dist_total.inc(len(destinations))
self.sent_pdus_destination_dist_count.inc()
assert pdu.internal_metadata.stream_ordering
synapse/federation/sender/per_destination_queue.py
+15 -13
View File
@@ -55,17 +55,6 @@ MAX_EDUS_PER_TRANSACTION = 100
logger = logging.getLogger(__name__)
sent_edus_counter = Counter(
"synapse_federation_client_sent_edus", "Total number of EDUs successfully sent"
)
sent_edus_by_type = Counter(
"synapse_federation_client_sent_edus_by_type",
"Number of sent EDUs successfully sent, by event type",
["type"],
)
# If the retry interval is larger than this then we enter "catchup" mode
CATCHUP_RETRY_INTERVAL = 60 * 60 * 1000
@@ -164,6 +153,19 @@ class PerDestinationQueue:
# stream_id of last successfully sent device list update.
self._last_device_list_stream_id = 0
self.sent_edus_counter = Counter(
"synapse_federation_client_sent_edus",
"Total number of EDUs successfully sent",
registry=hs.metrics_collector_registry,
)
self.sent_edus_by_type = Counter(
"synapse_federation_client_sent_edus_by_type",
"Number of sent EDUs successfully sent, by event type",
["type"],
registry=hs.metrics_collector_registry,
)
def __str__(self) -> str:
return "PerDestinationQueue[%s]" % self._destination
@@ -361,9 +363,9 @@ class PerDestinationQueue:
)
sent_transactions_counter.inc()
sent_edus_counter.inc(len(pending_edus))
self.sent_edus_counter.inc(len(pending_edus))
for edu in pending_edus:
sent_edus_by_type.labels(edu.edu_type).inc()
self.sent_edus_by_type.labels(edu.edu_type).inc()
except NotRetryingDestination as e:
logger.debug(
synapse/handlers/appservice.py
+7 -3
View File
@@ -68,8 +68,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
events_processed_counter = Counter("synapse_handlers_appservice_events_processed", "")
class ApplicationServicesHandler:
def __init__(self, hs: "HomeServer"):
@@ -95,6 +93,12 @@ class ApplicationServicesHandler:
name="appservice_ephemeral_events"
)
self.events_processed_counter = Counter(
"synapse_handlers_appservice_events_processed",
"",
registry=hs.metrics_collector_registry,
)
def notify_interested_services(self, max_token: RoomStreamToken) -> None:
"""Notifies (pushes) all application services interested in this event.
@@ -200,7 +204,7 @@ class ApplicationServicesHandler:
"appservice_sender"
).set(upper_bound)
events_processed_counter.inc(len(events))
self.events_processed_counter.inc(len(events))
event_processing_loop_room_count.labels("appservice_sender").inc(
len(events_by_room)
synapse/handlers/auth.py
+10 -9
View File
@@ -92,12 +92,6 @@ logger = logging.getLogger(__name__)
INVALID_USERNAME_OR_PASSWORD = "Invalid username or password"
invalid_login_token_counter = Counter(
"synapse_user_login_invalid_login_tokens",
"Counts the number of rejected m.login.token on /login",
["reason"],
)
def convert_client_dict_legacy_fields_to_identifier(
submission: JsonDict,
@@ -281,6 +275,13 @@ class AuthHandler:
self.msc3861_oauth_delegation_enabled = hs.config.experimental.msc3861.enabled
self.invalid_login_token_counter = Counter(
"synapse_user_login_invalid_login_tokens",
"Counts the number of rejected m.login.token on /login",
["reason"],
registry=hs.metrics_collector_registry,
)
async def validate_user_via_ui_auth(
self,
requester: Requester,
@@ -1477,11 +1478,11 @@ class AuthHandler:
try:
return await self.store.consume_login_token(login_token)
except LoginTokenExpired:
invalid_login_token_counter.labels("expired").inc()
self.invalid_login_token_counter.labels("expired").inc()
except LoginTokenReused:
invalid_login_token_counter.labels("reused").inc()
self.invalid_login_token_counter.labels("reused").inc()
except NotFoundError:
invalid_login_token_counter.labels("not found").inc()
self.invalid_login_token_counter.labels("not found").inc()
raise AuthError(403, "Invalid login token", errcode=Codes.FORBIDDEN)
synapse/handlers/federation_event.py
+39 -37
View File
@@ -105,41 +105,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
soft_failed_event_counter = Counter(
"synapse_federation_soft_failed_events_total",
"Events received over federation that we marked as soft_failed",
)
# Added to debug performance and track progress on optimizations
backfill_processing_after_timer = Histogram(
"synapse_federation_backfill_processing_after_time_seconds",
"sec",
[],
buckets=(
0.1,
0.25,
0.5,
1.0,
2.5,
5.0,
7.5,
10.0,
15.0,
20.0,
25.0,
30.0,
40.0,
50.0,
60.0,
80.0,
100.0,
120.0,
150.0,
180.0,
"+Inf",
),
)
class FederationEventHandler:
"""Handles events that originated from federation.
@@ -195,6 +160,43 @@ class FederationEventHandler:
self._room_pdu_linearizer = Linearizer("fed_room_pdu")
self.soft_failed_event_counter = Counter(
"synapse_federation_soft_failed_events_total",
"Events received over federation that we marked as soft_failed",
registry=hs.metrics_collector_registry,
)
# Added to debug performance and track progress on optimizations
self.backfill_processing_after_timer = Histogram(
"synapse_federation_backfill_processing_after_time_seconds",
"sec",
[],
buckets=(
0.1,
0.25,
0.5,
1.0,
2.5,
5.0,
7.5,
10.0,
15.0,
20.0,
25.0,
30.0,
40.0,
50.0,
60.0,
80.0,
100.0,
120.0,
150.0,
180.0,
"+Inf",
),
registry=hs.metrics_collector_registry,
)
async def on_receive_pdu(self, origin: str, pdu: EventBase) -> None:
"""Process a PDU received via a federation /send/ transaction
@@ -698,7 +700,7 @@ class FederationEventHandler:
if not events:
return
with backfill_processing_after_timer.time():
with self.backfill_processing_after_timer.time():
# if there are any events in the wrong room, the remote server is buggy and
# should not be trusted.
for ev in events:
@@ -2062,7 +2064,7 @@ class FederationEventHandler:
"hs": origin,
},
)
soft_failed_event_counter.inc()
self.soft_failed_event_counter.inc()
event.internal_metadata.soft_failed = True
async def _load_or_fetch_auth_events_for_event(
synapse/handlers/presence.py
+80 -37
View File
@@ -137,26 +137,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
notified_presence_counter = Counter("synapse_handler_presence_notified_presence", "")
federation_presence_out_counter = Counter(
"synapse_handler_presence_federation_presence_out", ""
)
presence_updates_counter = Counter("synapse_handler_presence_presence_updates", "")
timers_fired_counter = Counter("synapse_handler_presence_timers_fired", "")
federation_presence_counter = Counter(
"synapse_handler_presence_federation_presence", ""
)
bump_active_time_counter = Counter("synapse_handler_presence_bump_active_time", "")
get_updates_counter = Counter("synapse_handler_presence_get_updates", "", ["type"])
notify_reason_counter = Counter(
"synapse_handler_presence_notify_reason", "", ["locality", "reason"]
)
state_transition_counter = Counter(
"synapse_handler_presence_state_transition", "", ["locality", "from", "to"]
)
# If a user was last active in the last LAST_ACTIVE_GRANULARITY, consider them
# "currently_active"
LAST_ACTIVE_GRANULARITY = 60 * 1000
@@ -222,6 +202,19 @@ class BasePresenceHandler(abc.ABC):
# The combined status across all user devices.
self.user_to_current_state = {state.user_id: state for state in active_presence}
self.notify_reason_counter = Counter(
"synapse_handler_presence_notify_reason",
"",
["locality", "reason"],
registry=hs.metrics_collector_registry,
)
self.state_transition_counter = Counter(
"synapse_handler_presence_state_transition",
"",
["locality", "from", "to"],
registry=hs.metrics_collector_registry,
)
@abc.abstractmethod
async def user_syncing(
self,
@@ -666,7 +659,7 @@ class WorkerPresenceHandler(BasePresenceHandler):
old_state = self.user_to_current_state.get(new_state.user_id)
self.user_to_current_state[new_state.user_id] = new_state
is_mine = self.is_mine_id(new_state.user_id)
if not old_state or should_notify(old_state, new_state, is_mine):
if not old_state or should_notify(self, old_state, new_state, is_mine):
state_to_notify.append(new_state)
stream_id = token
@@ -761,6 +754,38 @@ class PresenceHandler(BasePresenceHandler):
"",
[],
lambda: len(self.user_to_current_state),
registry=hs.metrics_collector_registry,
)
self.notified_presence_counter = Counter(
"synapse_handler_presence_notified_presence",
"",
registry=hs.metrics_collector_registry,
)
self.federation_presence_out_counter = Counter(
"synapse_handler_presence_federation_presence_out",
"",
registry=hs.metrics_collector_registry,
)
self.presence_updates_counter = Counter(
"synapse_handler_presence_presence_updates",
"",
registry=hs.metrics_collector_registry,
)
self.timers_fired_counter = Counter(
"synapse_handler_presence_timers_fired",
"",
registry=hs.metrics_collector_registry,
)
self.federation_presence_counter = Counter(
"synapse_handler_presence_federation_presence",
"",
registry=hs.metrics_collector_registry,
)
self.bump_active_time_counter = Counter(
"synapse_handler_presence_bump_active_time",
"",
registry=hs.metrics_collector_registry,
)
# The per-device presence state, maps user to devices to per-device presence state.
@@ -966,6 +991,7 @@ class PresenceHandler(BasePresenceHandler):
)
new_state, should_notify, should_ping = handle_update(
self,
prev_state,
new_state,
is_mine=self.is_mine_id(user_id),
@@ -988,10 +1014,10 @@ class PresenceHandler(BasePresenceHandler):
# TODO: We should probably ensure there are no races hereafter
presence_updates_counter.inc(len(new_states))
self.presence_updates_counter.inc(len(new_states))
if to_notify:
notified_presence_counter.inc(len(to_notify))
self.notified_presence_counter.inc(len(to_notify))
await self._persist_and_notify(list(to_notify.values()))
self.unpersisted_users_changes |= {s.user_id for s in new_states}
@@ -1010,7 +1036,7 @@ class PresenceHandler(BasePresenceHandler):
if user_id not in to_notify
}
if to_federation_ping:
federation_presence_out_counter.inc(len(to_federation_ping))
self.federation_presence_out_counter.inc(len(to_federation_ping))
hosts_to_states = await get_interested_remotes(
self.store,
@@ -1060,7 +1086,7 @@ class PresenceHandler(BasePresenceHandler):
for user_id in users_to_check
]
timers_fired_counter.inc(len(states))
self.timers_fired_counter.inc(len(states))
# Set of user ID & device IDs which are currently syncing.
syncing_user_devices = {
@@ -1094,7 +1120,7 @@ class PresenceHandler(BasePresenceHandler):
user_id = user.to_string()
bump_active_time_counter.inc()
self.bump_active_time_counter.inc()
now = self.clock.time_msec()
@@ -1346,7 +1372,7 @@ class PresenceHandler(BasePresenceHandler):
updates.append(prev_state.copy_and_replace(**new_fields))
if updates:
federation_presence_counter.inc(len(updates))
self.federation_presence_counter.inc(len(updates))
await self._update_states(updates)
async def set_state(
@@ -1655,7 +1681,10 @@ class PresenceHandler(BasePresenceHandler):
def should_notify(
old_state: UserPresenceState, new_state: UserPresenceState, is_mine: bool
base_presence_handler: BasePresenceHandler,
old_state: UserPresenceState,
new_state: UserPresenceState,
is_mine: bool,
) -> bool:
"""Decides if a presence state change should be sent to interested parties."""
user_location = "remote"
@@ -1666,19 +1695,25 @@ def should_notify(
return False
if old_state.status_msg != new_state.status_msg:
notify_reason_counter.labels(user_location, "status_msg_change").inc()
base_presence_handler.notify_reason_counter.labels(
user_location, "status_msg_change"
).inc()
return True
if old_state.state != new_state.state:
notify_reason_counter.labels(user_location, "state_change").inc()
state_transition_counter.labels(
base_presence_handler.notify_reason_counter.labels(
user_location, "state_change"
).inc()
base_presence_handler.state_transition_counter.labels(
user_location, old_state.state, new_state.state
).inc()
return True
if old_state.state == PresenceState.ONLINE:
if new_state.currently_active != old_state.currently_active:
notify_reason_counter.labels(user_location, "current_active_change").inc()
base_presence_handler.notify_reason_counter.labels(
user_location, "current_active_change"
).inc()
return True
if (
@@ -1687,14 +1722,14 @@ def should_notify(
):
# Only notify about last active bumps if we're not currently active
if not new_state.currently_active:
notify_reason_counter.labels(
base_presence_handler.notify_reason_counter.labels(
user_location, "last_active_change_online"
).inc()
return True
elif new_state.last_active_ts - old_state.last_active_ts > LAST_ACTIVE_GRANULARITY:
# Always notify for a transition where last active gets bumped.
notify_reason_counter.labels(
base_presence_handler.notify_reason_counter.labels(
user_location, "last_active_change_not_online"
).inc()
return True
@@ -1764,6 +1799,13 @@ class PresenceEventSource(EventSource[int, UserPresenceState]):
self.clock = hs.get_clock()
self.store = hs.get_datastores().main
self.get_updates_counter = Counter(
"synapse_handler_presence_get_updates",
"",
["type"],
registry=hs.metrics_collector_registry,
)
async def get_new_events(
self,
user: UserID,
@@ -1870,7 +1912,7 @@ class PresenceEventSource(EventSource[int, UserPresenceState]):
# If we have the full list of changes for presence we can
# simply check which ones share a room with the user.
get_updates_counter.labels("stream").inc()
self.get_updates_counter.labels("stream").inc()
sharing_users = await self.store.do_users_share_a_room(
user_id, updated_users
@@ -1883,7 +1925,7 @@ class PresenceEventSource(EventSource[int, UserPresenceState]):
else:
# Too many possible updates. Find all users we can see and check
# if any of them have changed.
get_updates_counter.labels("full").inc()
self.get_updates_counter.labels("full").inc()
users_interested_in = (
await self.store.get_users_who_share_room_with_user(user_id)
@@ -2130,6 +2172,7 @@ def handle_timeout(
def handle_update(
base_presence_handler: BasePresenceHandler,
prev_state: UserPresenceState,
new_state: UserPresenceState,
is_mine: bool,
@@ -2213,7 +2256,7 @@ def handle_update(
)
# Check whether the change was something worth notifying about
if should_notify(prev_state, new_state, is_mine):
if should_notify(base_presence_handler, prev_state, new_state, is_mine):
new_state = new_state.copy_and_replace(last_federation_update_ts=now)
persist_and_notify = True