0
0
Fork 1
mirror of https://mau.dev/maunium/synapse.git synced 2024-12-15 02:23:49 +01:00

Merge pull request #3255 from matrix-org/rav/fix_transactions

Stop the transaction cache caching failures
This commit is contained in:
Richard van der Hoff 2018-05-21 17:58:08 +01:00 committed by GitHub
commit 413482f578
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 93 additions and 18 deletions

View file

@ -19,6 +19,7 @@ import logging
from synapse.api.auth import get_access_token_from_request from synapse.api.auth import get_access_token_from_request
from synapse.util.async import ObservableDeferred from synapse.util.async import ObservableDeferred
from synapse.util.logcontext import make_deferred_yieldable, run_in_background
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -80,27 +81,26 @@ class HttpTransactionCache(object):
Returns: Returns:
Deferred which resolves to a tuple of (response_code, response_dict). Deferred which resolves to a tuple of (response_code, response_dict).
""" """
try: if txn_key in self.transactions:
return self.transactions[txn_key][0].observe() observable = self.transactions[txn_key][0]
except (KeyError, IndexError): else:
pass # execute the function instead. # execute the function instead.
deferred = run_in_background(fn, *args, **kwargs)
deferred = fn(*args, **kwargs) observable = ObservableDeferred(deferred)
self.transactions[txn_key] = (observable, self.clock.time_msec())
# if the request fails with a Twisted failure, remove it # if the request fails with an exception, remove it
# from the transaction map. This is done to ensure that we don't # from the transaction map. This is done to ensure that we don't
# cache transient errors like rate-limiting errors, etc. # cache transient errors like rate-limiting errors, etc.
def remove_from_map(err): def remove_from_map(err):
self.transactions.pop(txn_key, None) self.transactions.pop(txn_key, None)
return err # we deliberately do not propagate the error any further, as we
deferred.addErrback(remove_from_map) # expect the observers to have reported it.
# We don't add any other errbacks to the raw deferred, so we ask deferred.addErrback(remove_from_map)
# ObservableDeferred to swallow the error. This is fine as the error will
# still be reported to the observers. return make_deferred_yieldable(observable.observe())
observable = ObservableDeferred(deferred, consumeErrors=True)
self.transactions[txn_key] = (observable, self.clock.time_msec())
return observable.observe()
def _cleanup(self): def _cleanup(self):
now = self.clock.time_msec() now = self.clock.time_msec()

View file

@ -2,6 +2,9 @@ from synapse.rest.client.transactions import HttpTransactionCache
from synapse.rest.client.transactions import CLEANUP_PERIOD_MS from synapse.rest.client.transactions import CLEANUP_PERIOD_MS
from twisted.internet import defer from twisted.internet import defer
from mock import Mock, call from mock import Mock, call
from synapse.util import async
from synapse.util.logcontext import LoggingContext
from tests import unittest from tests import unittest
from tests.utils import MockClock from tests.utils import MockClock
@ -39,6 +42,78 @@ class HttpTransactionCacheTestCase(unittest.TestCase):
# expect only a single call to do the work # expect only a single call to do the work
cb.assert_called_once_with("some_arg", keyword="arg", changing_args=0) cb.assert_called_once_with("some_arg", keyword="arg", changing_args=0)
@defer.inlineCallbacks
def test_logcontexts_with_async_result(self):
@defer.inlineCallbacks
def cb():
yield async.sleep(0)
defer.returnValue("yay")
@defer.inlineCallbacks
def test():
with LoggingContext("c") as c1:
res = yield self.cache.fetch_or_execute(self.mock_key, cb)
self.assertIs(LoggingContext.current_context(), c1)
self.assertEqual(res, "yay")
# run the test twice in parallel
d = defer.gatherResults([test(), test()])
self.assertIs(LoggingContext.current_context(), LoggingContext.sentinel)
yield d
self.assertIs(LoggingContext.current_context(), LoggingContext.sentinel)
@defer.inlineCallbacks
def test_does_not_cache_exceptions(self):
"""Checks that, if the callback throws an exception, it is called again
for the next request.
"""
called = [False]
def cb():
if called[0]:
# return a valid result the second time
return defer.succeed(self.mock_http_response)
called[0] = True
raise Exception("boo")
with LoggingContext("test") as test_context:
try:
yield self.cache.fetch_or_execute(self.mock_key, cb)
except Exception as e:
self.assertEqual(e.message, "boo")
self.assertIs(LoggingContext.current_context(), test_context)
res = yield self.cache.fetch_or_execute(self.mock_key, cb)
self.assertEqual(res, self.mock_http_response)
self.assertIs(LoggingContext.current_context(), test_context)
@defer.inlineCallbacks
def test_does_not_cache_failures(self):
"""Checks that, if the callback returns a failure, it is called again
for the next request.
"""
called = [False]
def cb():
if called[0]:
# return a valid result the second time
return defer.succeed(self.mock_http_response)
called[0] = True
return defer.fail(Exception("boo"))
with LoggingContext("test") as test_context:
try:
yield self.cache.fetch_or_execute(self.mock_key, cb)
except Exception as e:
self.assertEqual(e.message, "boo")
self.assertIs(LoggingContext.current_context(), test_context)
res = yield self.cache.fetch_or_execute(self.mock_key, cb)
self.assertEqual(res, self.mock_http_response)
self.assertIs(LoggingContext.current_context(), test_context)
@defer.inlineCallbacks @defer.inlineCallbacks
def test_cleans_up(self): def test_cleans_up(self):
cb = Mock( cb = Mock(