forked-synapse/synapse/appservice/scheduler.py

312 lines
11 KiB
Python
Raw Normal View History

2016-01-06 23:26:29 -05:00
# 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.
2015-03-05 12:45:52 -05:00
Reset backoff.
This is all tied together by the AppServiceScheduler which DIs the required
components.
"""
2018-07-09 02:09:20 -04:00
import logging
from typing import List, Optional
2018-07-09 02:09:20 -04:00
from synapse.appservice import ApplicationService, ApplicationServiceState
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.types import JsonDict
2016-08-17 07:03:04 -04:00
2015-03-06 10:12:24 -05:00
logger = logging.getLogger(__name__)
2015-03-06 06:50:27 -05:00
# 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
2020-09-04 06:54:56 -04:00
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):
self.clock = hs.get_clock()
self.store = hs.get_datastore()
self.as_api = hs.get_application_service_api()
self.txn_ctrl = _TransactionController(self.clock, self.store, self.as_api)
2016-08-17 07:03:04 -04:00
self.queuer = _ServiceQueuer(self.txn_ctrl, self.clock)
2020-07-30 07:27:39 -04:00
async def start(self):
logger.info("Starting appservice scheduler")
# check for any DOWN ASes and start recoverers for them.
2020-07-30 07:27:39 -04:00
services = await self.store.get_appservices_by_state(
ApplicationServiceState.DOWN
)
for service in services:
self.txn_ctrl.start_recoverer(service)
def submit_event_for_as(self, service: ApplicationService, event: EventBase):
self.queuer.enqueue_event(service, event)
def submit_ephemeral_events_for_as(
self, service: ApplicationService, events: List[JsonDict]
):
self.queuer.enqueue_ephemeral(service, events)
2020-09-04 06:54:56 -04:00
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.
"""
2016-08-17 07:03:04 -04:00
def __init__(self, txn_ctrl, clock):
2015-03-16 10:03:16 -04:00
self.queued_events = {} # dict of {service_id: [events]}
self.queued_ephemeral = {} # dict of {service_id: [events]}
# the appservices which currently have a transaction in flight
2016-08-17 07:03:04 -04:00
self.requests_in_flight = set()
self.txn_ctrl = txn_ctrl
2016-08-17 07:03:04 -04:00
self.clock = clock
def _start_background_request(self, service):
# start a sender for this appservice if we don't already have one
2016-08-17 07:03:04 -04:00
if service.id in self.requests_in_flight:
return
run_as_background_process(
2019-06-20 05:32:02 -04:00
"as-sender-%s" % (service.id,), self._send_request, service
)
def enqueue_event(self, service: ApplicationService, event: EventBase):
self.queued_events.setdefault(service.id, []).append(event)
self._start_background_request(service)
def enqueue_ephemeral(self, service: ApplicationService, events: List[JsonDict]):
self.queued_ephemeral.setdefault(service.id, []).extend(events)
self._start_background_request(service)
async def _send_request(self, service: ApplicationService):
# sanity-check: we shouldn't get here if this service already has a sender
# running.
2019-06-20 05:32:02 -04:00
assert service.id not in self.requests_in_flight
2016-08-17 09:52:26 -04:00
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]
if not events and not ephemeral:
2016-08-17 09:52:26 -04:00
return
try:
await self.txn_ctrl.send(service, events, ephemeral)
except Exception:
logger.exception("AS request failed")
2016-08-17 09:52:26 -04:00
finally:
self.requests_in_flight.discard(service.id)
2020-09-04 06:54:56 -04:00
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.)
Args:
clock (synapse.util.Clock):
store (synapse.storage.DataStore):
as_api (synapse.appservice.api.ApplicationServiceApi):
"""
def __init__(self, clock, store, as_api):
self.clock = clock
self.store = store
self.as_api = as_api
# map from service id to recoverer instance
self.recoverers = {}
# for UTs
self.RECOVERER_CLASS = _Recoverer
async def send(
self,
service: ApplicationService,
events: List[EventBase],
ephemeral: Optional[List[JsonDict]] = None,
):
try:
txn = await self.store.create_appservice_txn(
service=service, events=events, ephemeral=ephemeral or []
)
2020-07-30 07:27:39 -04:00
service_is_up = await self._is_service_up(service)
if service_is_up:
2020-07-30 07:27:39 -04:00
sent = await txn.send(self.as_api)
if sent:
2020-07-30 07:27:39 -04:00
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)
2020-07-30 07:27:39 -04:00
async def on_recovered(self, recoverer):
2019-06-20 05:32:02 -04:00
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))
2020-07-30 07:27:39 -04:00
await self.store.set_appservice_state(
2019-06-20 05:32:02 -04:00
recoverer.service, ApplicationServiceState.UP
2015-03-06 10:12:24 -05:00
)
2020-07-30 07:27:39 -04:00
async def _on_txn_fail(self, service):
try:
2020-07-30 07:27:39 -04:00
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):
"""Start a Recoverer for the given service
Args:
service (synapse.appservice.ApplicationService):
"""
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:
2020-07-30 07:27:39 -04:00
state = await self.store.get_appservice_state(service)
return state == ApplicationServiceState.UP or state is None
2020-09-04 06:54:56 -04:00
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, store, as_api, service, callback):
self.clock = clock
self.store = store
self.as_api = as_api
self.service = service
self.callback = callback
self.backoff_counter = 1
def recover(self):
def _retry():
run_as_background_process(
2019-06-20 05:32:02 -04:00
"as-recoverer-%s" % (self.service.id,), self.retry
)
2019-06-20 05:32:02 -04:00
delay = 2 ** self.backoff_counter
logger.info("Scheduling retries on %s in %fs", self.service.id, delay)
self.clock.call_later(delay, _retry)
def _backoff(self):
# cap the backoff to be around 8.5min => (2^9) = 512 secs
if self.backoff_counter < 9:
self.backoff_counter += 1
self.recover()
2020-07-30 07:27:39 -04:00
async def retry(self):
logger.info("Starting retries on %s", self.service.id)
try:
while True:
2020-07-30 07:27:39 -04:00
txn = await self.store.get_oldest_unsent_txn(self.service)
if not txn:
# nothing left: we're done!
2020-07-30 07:27:39 -04:00
await self.callback(self)
return
2019-06-20 05:32:02 -04:00
logger.info(
"Retrying transaction %s for AS ID %s", txn.id, txn.service.id
)
2020-07-30 07:27:39 -04:00
sent = await txn.send(self.as_api)
if not sent:
break
2020-07-30 07:27:39 -04:00
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()