path: root/synapse/appservice/scheduler.py

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