diff --git a/README.rst b/README.rst
index 358038ade0..f1ccc8dc45 100644
--- a/README.rst
+++ b/README.rst
@@ -236,9 +236,6 @@ The advantages of Postgres include:
pointing at the same DB master, as well as enabling DB replication in
synapse itself.
-The only disadvantage is that the code is relatively new as of April 2015 and
-may have a few regressions relative to SQLite.
-
For information on how to install and use PostgreSQL, please see
`docs/postgres.rst <docs/postgres.rst>`_.
diff --git a/synapse/appservice/api.py b/synapse/appservice/api.py
index 775417eb21..cc4af23962 100644
--- a/synapse/appservice/api.py
+++ b/synapse/appservice/api.py
@@ -67,6 +67,8 @@ class ApplicationServiceApi(SimpleHttpClient):
@defer.inlineCallbacks
def query_user(self, service, user_id):
+ if service.url is None:
+ defer.returnValue(False)
uri = service.url + ("/users/%s" % urllib.quote(user_id))
response = None
try:
@@ -86,6 +88,8 @@ class ApplicationServiceApi(SimpleHttpClient):
@defer.inlineCallbacks
def query_alias(self, service, alias):
+ if service.url is None:
+ defer.returnValue(False)
uri = service.url + ("/rooms/%s" % urllib.quote(alias))
response = None
try:
@@ -113,6 +117,8 @@ class ApplicationServiceApi(SimpleHttpClient):
raise ValueError(
"Unrecognised 'kind' argument %r to query_3pe()", kind
)
+ if service.url is None:
+ defer.returnValue([])
uri = "%s%s/thirdparty/%s/%s" % (
service.url,
@@ -145,6 +151,9 @@ class ApplicationServiceApi(SimpleHttpClient):
defer.returnValue([])
def get_3pe_protocol(self, service, protocol):
+ if service.url is None:
+ defer.returnValue({})
+
@defer.inlineCallbacks
def _get():
uri = "%s%s/thirdparty/protocol/%s" % (
@@ -166,6 +175,9 @@ class ApplicationServiceApi(SimpleHttpClient):
@defer.inlineCallbacks
def push_bulk(self, service, events, txn_id=None):
+ if service.url is None:
+ defer.returnValue(True)
+
events = self._serialize(events)
if txn_id is None:
diff --git a/synapse/config/appservice.py b/synapse/config/appservice.py
index dfe43b0b4c..d7537e8d44 100644
--- a/synapse/config/appservice.py
+++ b/synapse/config/appservice.py
@@ -86,7 +86,7 @@ def load_appservices(hostname, config_files):
def _load_appservice(hostname, as_info, config_filename):
required_string_fields = [
- "id", "url", "as_token", "hs_token", "sender_localpart"
+ "id", "as_token", "hs_token", "sender_localpart"
]
for field in required_string_fields:
if not isinstance(as_info.get(field), basestring):
@@ -94,6 +94,14 @@ def _load_appservice(hostname, as_info, config_filename):
field, config_filename,
))
+ # 'url' must either be a string or explicitly null, not missing
+ # to avoid accidentally turning off push for ASes.
+ if (not isinstance(as_info.get("url"), basestring) and
+ as_info.get("url", "") is not None):
+ raise KeyError(
+ "Required string field or explicit null: 'url' (%s)" % (config_filename,)
+ )
+
localpart = as_info["sender_localpart"]
if urllib.quote(localpart) != localpart:
raise ValueError(
@@ -132,6 +140,13 @@ def _load_appservice(hostname, as_info, config_filename):
for p in protocols:
if not isinstance(p, str):
raise KeyError("Bad value for 'protocols' item")
+
+ if as_info["url"] is None:
+ logger.info(
+ "(%s) Explicitly empty 'url' provided. This application service"
+ " will not receive events or queries.",
+ config_filename,
+ )
return ApplicationService(
token=as_info["as_token"],
url=as_info["url"],
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 73752b2f89..d22adadc38 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -191,6 +191,13 @@ class PresenceHandler(object):
5000,
)
+ self.clock.call_later(
+ 60,
+ self.clock.looping_call,
+ self._persist_unpersisted_changes,
+ 60 * 1000,
+ )
+
metrics.register_callback("wheel_timer_size", lambda: len(self.wheel_timer))
@defer.inlineCallbacks
@@ -217,6 +224,27 @@ class PresenceHandler(object):
logger.info("Finished _on_shutdown")
@defer.inlineCallbacks
+ def _persist_unpersisted_changes(self):
+ """We periodically persist the unpersisted changes, as otherwise they
+ may stack up and slow down shutdown times.
+ """
+ logger.info(
+ "Performing _persist_unpersisted_changes. Persiting %d unpersisted changes",
+ len(self.user_to_current_state)
+ )
+
+ unpersisted = self.unpersisted_users_changes
+ self.unpersisted_users_changes = set()
+
+ if unpersisted:
+ yield self.store.update_presence([
+ self.user_to_current_state[user_id]
+ for user_id in unpersisted
+ ])
+
+ logger.info("Finished _persist_unpersisted_changes")
+
+ @defer.inlineCallbacks
def _update_states(self, new_states):
"""Updates presence of users. Sets the appropriate timeouts. Pokes
the notifier and federation if and only if the changed presence state
@@ -922,7 +950,12 @@ def should_notify(old_state, new_state):
if new_state.currently_active != old_state.currently_active:
return True
- if new_state.last_active_ts - old_state.last_active_ts > LAST_ACTIVE_GRANULARITY:
+ if new_state.last_active_ts - old_state.last_active_ts > LAST_ACTIVE_GRANULARITY:
+ # Only notify about last active bumps if we're not currently acive
+ if not (old_state.currently_active and new_state.currently_active):
+ 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.
return True
diff --git a/synapse/push/mailer.py b/synapse/push/mailer.py
index 81b1af4a19..2cafcfd8f5 100644
--- a/synapse/push/mailer.py
+++ b/synapse/push/mailer.py
@@ -338,7 +338,7 @@ class Mailer(object):
# want the generated-from-names one here otherwise we'll
# end up with, "new message from Bob in the Bob room"
room_name = yield calculate_room_name(
- state_by_room[room_id], user_id, fallback_to_members=False
+ self.store, state_by_room[room_id], user_id, fallback_to_members=False
)
my_member_event = state_by_room[room_id][("m.room.member", user_id)]
diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py
index 84993b33b3..533676814a 100644
--- a/synapse/replication/resource.py
+++ b/synapse/replication/resource.py
@@ -42,6 +42,7 @@ STREAM_NAMES = (
("pushers",),
("state",),
("caches",),
+ ("to_device",),
)
@@ -144,6 +145,7 @@ class ReplicationResource(Resource):
pushers_token,
state_token,
caches_token,
+ int(stream_token.to_device_key),
))
@request_handler()
@@ -193,6 +195,7 @@ class ReplicationResource(Resource):
yield self.pushers(writer, current_token, limit, request_streams)
yield self.state(writer, current_token, limit, request_streams)
yield self.caches(writer, current_token, limit, request_streams)
+ yield self.to_device(writer, current_token, limit, request_streams)
self.streams(writer, current_token, request_streams)
logger.info("Replicated %d rows", writer.total)
@@ -398,6 +401,20 @@ class ReplicationResource(Resource):
"position", "cache_func", "keys", "invalidation_ts"
))
+ @defer.inlineCallbacks
+ def to_device(self, writer, current_token, limit, request_streams):
+ current_position = current_token.to_device
+
+ to_device = request_streams.get("to_device")
+
+ if to_device is not None:
+ to_device_rows = yield self.store.get_all_new_device_messages(
+ to_device, current_position, limit
+ )
+ writer.write_header_and_rows("to_device", to_device_rows, (
+ "position", "user_id", "device_id", "message_json"
+ ))
+
class _Writer(object):
"""Writes the streams as a JSON object as the response to the request"""
@@ -426,7 +443,7 @@ class _Writer(object):
class _ReplicationToken(collections.namedtuple("_ReplicationToken", (
"events", "presence", "typing", "receipts", "account_data", "backfill",
- "push_rules", "pushers", "state", "caches",
+ "push_rules", "pushers", "state", "caches", "to_device",
))):
__slots__ = []
diff --git a/synapse/replication/slave/storage/deviceinbox.py b/synapse/replication/slave/storage/deviceinbox.py
index 7583d23708..64d8eb2af1 100644
--- a/synapse/replication/slave/storage/deviceinbox.py
+++ b/synapse/replication/slave/storage/deviceinbox.py
@@ -28,3 +28,15 @@ class SlavedDeviceInboxStore(BaseSlavedStore):
get_to_device_stream_token = DataStore.get_to_device_stream_token.__func__
get_new_messages_for_device = DataStore.get_new_messages_for_device.__func__
delete_messages_for_device = DataStore.delete_messages_for_device.__func__
+
+ def stream_positions(self):
+ result = super(SlavedDeviceInboxStore, self).stream_positions()
+ result["to_device"] = self._device_inbox_id_gen.get_current_token()
+ return result
+
+ def process_replication(self, result):
+ stream = result.get("to_device")
+ if stream:
+ self._device_inbox_id_gen.advance(int(stream["position"]))
+
+ return super(SlavedDeviceInboxStore, self).process_replication(result)
diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py
index 00533741af..7c0991ca55 100644
--- a/synapse/rest/client/v2_alpha/sendtodevice.py
+++ b/synapse/rest/client/v2_alpha/sendtodevice.py
@@ -40,6 +40,7 @@ class SendToDeviceRestServlet(servlet.RestServlet):
self.hs = hs
self.auth = hs.get_auth()
self.store = hs.get_datastore()
+ self.notifier = hs.get_notifier()
self.is_mine_id = hs.is_mine_id
self.txns = HttpTransactionStore()
@@ -71,9 +72,14 @@ class SendToDeviceRestServlet(servlet.RestServlet):
}
for device_id, message_content in by_device.items()
}
- local_messages[user_id] = messages_by_device
+ if messages_by_device:
+ local_messages[user_id] = messages_by_device
- yield self.store.add_messages_to_device_inbox(local_messages)
+ stream_id = yield self.store.add_messages_to_device_inbox(local_messages)
+
+ self.notifier.on_new_event(
+ "to_device", stream_id, users=local_messages.keys()
+ )
response = (200, {})
self.txns.store_client_transaction(request, txn_id, response)
diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py
index e682fe1bd1..2fa0a218b9 100644
--- a/synapse/storage/deviceinbox.py
+++ b/synapse/storage/deviceinbox.py
@@ -33,7 +33,8 @@ class DeviceInboxStore(SQLBaseStore):
messages_by_user_and_device(dict):
Dictionary of user_id to device_id to message.
Returns:
- A deferred that resolves when the messages have been inserted.
+ A deferred stream_id that resolves when the messages have been
+ inserted.
"""
def select_devices_txn(txn, user_id, devices):
@@ -81,6 +82,8 @@ class DeviceInboxStore(SQLBaseStore):
stream_id
)
+ defer.returnValue(self._device_inbox_id_gen.get_current_token())
+
def get_new_messages_for_device(
self, user_id, device_id, current_stream_id, limit=100
):
@@ -136,5 +139,44 @@ class DeviceInboxStore(SQLBaseStore):
"delete_messages_for_device", delete_messages_for_device_txn
)
+ def get_all_new_device_messages(self, last_pos, current_pos, limit):
+ """
+ Args:
+ last_pos(int):
+ current_pos(int):
+ limit(int):
+ Returns:
+ A deferred list of rows from the device inbox
+ """
+ if last_pos == current_pos:
+ return defer.succeed([])
+
+ def get_all_new_device_messages_txn(txn):
+ sql = (
+ "SELECT stream_id FROM device_inbox"
+ " WHERE ? < stream_id AND stream_id <= ?"
+ " GROUP BY stream_id"
+ " ORDER BY stream_id ASC"
+ " LIMIT ?"
+ )
+ txn.execute(sql, (last_pos, current_pos, limit))
+ stream_ids = txn.fetchall()
+ if not stream_ids:
+ return []
+ max_stream_id_in_limit = stream_ids[-1]
+
+ sql = (
+ "SELECT stream_id, user_id, device_id, message_json"
+ " FROM device_inbox"
+ " WHERE ? < stream_id AND stream_id <= ?"
+ " ORDER BY stream_id ASC"
+ )
+ txn.execute(sql, (last_pos, max_stream_id_in_limit))
+ return txn.fetchall()
+
+ return self.runInteraction(
+ "get_all_new_device_messages", get_all_new_device_messages_txn
+ )
+
def get_to_device_stream_token(self):
return self._device_inbox_id_gen.get_current_token()
diff --git a/synapse/storage/schema/delta/34/sent_txn_purge.py b/synapse/storage/schema/delta/34/sent_txn_purge.py
new file mode 100644
index 0000000000..81948e3431
--- /dev/null
+++ b/synapse/storage/schema/delta/34/sent_txn_purge.py
@@ -0,0 +1,32 @@
+# Copyright 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.
+
+from synapse.storage.engines import PostgresEngine
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def run_create(cur, database_engine, *args, **kwargs):
+ if isinstance(database_engine, PostgresEngine):
+ cur.execute("TRUNCATE sent_transactions")
+ else:
+ cur.execute("DELETE FROM sent_transactions")
+
+ cur.execute("CREATE INDEX sent_transactions_ts ON sent_transactions(ts)")
+
+
+def run_upgrade(cur, database_engine, *args, **kwargs):
+ pass
diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
index 58d4de4f1d..1c588bd46b 100644
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -387,8 +387,10 @@ class TransactionStore(SQLBaseStore):
def _cleanup_transactions(self):
now = self._clock.time_msec()
month_ago = now - 30 * 24 * 60 * 60 * 1000
+ six_hours_ago = now - 6 * 60 * 60 * 1000
def _cleanup_transactions_txn(txn):
txn.execute("DELETE FROM received_transactions WHERE ts < ?", (month_ago,))
+ txn.execute("DELETE FROM sent_transactions WHERE ts < ?", (six_hours_ago,))
return self.runInteraction("_persist_in_mem_txns", _cleanup_transactions_txn)
diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py
index b531ba8540..d9e8f634ae 100644
--- a/tests/handlers/test_presence.py
+++ b/tests/handlers/test_presence.py
@@ -115,6 +115,53 @@ class PresenceUpdateTestCase(unittest.TestCase):
),
], any_order=True)
+ def test_online_to_online_last_active_noop(self):
+ wheel_timer = Mock()
+ user_id = "@foo:bar"
+ now = 5000000
+
+ prev_state = UserPresenceState.default(user_id)
+ prev_state = prev_state.copy_and_replace(
+ state=PresenceState.ONLINE,
+ last_active_ts=now - LAST_ACTIVE_GRANULARITY - 10,
+ currently_active=True,
+ )
+
+ new_state = prev_state.copy_and_replace(
+ state=PresenceState.ONLINE,
+ last_active_ts=now,
+ )
+
+ state, persist_and_notify, federation_ping = handle_update(
+ prev_state, new_state, is_mine=True, wheel_timer=wheel_timer, now=now
+ )
+
+ self.assertFalse(persist_and_notify)
+ self.assertTrue(federation_ping)
+ self.assertTrue(state.currently_active)
+ self.assertEquals(new_state.state, state.state)
+ self.assertEquals(new_state.status_msg, state.status_msg)
+ self.assertEquals(state.last_federation_update_ts, now)
+
+ self.assertEquals(wheel_timer.insert.call_count, 3)
+ wheel_timer.insert.assert_has_calls([
+ call(
+ now=now,
+ obj=user_id,
+ then=new_state.last_active_ts + IDLE_TIMER
+ ),
+ call(
+ now=now,
+ obj=user_id,
+ then=new_state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT
+ ),
+ call(
+ now=now,
+ obj=user_id,
+ then=new_state.last_active_ts + LAST_ACTIVE_GRANULARITY
+ ),
+ ], any_order=True)
+
def test_online_to_online_last_active(self):
wheel_timer = Mock()
user_id = "@foo:bar"
|
