519 lines
21 KiB
Python
519 lines
21 KiB
Python
# Copyright 2015, 2016 OpenMarket Ltd
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
"""
|
|
This module controls the reliability for application service transactions.
|
|
|
|
The nominal flow through this module looks like:
|
|
__________
|
|
1---ASa[e]-->| Service |--> Queue ASa[f]
|
|
2----ASb[e]->| Queuer |
|
|
3--ASa[f]--->|__________|-----------+ ASa[e], ASb[e]
|
|
V
|
|
-````````- +------------+
|
|
|````````|<--StoreTxn-|Transaction |
|
|
|Database| | Controller |---> SEND TO AS
|
|
`--------` +------------+
|
|
What happens on SEND TO AS depends on the state of the Application Service:
|
|
- If the AS is marked as DOWN, do nothing.
|
|
- If the AS is marked as UP, send the transaction.
|
|
* SUCCESS : Increment where the AS is up to txn-wise and nuke the txn
|
|
contents from the db.
|
|
* FAILURE : Marked AS as DOWN and start Recoverer.
|
|
|
|
Recoverer attempts to recover ASes who have died. The flow for this looks like:
|
|
,--------------------- backoff++ --------------.
|
|
V |
|
|
START ---> Wait exp ------> Get oldest txn ID from ----> FAILURE
|
|
backoff DB and try to send it
|
|
^ |___________
|
|
Mark AS as | V
|
|
UP & quit +---------- YES SUCCESS
|
|
| | |
|
|
NO <--- Have more txns? <------ Mark txn success & nuke <-+
|
|
from db; incr AS pos.
|
|
Reset backoff.
|
|
|
|
This is all tied together by the AppServiceScheduler which DIs the required
|
|
components.
|
|
"""
|
|
import logging
|
|
from typing import (
|
|
TYPE_CHECKING,
|
|
Awaitable,
|
|
Callable,
|
|
Collection,
|
|
Dict,
|
|
Iterable,
|
|
List,
|
|
Optional,
|
|
Sequence,
|
|
Set,
|
|
Tuple,
|
|
)
|
|
|
|
from synapse.appservice import (
|
|
ApplicationService,
|
|
ApplicationServiceState,
|
|
TransactionOneTimeKeysCount,
|
|
TransactionUnusedFallbackKeys,
|
|
)
|
|
from synapse.appservice.api import ApplicationServiceApi
|
|
from synapse.events import EventBase
|
|
from synapse.logging.context import run_in_background
|
|
from synapse.metrics.background_process_metrics import run_as_background_process
|
|
from synapse.storage.databases.main import DataStore
|
|
from synapse.types import DeviceListUpdates, JsonMapping
|
|
from synapse.util import Clock
|
|
|
|
if TYPE_CHECKING:
|
|
from synapse.server import HomeServer
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Maximum number of events to provide in an AS transaction.
|
|
MAX_PERSISTENT_EVENTS_PER_TRANSACTION = 100
|
|
|
|
# Maximum number of ephemeral events to provide in an AS transaction.
|
|
MAX_EPHEMERAL_EVENTS_PER_TRANSACTION = 100
|
|
|
|
# Maximum number of to-device messages to provide in an AS transaction.
|
|
MAX_TO_DEVICE_MESSAGES_PER_TRANSACTION = 100
|
|
|
|
|
|
class ApplicationServiceScheduler:
|
|
"""Public facing API for this module. Does the required DI to tie the
|
|
components together. This also serves as the "event_pool", which in this
|
|
case is a simple array.
|
|
"""
|
|
|
|
def __init__(self, hs: "HomeServer"):
|
|
self.clock = hs.get_clock()
|
|
self.store = hs.get_datastores().main
|
|
self.as_api = hs.get_application_service_api()
|
|
|
|
self.txn_ctrl = _TransactionController(self.clock, self.store, self.as_api)
|
|
self.queuer = _ServiceQueuer(self.txn_ctrl, self.clock, hs)
|
|
|
|
async def start(self) -> None:
|
|
logger.info("Starting appservice scheduler")
|
|
|
|
# check for any DOWN ASes and start recoverers for them.
|
|
services = await self.store.get_appservices_by_state(
|
|
ApplicationServiceState.DOWN
|
|
)
|
|
|
|
for service in services:
|
|
self.txn_ctrl.start_recoverer(service)
|
|
|
|
def enqueue_for_appservice(
|
|
self,
|
|
appservice: ApplicationService,
|
|
events: Optional[Collection[EventBase]] = None,
|
|
ephemeral: Optional[Collection[JsonMapping]] = None,
|
|
to_device_messages: Optional[Collection[JsonMapping]] = None,
|
|
device_list_summary: Optional[DeviceListUpdates] = None,
|
|
) -> None:
|
|
"""
|
|
Enqueue some data to be sent off to an application service.
|
|
|
|
Args:
|
|
appservice: The application service to create and send a transaction to.
|
|
events: The persistent room events to send.
|
|
ephemeral: The ephemeral events to send.
|
|
to_device_messages: The to-device messages to send. These differ from normal
|
|
to-device messages sent to clients, as they have 'to_device_id' and
|
|
'to_user_id' fields.
|
|
device_list_summary: A summary of users that the application service either needs
|
|
to refresh the device lists of, or those that the application service need no
|
|
longer track the device lists of.
|
|
"""
|
|
# We purposefully allow this method to run with empty events/ephemeral
|
|
# collections, so that callers do not need to check iterable size themselves.
|
|
if (
|
|
not events
|
|
and not ephemeral
|
|
and not to_device_messages
|
|
and not device_list_summary
|
|
):
|
|
return
|
|
|
|
if events:
|
|
self.queuer.queued_events.setdefault(appservice.id, []).extend(events)
|
|
if ephemeral:
|
|
self.queuer.queued_ephemeral.setdefault(appservice.id, []).extend(ephemeral)
|
|
if to_device_messages:
|
|
self.queuer.queued_to_device_messages.setdefault(appservice.id, []).extend(
|
|
to_device_messages
|
|
)
|
|
if device_list_summary:
|
|
self.queuer.queued_device_list_summaries.setdefault(
|
|
appservice.id, []
|
|
).append(device_list_summary)
|
|
|
|
# Kick off a new application service transaction
|
|
self.queuer.start_background_request(appservice)
|
|
|
|
|
|
class _ServiceQueuer:
|
|
"""Queue of events waiting to be sent to appservices.
|
|
|
|
Groups events into transactions per-appservice, and sends them on to the
|
|
TransactionController. Makes sure that we only have one transaction in flight per
|
|
appservice at a given time.
|
|
"""
|
|
|
|
def __init__(
|
|
self, txn_ctrl: "_TransactionController", clock: Clock, hs: "HomeServer"
|
|
):
|
|
# dict of {service_id: [events]}
|
|
self.queued_events: Dict[str, List[EventBase]] = {}
|
|
# dict of {service_id: [events]}
|
|
self.queued_ephemeral: Dict[str, List[JsonMapping]] = {}
|
|
# dict of {service_id: [to_device_message_json]}
|
|
self.queued_to_device_messages: Dict[str, List[JsonMapping]] = {}
|
|
# dict of {service_id: [device_list_summary]}
|
|
self.queued_device_list_summaries: Dict[str, List[DeviceListUpdates]] = {}
|
|
|
|
# the appservices which currently have a transaction in flight
|
|
self.requests_in_flight: Set[str] = set()
|
|
self.txn_ctrl = txn_ctrl
|
|
self.clock = clock
|
|
self._msc3202_transaction_extensions_enabled: bool = (
|
|
hs.config.experimental.msc3202_transaction_extensions
|
|
)
|
|
self._store = hs.get_datastores().main
|
|
|
|
def start_background_request(self, service: ApplicationService) -> None:
|
|
# start a sender for this appservice if we don't already have one
|
|
if service.id in self.requests_in_flight:
|
|
return
|
|
|
|
run_as_background_process("as-sender", self._send_request, service)
|
|
|
|
async def _send_request(self, service: ApplicationService) -> None:
|
|
# sanity-check: we shouldn't get here if this service already has a sender
|
|
# running.
|
|
assert service.id not in self.requests_in_flight
|
|
|
|
self.requests_in_flight.add(service.id)
|
|
try:
|
|
while True:
|
|
all_events = self.queued_events.get(service.id, [])
|
|
events = all_events[:MAX_PERSISTENT_EVENTS_PER_TRANSACTION]
|
|
del all_events[:MAX_PERSISTENT_EVENTS_PER_TRANSACTION]
|
|
|
|
all_events_ephemeral = self.queued_ephemeral.get(service.id, [])
|
|
ephemeral = all_events_ephemeral[:MAX_EPHEMERAL_EVENTS_PER_TRANSACTION]
|
|
del all_events_ephemeral[:MAX_EPHEMERAL_EVENTS_PER_TRANSACTION]
|
|
|
|
all_to_device_messages = self.queued_to_device_messages.get(
|
|
service.id, []
|
|
)
|
|
to_device_messages_to_send = all_to_device_messages[
|
|
:MAX_TO_DEVICE_MESSAGES_PER_TRANSACTION
|
|
]
|
|
del all_to_device_messages[:MAX_TO_DEVICE_MESSAGES_PER_TRANSACTION]
|
|
|
|
# Consolidate any pending device list summaries into a single, up-to-date
|
|
# summary.
|
|
# Note: this code assumes that in a single DeviceListUpdates, a user will
|
|
# never be in both "changed" and "left" sets.
|
|
device_list_summary = DeviceListUpdates()
|
|
for summary in self.queued_device_list_summaries.get(service.id, []):
|
|
# For every user in the incoming "changed" set:
|
|
# * Remove them from the existing "left" set if necessary
|
|
# (as we need to start tracking them again)
|
|
# * Add them to the existing "changed" set if necessary.
|
|
device_list_summary.left.difference_update(summary.changed)
|
|
device_list_summary.changed.update(summary.changed)
|
|
|
|
# For every user in the incoming "left" set:
|
|
# * Remove them from the existing "changed" set if necessary
|
|
# (we no longer need to track them)
|
|
# * Add them to the existing "left" set if necessary.
|
|
device_list_summary.changed.difference_update(summary.left)
|
|
device_list_summary.left.update(summary.left)
|
|
self.queued_device_list_summaries.clear()
|
|
|
|
if (
|
|
not events
|
|
and not ephemeral
|
|
and not to_device_messages_to_send
|
|
# DeviceListUpdates is True if either the 'changed' or 'left' sets have
|
|
# at least one entry, otherwise False
|
|
and not device_list_summary
|
|
):
|
|
return
|
|
|
|
one_time_keys_count: Optional[TransactionOneTimeKeysCount] = None
|
|
unused_fallback_keys: Optional[TransactionUnusedFallbackKeys] = None
|
|
|
|
if (
|
|
self._msc3202_transaction_extensions_enabled
|
|
and service.msc3202_transaction_extensions
|
|
):
|
|
# Compute the one-time key counts and fallback key usage states
|
|
# for the users which are mentioned in this transaction,
|
|
# as well as the appservice's sender.
|
|
(
|
|
one_time_keys_count,
|
|
unused_fallback_keys,
|
|
) = await self._compute_msc3202_otk_counts_and_fallback_keys(
|
|
service, events, ephemeral, to_device_messages_to_send
|
|
)
|
|
|
|
try:
|
|
await self.txn_ctrl.send(
|
|
service,
|
|
events,
|
|
ephemeral,
|
|
to_device_messages_to_send,
|
|
one_time_keys_count,
|
|
unused_fallback_keys,
|
|
device_list_summary,
|
|
)
|
|
except Exception:
|
|
logger.exception("AS request failed")
|
|
finally:
|
|
self.requests_in_flight.discard(service.id)
|
|
|
|
async def _compute_msc3202_otk_counts_and_fallback_keys(
|
|
self,
|
|
service: ApplicationService,
|
|
events: Iterable[EventBase],
|
|
ephemerals: Iterable[JsonMapping],
|
|
to_device_messages: Iterable[JsonMapping],
|
|
) -> Tuple[TransactionOneTimeKeysCount, TransactionUnusedFallbackKeys]:
|
|
"""
|
|
Given a list of the events, ephemeral messages and to-device messages,
|
|
- first computes a list of application services users that may have
|
|
interesting updates to the one-time key counts or fallback key usage.
|
|
- then computes one-time key counts and fallback key usages for those users.
|
|
Given a list of application service users that are interesting,
|
|
compute one-time key counts and fallback key usages for the users.
|
|
"""
|
|
|
|
# Set of 'interesting' users who may have updates
|
|
users: Set[str] = set()
|
|
|
|
# The sender is always included
|
|
users.add(service.sender)
|
|
|
|
# All AS users that would receive the PDUs or EDUs sent to these rooms
|
|
# are classed as 'interesting'.
|
|
rooms_of_interesting_users: Set[str] = set()
|
|
# PDUs
|
|
rooms_of_interesting_users.update(event.room_id for event in events)
|
|
# EDUs
|
|
rooms_of_interesting_users.update(
|
|
ephemeral["room_id"]
|
|
for ephemeral in ephemerals
|
|
if ephemeral.get("room_id") is not None
|
|
)
|
|
|
|
# Look up the AS users in those rooms
|
|
for room_id in rooms_of_interesting_users:
|
|
users.update(
|
|
await self._store.get_app_service_users_in_room(room_id, service)
|
|
)
|
|
|
|
# Add recipients of to-device messages.
|
|
users.update(
|
|
device_message["to_user_id"] for device_message in to_device_messages
|
|
)
|
|
|
|
# Compute and return the counts / fallback key usage states
|
|
otk_counts = await self._store.count_bulk_e2e_one_time_keys_for_as(users)
|
|
unused_fbks = await self._store.get_e2e_bulk_unused_fallback_key_types(users)
|
|
return otk_counts, unused_fbks
|
|
|
|
|
|
class _TransactionController:
|
|
"""Transaction manager.
|
|
|
|
Builds AppServiceTransactions and runs their lifecycle. Also starts a Recoverer
|
|
if a transaction fails.
|
|
|
|
(Note we have only have one of these in the homeserver.)
|
|
"""
|
|
|
|
def __init__(self, clock: Clock, store: DataStore, as_api: ApplicationServiceApi):
|
|
self.clock = clock
|
|
self.store = store
|
|
self.as_api = as_api
|
|
|
|
# map from service id to recoverer instance
|
|
self.recoverers: Dict[str, "_Recoverer"] = {}
|
|
|
|
# for UTs
|
|
self.RECOVERER_CLASS = _Recoverer
|
|
|
|
async def send(
|
|
self,
|
|
service: ApplicationService,
|
|
events: Sequence[EventBase],
|
|
ephemeral: Optional[List[JsonMapping]] = None,
|
|
to_device_messages: Optional[List[JsonMapping]] = None,
|
|
one_time_keys_count: Optional[TransactionOneTimeKeysCount] = None,
|
|
unused_fallback_keys: Optional[TransactionUnusedFallbackKeys] = None,
|
|
device_list_summary: Optional[DeviceListUpdates] = None,
|
|
) -> None:
|
|
"""
|
|
Create a transaction with the given data and send to the provided
|
|
application service.
|
|
|
|
Args:
|
|
service: The application service to send the transaction to.
|
|
events: The persistent events to include in the transaction.
|
|
ephemeral: The ephemeral events to include in the transaction.
|
|
to_device_messages: The to-device messages to include in the transaction.
|
|
one_time_keys_count: Counts of remaining one-time keys for relevant
|
|
appservice devices in the transaction.
|
|
unused_fallback_keys: Lists of unused fallback keys for relevant
|
|
appservice devices in the transaction.
|
|
device_list_summary: The device list summary to include in the transaction.
|
|
"""
|
|
try:
|
|
service_is_up = await self._is_service_up(service)
|
|
# Don't create empty txns when in recovery mode (ephemeral events are dropped)
|
|
if not service_is_up and not events:
|
|
return
|
|
|
|
txn = await self.store.create_appservice_txn(
|
|
service=service,
|
|
events=events,
|
|
ephemeral=ephemeral or [],
|
|
to_device_messages=to_device_messages or [],
|
|
one_time_keys_count=one_time_keys_count or {},
|
|
unused_fallback_keys=unused_fallback_keys or {},
|
|
device_list_summary=device_list_summary or DeviceListUpdates(),
|
|
)
|
|
if service_is_up:
|
|
sent = await txn.send(self.as_api)
|
|
if sent:
|
|
await txn.complete(self.store)
|
|
else:
|
|
run_in_background(self._on_txn_fail, service)
|
|
except Exception:
|
|
logger.exception("Error creating appservice transaction")
|
|
run_in_background(self._on_txn_fail, service)
|
|
|
|
async def on_recovered(self, recoverer: "_Recoverer") -> None:
|
|
logger.info(
|
|
"Successfully recovered application service AS ID %s", recoverer.service.id
|
|
)
|
|
self.recoverers.pop(recoverer.service.id)
|
|
logger.info("Remaining active recoverers: %s", len(self.recoverers))
|
|
await self.store.set_appservice_state(
|
|
recoverer.service, ApplicationServiceState.UP
|
|
)
|
|
|
|
async def _on_txn_fail(self, service: ApplicationService) -> None:
|
|
try:
|
|
await self.store.set_appservice_state(service, ApplicationServiceState.DOWN)
|
|
self.start_recoverer(service)
|
|
except Exception:
|
|
logger.exception("Error starting AS recoverer")
|
|
|
|
def start_recoverer(self, service: ApplicationService) -> None:
|
|
"""Start a Recoverer for the given service
|
|
|
|
Args:
|
|
service:
|
|
"""
|
|
logger.info("Starting recoverer for AS ID %s", service.id)
|
|
assert service.id not in self.recoverers
|
|
recoverer = self.RECOVERER_CLASS(
|
|
self.clock, self.store, self.as_api, service, self.on_recovered
|
|
)
|
|
self.recoverers[service.id] = recoverer
|
|
recoverer.recover()
|
|
logger.info("Now %i active recoverers", len(self.recoverers))
|
|
|
|
async def _is_service_up(self, service: ApplicationService) -> bool:
|
|
state = await self.store.get_appservice_state(service)
|
|
return state == ApplicationServiceState.UP or state is None
|
|
|
|
|
|
class _Recoverer:
|
|
"""Manages retries and backoff for a DOWN appservice.
|
|
|
|
We have one of these for each appservice which is currently considered DOWN.
|
|
|
|
Args:
|
|
clock (synapse.util.Clock):
|
|
store (synapse.storage.DataStore):
|
|
as_api (synapse.appservice.api.ApplicationServiceApi):
|
|
service (synapse.appservice.ApplicationService): the service we are managing
|
|
callback (callable[_Recoverer]): called once the service recovers.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
clock: Clock,
|
|
store: DataStore,
|
|
as_api: ApplicationServiceApi,
|
|
service: ApplicationService,
|
|
callback: Callable[["_Recoverer"], Awaitable[None]],
|
|
):
|
|
self.clock = clock
|
|
self.store = store
|
|
self.as_api = as_api
|
|
self.service = service
|
|
self.callback = callback
|
|
self.backoff_counter = 1
|
|
|
|
def recover(self) -> None:
|
|
delay = 2**self.backoff_counter
|
|
logger.info("Scheduling retries on %s in %fs", self.service.id, delay)
|
|
self.clock.call_later(
|
|
delay, run_as_background_process, "as-recoverer", self.retry
|
|
)
|
|
|
|
def _backoff(self) -> None:
|
|
# cap the backoff to be around 8.5min => (2^9) = 512 secs
|
|
if self.backoff_counter < 9:
|
|
self.backoff_counter += 1
|
|
self.recover()
|
|
|
|
async def retry(self) -> None:
|
|
logger.info("Starting retries on %s", self.service.id)
|
|
try:
|
|
while True:
|
|
txn = await self.store.get_oldest_unsent_txn(self.service)
|
|
if not txn:
|
|
# nothing left: we're done!
|
|
await self.callback(self)
|
|
return
|
|
|
|
logger.info(
|
|
"Retrying transaction %s for AS ID %s", txn.id, txn.service.id
|
|
)
|
|
sent = await txn.send(self.as_api)
|
|
if not sent:
|
|
break
|
|
|
|
await txn.complete(self.store)
|
|
|
|
# reset the backoff counter and then process the next transaction
|
|
self.backoff_counter = 1
|
|
|
|
except Exception:
|
|
logger.exception("Unexpected error running retries")
|
|
|
|
# we didn't manage to send all of the transactions before we got an error of
|
|
# some flavour: reschedule the next retry.
|
|
self._backoff()
|