summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--common/src/leap/soledad/common/couch.py297
-rw-r--r--common/src/leap/soledad/common/ddocs/syncs/updates/put.js151
-rw-r--r--common/src/leap/soledad/common/ddocs/syncs/views/log/map.js12
-rw-r--r--common/src/leap/soledad/common/tests/test_couch.py8
-rw-r--r--common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py53
-rw-r--r--server/pkg/requirements.pip1
-rw-r--r--server/src/leap/soledad/server/caching.py32
-rw-r--r--server/src/leap/soledad/server/state.py141
-rw-r--r--server/src/leap/soledad/server/sync.py179
9 files changed, 297 insertions, 577 deletions
diff --git a/common/src/leap/soledad/common/couch.py b/common/src/leap/soledad/common/couch.py
index 1c762036..38041c09 100644
--- a/common/src/leap/soledad/common/couch.py
+++ b/common/src/leap/soledad/common/couch.py
@@ -26,13 +26,12 @@ import logging
import binascii
import time
import sys
-import threading
from StringIO import StringIO
-from collections import defaultdict
from urlparse import urljoin
from contextlib import contextmanager
+from multiprocessing.pool import ThreadPool
from couchdb.client import Server, Database
@@ -103,6 +102,7 @@ class CouchDocument(SoledadDocument):
SoledadDocument.__init__(self, doc_id, rev, json, has_conflicts)
self.couch_rev = None
self.transactions = None
+ self._conflicts = None
def get_conflicts(self):
"""
@@ -111,7 +111,7 @@ class CouchDocument(SoledadDocument):
:return: The conflicted versions of the document.
:rtype: [CouchDocument]
"""
- return self._conflicts
+ return self._conflicts or []
def set_conflicts(self, conflicts):
"""
@@ -357,61 +357,19 @@ def couch_server(url):
:type url: str
"""
session = Session(timeout=COUCH_TIMEOUT)
- server = Server(url=url, session=session)
+ server = Server(url=url, full_commit=False, session=session)
yield server
+THREAD_POOL = ThreadPool(20)
+
+
class CouchDatabase(CommonBackend):
"""
A U1DB implementation that uses CouchDB as its persistence layer.
"""
- # We spawn threads to parallelize the CouchDatabase.get_docs() method
- MAX_GET_DOCS_THREADS = 20
-
- update_handler_lock = defaultdict(threading.Lock)
- sync_info_lock = defaultdict(threading.Lock)
-
- class _GetDocThread(threading.Thread):
-
- """
- A thread that gets a document from a database.
-
- TODO: switch this for a twisted deferred to thread. This depends on
- replacing python-couchdb for paisley in this module.
- """
-
- def __init__(self, db, doc_id, check_for_conflicts,
- release_fun):
- """
- :param db: The database from where to get the document.
- :type db: CouchDatabase
- :param doc_id: The doc_id of the document to be retrieved.
- :type doc_id: str
- :param check_for_conflicts: Whether the get_doc() method should
- check for existing conflicts.
- :type check_for_conflicts: bool
- :param release_fun: A function that releases a semaphore, to be
- called after the document is fetched.
- :type release_fun: function
- """
- threading.Thread.__init__(self)
- self._db = db
- self._doc_id = doc_id
- self._check_for_conflicts = check_for_conflicts
- self._release_fun = release_fun
- self._doc = None
-
- def run(self):
- """
- Fetch the document, store it as a property, and call the release
- function.
- """
- self._doc = self._db._get_doc(
- self._doc_id, self._check_for_conflicts)
- self._release_fun()
-
@classmethod
def open_database(cls, url, create, replica_uid=None, ensure_ddocs=False):
"""
@@ -476,9 +434,23 @@ class CouchDatabase(CommonBackend):
self._set_replica_uid(replica_uid)
if ensure_ddocs:
self.ensure_ddocs_on_db()
- # initialize a thread pool for parallelizing get_docs()
- self._sem_pool = threading.BoundedSemaphore(
- value=self.MAX_GET_DOCS_THREADS)
+ self._cache = None
+
+ @property
+ def cache(self):
+ if self._cache is not None:
+ return self._cache
+ else:
+ return {}
+
+ def init_caching(self, cache):
+ """
+ Start using cache by setting internal _cache attribute.
+
+ :param cache: the cache instance, anything that behaves like a dict
+ :type cache: dict
+ """
+ self._cache = cache
def ensure_ddocs_on_db(self):
"""
@@ -557,10 +529,14 @@ class CouchDatabase(CommonBackend):
:rtype: str
"""
if self._real_replica_uid is not None:
+ self.cache[self._url] = {'replica_uid': self._real_replica_uid}
return self._real_replica_uid
+ if self._url in self.cache:
+ return self.cache[self._url]['replica_uid']
try:
# grab replica_uid from server
doc = self._database['u1db_config']
+ self.cache[self._url] = doc
self._real_replica_uid = doc['replica_uid']
return self._real_replica_uid
except ResourceNotFound:
@@ -595,10 +571,13 @@ class CouchDatabase(CommonBackend):
unknown reason.
"""
# query a couch list function
+ if self.replica_uid + '_gen' in self.cache:
+ return self.cache[self.replica_uid + '_gen']['generation']
ddoc_path = ['_design', 'transactions', '_list', 'generation', 'log']
res = self._database.resource(*ddoc_path)
try:
response = res.get_json()
+ self.cache[self.replica_uid + '_gen'] = response[2]
return response[2]['generation']
except ResourceNotFound as e:
raise_missing_design_doc_error(e, ddoc_path)
@@ -626,11 +605,15 @@ class CouchDatabase(CommonBackend):
design document for an yet
unknown reason.
"""
+ if self.replica_uid + '_gen' in self.cache:
+ response = self.cache[self.replica_uid + '_gen']
+ return (response['generation'], response['transaction_id'])
# query a couch list function
ddoc_path = ['_design', 'transactions', '_list', 'generation', 'log']
res = self._database.resource(*ddoc_path)
try:
response = res.get_json()
+ self.cache[self.replica_uid + '_gen'] = response[2]
return (response[2]['generation'], response[2]['transaction_id'])
except ResourceNotFound as e:
raise_missing_design_doc_error(e, ddoc_path)
@@ -734,6 +717,10 @@ class CouchDatabase(CommonBackend):
attachments=True)[2]
except ResourceNotFound:
return None
+ return self.__parse_doc_from_couch(result, doc_id, check_for_conflicts)
+
+ def __parse_doc_from_couch(self, result, doc_id,
+ check_for_conflicts=False):
# restrict to u1db documents
if 'u1db_rev' not in result:
return None
@@ -798,11 +785,8 @@ class CouchDatabase(CommonBackend):
"""
generation = self._get_generation()
- results = []
- for row in self._database.view('_all_docs'):
- doc = self.get_doc(row.id, include_deleted=include_deleted)
- if doc is not None:
- results.append(doc)
+ results = list(self.get_docs(self._database,
+ include_deleted=include_deleted))
return (generation, results)
def _put_doc(self, old_doc, doc):
@@ -890,6 +874,10 @@ class CouchDatabase(CommonBackend):
doc.doc_id, body=buf.getvalue(), headers=envelope.headers)
except ResourceConflict:
raise RevisionConflict()
+ if self.replica_uid + '_gen' in self.cache:
+ gen_info = self.cache[self.replica_uid + '_gen']
+ gen_info['generation'] += 1
+ gen_info['transaction_id'] = transactions[-1][1]
def put_doc(self, doc):
"""
@@ -1092,14 +1080,22 @@ class CouchDatabase(CommonBackend):
synchronized with the replica, this is (0, '').
:rtype: (int, str)
"""
- # query a couch view
- result = self._database.view('syncs/log')
- if len(result[other_replica_uid].rows) == 0:
- return (0, '')
- return (
- result[other_replica_uid].rows[0]['value']['known_generation'],
- result[other_replica_uid].rows[0]['value']['known_transaction_id']
- )
+ if other_replica_uid in self.cache:
+ return self.cache[other_replica_uid]
+
+ doc_id = 'u1db_sync_%s' % other_replica_uid
+ try:
+ doc = self._database[doc_id]
+ except ResourceNotFound:
+ doc = {
+ '_id': doc_id,
+ 'generation': 0,
+ 'transaction_id': '',
+ }
+ self._database.save(doc)
+ result = doc['generation'], doc['transaction_id']
+ self.cache[other_replica_uid] = result
+ return result
def _set_replica_gen_and_trans_id(self, other_replica_uid,
other_generation, other_transaction_id,
@@ -1159,42 +1155,17 @@ class CouchDatabase(CommonBackend):
:type doc_idx: int
:param sync_id: The id of the current sync session.
:type sync_id: str
-
- :raise MissingDesignDocError: Raised when tried to access a missing
- design document.
- :raise MissingDesignDocListFunctionError: Raised when trying to access
- a missing list function on a
- design document.
- :raise MissingDesignDocNamedViewError: Raised when trying to access a
- missing named view on a design
- document.
- :raise MissingDesignDocDeletedError: Raised when trying to access a
- deleted design document.
- :raise MissingDesignDocUnknownError: Raised when failed to access a
- design document for an yet
- unknown reason.
"""
- # query a couch update function
- ddoc_path = ['_design', 'syncs', '_update', 'put', 'u1db_sync_log']
- res = self._database.resource(*ddoc_path)
+ self.cache[other_replica_uid] = (other_generation,
+ other_transaction_id)
+ doc_id = 'u1db_sync_%s' % other_replica_uid
try:
- with CouchDatabase.update_handler_lock[self._get_replica_uid()]:
- body = {
- 'other_replica_uid': other_replica_uid,
- 'other_generation': other_generation,
- 'other_transaction_id': other_transaction_id,
- }
- if number_of_docs is not None:
- body['number_of_docs'] = number_of_docs
- if doc_idx is not None:
- body['doc_idx'] = doc_idx
- if sync_id is not None:
- body['sync_id'] = sync_id
- res.put_json(
- body=body,
- headers={'content-type': 'application/json'})
- except ResourceNotFound as e:
- raise_missing_design_doc_error(e, ddoc_path)
+ doc = self._database[doc_id]
+ except ResourceNotFound:
+ doc = {'_id': doc_id}
+ doc['generation'] = other_generation
+ doc['transaction_id'] = other_transaction_id
+ self._database.save(doc)
def _force_doc_sync_conflict(self, doc):
"""
@@ -1203,10 +1174,11 @@ class CouchDatabase(CommonBackend):
:param doc: The document to be put.
:type doc: CouchDocument
"""
- my_doc = self._get_doc(doc.doc_id, check_for_conflicts=True)
- doc.prune_conflicts(
- vectorclock.VectorClockRev(doc.rev), self._replica_uid)
- doc.add_conflict(my_doc)
+ my_doc = self._get_doc(doc.doc_id)
+ self._prune_conflicts(doc, vectorclock.VectorClockRev(doc.rev))
+ doc.add_conflict(self._factory(doc.doc_id, my_doc.rev,
+ my_doc.get_json()))
+ doc.has_conflicts = True
self._put_doc(my_doc, doc)
def resolve_doc(self, doc, conflicted_doc_revs):
@@ -1320,42 +1292,27 @@ class CouchDatabase(CommonBackend):
"""
if not isinstance(doc, CouchDocument):
doc = self._factory(doc.doc_id, doc.rev, doc.get_json())
- self._save_source_info(replica_uid, replica_gen,
- replica_trans_id, number_of_docs,
- doc_idx, sync_id)
my_doc = self._get_doc(doc.doc_id, check_for_conflicts=True)
- if my_doc is not None:
- my_doc.set_conflicts(
- self.get_doc_conflicts(my_doc.doc_id, my_doc.couch_rev))
- state, save_doc = _process_incoming_doc(
- my_doc, doc, save_conflict, self.replica_uid)
- if save_doc:
- self._put_doc(my_doc, save_doc)
- doc.update(save_doc)
- return state, self._get_generation()
-
- def _save_source_info(self, replica_uid, replica_gen, replica_trans_id,
- number_of_docs, doc_idx, sync_id):
- """
- Validate and save source information.
- """
- self._validate_source(replica_uid, replica_gen, replica_trans_id)
- self._set_replica_gen_and_trans_id(
- replica_uid, replica_gen, replica_trans_id,
- number_of_docs=number_of_docs, doc_idx=doc_idx,
- sync_id=sync_id)
+ if my_doc:
+ doc.set_conflicts(my_doc.get_conflicts())
+ return CommonBackend._put_doc_if_newer(self, doc, save_conflict,
+ replica_uid, replica_gen,
+ replica_trans_id)
+
+ def _put_and_update_indexes(self, cur_doc, doc):
+ self._put_doc(cur_doc, doc)
def get_docs(self, doc_ids, check_for_conflicts=True,
include_deleted=False):
"""
Get the JSON content for many documents.
- :param doc_ids: A list of document identifiers.
+ :param doc_ids: A list of document identifiers or None for all.
:type doc_ids: list
:param check_for_conflicts: If set to False, then the conflict check
will be skipped, and 'None' will be
returned instead of True/False.
- :type check_for_conflictsa: bool
+ :type check_for_conflicts: bool
:param include_deleted: If set to True, deleted documents will be
returned with empty content. Otherwise deleted
documents will not be included in the results.
@@ -1363,31 +1320,14 @@ class CouchDatabase(CommonBackend):
in matching doc_ids order.
:rtype: iterable
"""
- # Workaround for:
- #
- # http://bugs.python.org/issue7980
- # https://leap.se/code/issues/5449
- #
- # python-couchdb uses time.strptime, which is not thread safe. In
- # order to avoid the problem described on the issues above, we preload
- # strptime here by evaluating the conversion of an arbitrary date.
- # This will not be needed when/if we switch from python-couchdb to
- # paisley.
- time.strptime('Mar 8 1917', '%b %d %Y')
- # spawn threads to retrieve docs
- threads = []
- for doc_id in doc_ids:
- self._sem_pool.acquire()
- t = self._GetDocThread(self, doc_id, check_for_conflicts,
- self._sem_pool.release)
- t.start()
- threads.append(t)
- # join threads and yield docs
- for t in threads:
- t.join()
- if t._doc.is_tombstone() and not include_deleted:
+ get_one = lambda doc_id: self._get_doc(doc_id, check_for_conflicts)
+ docs = [THREAD_POOL.apply_async(get_one, [doc_id])
+ for doc_id in doc_ids]
+ for doc in docs:
+ doc = doc.get()
+ if not doc or not include_deleted and doc.is_tombstone():
continue
- yield t._doc
+ yield doc
def _prune_conflicts(self, doc, doc_vcr):
"""
@@ -1459,10 +1399,11 @@ class CouchServerState(ServerState):
:return: The CouchDatabase object.
:rtype: CouchDatabase
"""
- return CouchDatabase(
+ db = CouchDatabase(
self.couch_url,
dbname,
ensure_ddocs=False)
+ return db
def ensure_database(self, dbname):
"""
@@ -1494,53 +1435,3 @@ class CouchServerState(ServerState):
delete databases.
"""
raise Unauthorized()
-
-
-def _process_incoming_doc(my_doc, other_doc, save_conflict, replica_uid):
- """
- Check document, save and return state.
- """
- # at this point, `doc` has arrived from the other syncing party, and
- # we will decide what to do with it.
- # First, we prepare the arriving doc to update couch database.
- new_doc = CouchDocument(
- other_doc.doc_id, other_doc.rev, other_doc.get_json())
- if my_doc is None:
- return 'inserted', new_doc
- new_doc.couch_rev = my_doc.couch_rev
- new_doc.set_conflicts(my_doc.get_conflicts())
- # fetch conflicts because we will eventually manipulate them
- # from now on, it works just like u1db sqlite backend
- doc_vcr = vectorclock.VectorClockRev(new_doc.rev)
- cur_vcr = vectorclock.VectorClockRev(my_doc.rev)
- if doc_vcr.is_newer(cur_vcr):
- rev = new_doc.rev
- new_doc.prune_conflicts(doc_vcr, replica_uid)
- if new_doc.rev != rev:
- # conflicts have been autoresolved
- return 'superseded', new_doc
- else:
- return'inserted', new_doc
- elif new_doc.rev == my_doc.rev:
- # magical convergence
- return 'converged', None
- elif cur_vcr.is_newer(doc_vcr):
- # Don't add this to seen_ids, because we have something newer,
- # so we should send it back, and we should not generate a
- # conflict
- other_doc.update(new_doc)
- return 'superseded', None
- elif my_doc.same_content_as(new_doc):
- # the documents have been edited to the same thing at both ends
- doc_vcr.maximize(cur_vcr)
- doc_vcr.increment(replica_uid)
- new_doc.rev = doc_vcr.as_str()
- return 'superseded', new_doc
- else:
- if save_conflict:
- new_doc.prune_conflicts(
- vectorclock.VectorClockRev(new_doc.rev), replica_uid)
- new_doc.add_conflict(my_doc)
- return 'conflicted', new_doc
- other_doc.update(new_doc)
- return 'conflicted', None
diff --git a/common/src/leap/soledad/common/ddocs/syncs/updates/put.js b/common/src/leap/soledad/common/ddocs/syncs/updates/put.js
deleted file mode 100644
index b0ae2de6..00000000
--- a/common/src/leap/soledad/common/ddocs/syncs/updates/put.js
+++ /dev/null
@@ -1,151 +0,0 @@
-/**
- * The u1db_sync_log document stores both the actual sync log and a list of
- * pending updates to the log, in case we receive incoming documents out of
- * the correct order (i.e. if there are parallel PUTs during the sync
- * process).
- *
- * The structure of the document is the following:
- *
- * {
- * 'syncs': [
- * ['<replica_uid>', <gen>, '<trans_id>'],
- * ...
- * ],
- * 'pending': {
- * 'other_replica_uid': {
- * 'sync_id': '<sync_id>',
- * 'log': [[<gen>, '<trans_id>'], ...]
- * },
- * ...
- * }
- * }
- *
- * The update function below does the following:
- *
- * 0. If we do not receive a sync_id, we just update the 'syncs' list with
- * the incoming info about the source replica state.
- *
- * 1. Otherwise, if the incoming sync_id differs from current stored
- * sync_id, then we assume that the previous sync session for that source
- * replica was interrupted and discard all pending data.
- *
- * 2. Then we append incoming info as pending data for that source replica
- * and current sync_id, and sort the pending data by generation.
- *
- * 3. Then we go through pending data and find the most recent generation
- * that we can use to update the actual sync log.
- *
- * 4. Finally, we insert the most up to date information into the sync log.
- */
-function(doc, req){
-
- // create the document if it doesn't exist
- if (!doc) {
- doc = {}
- doc['_id'] = 'u1db_sync_log';
- doc['syncs'] = [];
- }
-
- // get and validate incoming info
- var body = JSON.parse(req.body);
- var other_replica_uid = body['other_replica_uid'];
- var other_generation = parseInt(body['other_generation']);
- var other_transaction_id = body['other_transaction_id']
- var sync_id = body['sync_id'];
- var number_of_docs = body['number_of_docs'];
- var doc_idx = body['doc_idx'];
-
- // parse integers
- if (number_of_docs != null)
- number_of_docs = parseInt(number_of_docs);
- if (doc_idx != null)
- doc_idx = parseInt(doc_idx);
-
- if (other_replica_uid == null
- || other_generation == null
- || other_transaction_id == null)
- return [null, 'invalid data'];
-
- // create slot for pending logs
- if (doc['pending'] == null)
- doc['pending'] = {};
-
- // these are the values that will be actually inserted
- var current_gen = other_generation;
- var current_trans_id = other_transaction_id;
-
- /*------------- Wait for sequential values before storing -------------*/
-
- // we just try to obtain pending log if we received a sync_id
- if (sync_id != null) {
-
- // create slot for current source and sync_id pending log
- if (doc['pending'][other_replica_uid] == null
- || doc['pending'][other_replica_uid]['sync_id'] != sync_id) {
- doc['pending'][other_replica_uid] = {
- 'sync_id': sync_id,
- 'log': [],
- 'last_doc_idx': 0,
- }
- }
-
- // append incoming data to pending log
- doc['pending'][other_replica_uid]['log'].push([
- other_generation,
- other_transaction_id,
- doc_idx,
- ])
-
- // sort pending log according to generation
- doc['pending'][other_replica_uid]['log'].sort(function(a, b) {
- return a[0] - b[0];
- });
-
- // get most up-to-date information from pending log
- var last_doc_idx = doc['pending'][other_replica_uid]['last_doc_idx'];
- var pending_idx = doc['pending'][other_replica_uid]['log'][0][2];
-
- current_gen = null;
- current_trans_id = null;
-
- while (last_doc_idx + 1 == pending_idx) {
- pending = doc['pending'][other_replica_uid]['log'].shift()
- current_gen = pending[0];
- current_trans_id = pending[1];
- last_doc_idx = pending[2]
- if (doc['pending'][other_replica_uid]['log'].length == 0)
- break;
- pending_idx = doc['pending'][other_replica_uid]['log'][0][2];
- }
-
- // leave the sync log untouched if we still did not receive enough docs
- if (current_gen == null)
- return [doc, 'ok'];
-
- // update last index of received doc
- doc['pending'][other_replica_uid]['last_doc_idx'] = last_doc_idx;
-
- // eventually remove all pending data from that replica
- if (last_doc_idx == number_of_docs)
- delete doc['pending'][other_replica_uid]
- }
-
- /*--------------- Store source replica info on sync log ---------------*/
-
- // remove outdated info
- doc['syncs'] = doc['syncs'].filter(
- function (entry) {
- return entry[0] != other_replica_uid;
- }
- );
-
- // store in log
- doc['syncs'].push([
- other_replica_uid,
- current_gen,
- current_trans_id
- ]);
-
- return [doc, 'ok'];
-}
-
diff --git a/common/src/leap/soledad/common/ddocs/syncs/views/log/map.js b/common/src/leap/soledad/common/ddocs/syncs/views/log/map.js
deleted file mode 100644
index a63c7cf4..00000000
--- a/common/src/leap/soledad/common/ddocs/syncs/views/log/map.js
+++ /dev/null
@@ -1,12 +0,0 @@
-function(doc) {
- if (doc._id == 'u1db_sync_log') {
- if (doc.syncs)
- doc.syncs.forEach(function (entry) {
- emit(entry[0],
- {
- 'known_generation': entry[1],
- 'known_transaction_id': entry[2]
- });
- });
- }
-}
diff --git a/common/src/leap/soledad/common/tests/test_couch.py b/common/src/leap/soledad/common/tests/test_couch.py
index a08ffd16..c8d13667 100644
--- a/common/src/leap/soledad/common/tests/test_couch.py
+++ b/common/src/leap/soledad/common/tests/test_couch.py
@@ -1359,10 +1359,6 @@ class CouchDatabaseExceptionsTests(CouchDBTestCase):
self.assertRaises(
errors.MissingDesignDocError,
self.db.whats_changed)
- # _do_set_replica_gen_and_trans_id()
- self.assertRaises(
- errors.MissingDesignDocError,
- self.db._do_set_replica_gen_and_trans_id, 1, 2, 3)
def test_missing_design_doc_functions_raises(self):
"""
@@ -1489,10 +1485,6 @@ class CouchDatabaseExceptionsTests(CouchDBTestCase):
self.assertRaises(
errors.MissingDesignDocDeletedError,
self.db.whats_changed)
- # _do_set_replica_gen_and_trans_id()
- self.assertRaises(
- errors.MissingDesignDocDeletedError,
- self.db._do_set_replica_gen_and_trans_id, 1, 2, 3)
def test_ensure_ddoc_independently(self):
"""
diff --git a/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py b/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
index 25f709ca..3e8e8cce 100644
--- a/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
+++ b/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
@@ -163,6 +163,7 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
2,
len(filter(lambda t: t[0] == doc_id, transaction_log)))
+ @defer.inlineCallbacks
def test_correct_sync_log_after_sequential_syncs(self):
"""
Assert that the sync_log increases accordingly with sequential syncs.
@@ -170,21 +171,21 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
sol = self._soledad_instance(
auth_token='auth-token',
server_url=self.getURL())
+ source_replica_uid = sol._dbpool.replica_uid
- def _create_docs(results):
+ def _create_docs():
deferreds = []
for i in xrange(0, REPEAT_TIMES):
deferreds.append(sol.create_doc({}))
- return defer.DeferredList(deferreds)
+ return defer.gatherResults(deferreds)
def _assert_transaction_and_sync_logs(results, sync_idx):
# assert sizes of transaction and sync logs
self.assertEqual(
sync_idx * REPEAT_TIMES,
len(self.db._get_transaction_log()))
- self.assertEqual(
- 1 if sync_idx > 0 else 0,
- len(self.db._database.view('syncs/log').rows))
+ gen, _ = self.db._get_replica_gen_and_trans_id(source_replica_uid)
+ self.assertEqual(sync_idx * REPEAT_TIMES, gen)
def _assert_sync(results, sync_idx):
gen, docs = results
@@ -193,40 +194,28 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
# assert sizes of transaction and sync logs
self.assertEqual((sync_idx + 1) * REPEAT_TIMES,
len(self.db._get_transaction_log()))
- sync_log_rows = self.db._database.view('syncs/log').rows
- sync_log = sync_log_rows[0].value
- replica_uid = sync_log_rows[0].key
- known_gen = sync_log['known_generation']
- known_trans_id = sync_log['known_transaction_id']
- # assert sync_log has exactly 1 row
- self.assertEqual(1, len(sync_log_rows))
- # assert it has the correct replica_uid, gen and trans_id
- self.assertEqual(sol._dbpool.replica_uid, replica_uid)
+ target_known_gen, target_known_trans_id = \
+ self.db._get_replica_gen_and_trans_id(source_replica_uid)
+ # assert it has the correct gen and trans_id
conn_key = sol._dbpool._u1dbconnections.keys().pop()
conn = sol._dbpool._u1dbconnections[conn_key]
sol_gen, sol_trans_id = conn._get_generation_info()
- self.assertEqual(sol_gen, known_gen)
- self.assertEqual(sol_trans_id, known_trans_id)
-
- # create some documents
- d = _create_docs(None)
+ self.assertEqual(sol_gen, target_known_gen)
+ self.assertEqual(sol_trans_id, target_known_trans_id)
# sync first time and assert success
- d.addCallback(_assert_transaction_and_sync_logs, 0)
- d.addCallback(lambda _: sol.sync())
- d.addCallback(lambda _: sol.get_all_docs())
- d.addCallback(_assert_sync, 0)
+ results = yield _create_docs()
+ _assert_transaction_and_sync_logs(results, 0)
+ yield sol.sync()
+ results = yield sol.get_all_docs()
+ _assert_sync(results, 0)
# create more docs, sync second time and assert success
- d.addCallback(_create_docs)
- d.addCallback(_assert_transaction_and_sync_logs, 1)
- d.addCallback(lambda _: sol.sync())
- d.addCallback(lambda _: sol.get_all_docs())
- d.addCallback(_assert_sync, 1)
-
- d.addCallback(lambda _: sol.close())
-
- return d
+ results = yield _create_docs()
+ _assert_transaction_and_sync_logs(results, 1)
+ yield sol.sync()
+ results = yield sol.get_all_docs()
+ _assert_sync(results, 1)
#
# Concurrency tests
diff --git a/server/pkg/requirements.pip b/server/pkg/requirements.pip
index d75678b2..3e1aa992 100644
--- a/server/pkg/requirements.pip
+++ b/server/pkg/requirements.pip
@@ -4,6 +4,7 @@ u1db
routes
PyOpenSSL
twisted
+Beaker
# XXX -- fix me!
# oauth is not strictly needed by us, but we need it until u1db adds it to its
diff --git a/server/src/leap/soledad/server/caching.py b/server/src/leap/soledad/server/caching.py
new file mode 100644
index 00000000..9a049a39
--- /dev/null
+++ b/server/src/leap/soledad/server/caching.py
@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+# caching.py
+# Copyright (C) 2015 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+"""
+Server side caching. Using beaker for now.
+"""
+from beaker.cache import CacheManager
+
+
+def setup_caching():
+ _cache_manager = CacheManager(type='memory')
+ return _cache_manager
+
+
+_cache_manager = setup_caching()
+
+
+def get_cache_for(key, expire=3600):
+ return _cache_manager.get_cache(key, expire=expire)
diff --git a/server/src/leap/soledad/server/state.py b/server/src/leap/soledad/server/state.py
new file mode 100644
index 00000000..f269b77e
--- /dev/null
+++ b/server/src/leap/soledad/server/state.py
@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+# state.py
+# Copyright (C) 2015 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+"""
+Server side synchronization infrastructure.
+"""
+from leap.soledad.server import caching
+
+
+class ServerSyncState(object):
+ """
+ The state of one sync session, as stored on backend server.
+
+ On server side, the ongoing syncs metadata is maintained in
+ a caching layer.
+ """
+
+ def __init__(self, source_replica_uid, sync_id):
+ """
+ Initialize the sync state object.
+
+ :param sync_id: The id of current sync
+ :type sync_id: str
+ :param source_replica_uid: The source replica uid
+ :type source_replica_uid: str
+ """
+ self._source_replica_uid = source_replica_uid
+ self._sync_id = sync_id
+ caching_key = source_replica_uid + sync_id
+ self._storage = caching.get_cache_for(caching_key)
+
+ def _put_dict_info(self, key, value):
+ """
+ Put some information about the sync state.
+
+ :param key: The key for the info to be put.
+ :type key: str
+ :param value: The value for the info to be put.
+ :type value: str
+ """
+ if key not in self._storage:
+ self._storage[key] = []
+ info_list = self._storage.get(key)
+ info_list.append(value)
+ self._storage[key] = info_list
+
+ def put_seen_id(self, seen_id, gen):
+ """
+ Put one seen id on the sync state.
+
+ :param seen_id: The doc_id of a document seen during sync.
+ :type seen_id: str
+ :param gen: The corresponding db generation.
+ :type gen: int
+ """
+ self._put_dict_info(
+ 'seen_id',
+ (seen_id, gen))
+
+ def seen_ids(self):
+ """
+ Return all document ids seen during the sync.
+
+ :return: A dict with doc ids seen during the sync.
+ :rtype: dict
+ """
+ if 'seen_id' in self._storage:
+ seen_ids = self._storage.get('seen_id')
+ else:
+ seen_ids = []
+ return dict(seen_ids)
+
+ def put_changes_to_return(self, gen, trans_id, changes_to_return):
+ """
+ Put the calculated changes to return in the backend sync state.
+
+ :param gen: The target database generation that will be synced.
+ :type gen: int
+ :param trans_id: The target database transaction id that will be
+ synced.
+ :type trans_id: str
+ :param changes_to_return: A list of tuples with the changes to be
+ returned during the sync process.
+ :type changes_to_return: list
+ """
+ self._put_dict_info(
+ 'changes_to_return',
+ {
+ 'gen': gen,
+ 'trans_id': trans_id,
+ 'changes_to_return': changes_to_return,
+ }
+ )
+
+ def sync_info(self):
+ """
+ Return information about the current sync state.
+
+ :return: The generation and transaction id of the target database
+ which will be synced, and the number of documents to return,
+ or a tuple of Nones if those have not already been sent to
+ server.
+ :rtype: tuple
+ """
+ gen = trans_id = number_of_changes = None
+ if 'changes_to_return' in self._storage:
+ info = self._storage.get('changes_to_return')[0]
+ gen = info['gen']
+ trans_id = info['trans_id']
+ number_of_changes = len(info['changes_to_return'])
+ return gen, trans_id, number_of_changes
+
+ def next_change_to_return(self, received):
+ """
+ Return the next change to be returned to the source syncing replica.
+
+ :param received: How many documents the source replica has already
+ received during the current sync process.
+ :type received: int
+ """
+ gen = trans_id = next_change_to_return = None
+ if 'changes_to_return' in self._storage:
+ info = self._storage.get('changes_to_return')[0]
+ gen = info['gen']
+ trans_id = info['trans_id']
+ if received < len(info['changes_to_return']):
+ next_change_to_return = (info['changes_to_return'][received])
+ return gen, trans_id, next_change_to_return
diff --git a/server/src/leap/soledad/server/sync.py b/server/src/leap/soledad/server/sync.py
index 18c4ee40..619be565 100644
--- a/server/src/leap/soledad/server/sync.py
+++ b/server/src/leap/soledad/server/sync.py
@@ -17,183 +17,16 @@
"""
Server side synchronization infrastructure.
"""
-import json
-
-from leap.soledad.common.couch import CouchDatabase
from u1db import sync, Document
from u1db.remote import http_app
+from leap.soledad.server.state import ServerSyncState
+from leap.soledad.server.caching import get_cache_for
MAX_REQUEST_SIZE = 200 # in Mb
MAX_ENTRY_SIZE = 200 # in Mb
-class ServerSyncState(object):
- """
- The state of one sync session, as stored on backend server.
-
- This object performes queries to distinct design documents:
-
- _design/syncs/_update/state
- _design/syncs/_view/state
- _design/syncs/_view/seen_ids
- _design/syncs/_view/changes_to_return
-
- On server side, the ongoing syncs metadata is maintained in a document
- called 'u1db_sync_state'.
- """
-
- def __init__(self, db, source_replica_uid, sync_id):
- """
- Initialize the sync state object.
-
- :param db: The target syncing database.
- :type db: CouchDatabase.
- :param source_replica_uid: CouchDatabase
- :type source_replica_uid: str
- """
- self._db = db
- self._source_replica_uid = source_replica_uid
- self._sync_id = sync_id
-
- def _key(self, key):
- """
- Format a key to be used on couch views.
-
- :param key: The lookup key.
- :type key: json serializable object
-
- :return: The properly formatted key.
- :rtype: str
- """
- return json.dumps(key, separators=(',', ':'))
-
- def _put_info(self, key, value):
- """
- Put some information on the sync state document.
-
- This method works in conjunction with the
- _design/syncs/_update/state update handler couch backend.
-
- :param key: The key for the info to be put.
- :type key: str
- :param value: The value for the info to be put.
- :type value: str
- """
- ddoc_path = [
- '_design', 'syncs', '_update', 'state',
- 'u1db_sync_state']
- res = self._db._database.resource(*ddoc_path)
- with CouchDatabase.sync_info_lock[self._db.replica_uid]:
- res.put_json(
- body={
- 'sync_id': self._sync_id,
- 'source_replica_uid': self._source_replica_uid,
- key: value,
- },
- headers={'content-type': 'application/json'})
-
- def put_seen_id(self, seen_id, gen):
- """
- Put one seen id on the sync state document.
-
- :param seen_id: The doc_id of a document seen during sync.
- :type seen_id: str
- :param gen: The corresponding db generation for that document.
- :type gen: int
- """
- self._put_info(
- 'seen_id',
- [seen_id, gen])
-
- def seen_ids(self):
- """
- Return all document ids seen during the sync.
-
- :return: A list with doc ids seen during the sync.
- :rtype: list
- """
- ddoc_path = ['_design', 'syncs', '_view', 'seen_ids']
- resource = self._db._database.resource(*ddoc_path)
- response = resource.get_json(
- key=self._key([self._source_replica_uid, self._sync_id]))
- data = response[2]
- if data['rows']:
- entry = data['rows'].pop()
- return entry['value']['seen_ids']
- return []
-
- def put_changes_to_return(self, gen, trans_id, changes_to_return):
- """
- Put the calculated changes to return in the backend sync state
- document.
-
- :param gen: The target database generation that will be synced.
- :type gen: int
- :param trans_id: The target database transaction id that will be
- synced.
- :type trans_id: str
- :param changes_to_return: A list of tuples with the changes to be
- returned during the sync process.
- :type changes_to_return: list
- """
- self._put_info(
- 'changes_to_return',
- {
- 'gen': gen,
- 'trans_id': trans_id,
- 'changes_to_return': changes_to_return,
- }
- )
-
- def sync_info(self):
- """
- Return information about the current sync state.
-
- :return: The generation and transaction id of the target database
- which will be synced, and the number of documents to return,
- or a tuple of Nones if those have not already been sent to
- server.
- :rtype: tuple
- """
- ddoc_path = ['_design', 'syncs', '_view', 'state']
- resource = self._db._database.resource(*ddoc_path)
- response = resource.get_json(
- key=self._key([self._source_replica_uid, self._sync_id]))
- data = response[2]
- gen = None
- trans_id = None
- number_of_changes = None
- if data['rows'] and data['rows'][0]['value'] is not None:
- value = data['rows'][0]['value']
- gen = value['gen']
- trans_id = value['trans_id']
- number_of_changes = value['number_of_changes']
- return gen, trans_id, number_of_changes
-
- def next_change_to_return(self, received):
- """
- Return the next change to be returned to the source syncing replica.
-
- :param received: How many documents the source replica has already
- received during the current sync process.
- :type received: int
- """
- ddoc_path = ['_design', 'syncs', '_view', 'changes_to_return']
- resource = self._db._database.resource(*ddoc_path)
- response = resource.get_json(
- key=self._key(
- [self._source_replica_uid, self._sync_id, received]))
- data = response[2]
- if not data['rows']:
- return None, None, None
- value = data['rows'][0]['value']
- gen = value['gen']
- trans_id = value['trans_id']
- next_change_to_return = value['next_change_to_return']
- return gen, trans_id, tuple(next_change_to_return)
-
-
class SyncExchange(sync.SyncExchange):
def __init__(self, db, source_replica_uid, last_known_generation, sync_id):
@@ -216,8 +49,7 @@ class SyncExchange(sync.SyncExchange):
self.new_trans_id = None
self._trace_hook = None
# recover sync state
- self._sync_state = ServerSyncState(
- self._db, self.source_replica_uid, sync_id)
+ self._sync_state = ServerSyncState(self.source_replica_uid, sync_id)
def find_changes_to_return(self, received):
"""
@@ -353,10 +185,15 @@ class SyncResource(http_app.SyncResource):
:type ensure: bool
"""
# create or open the database
+ cache = get_cache_for('db-' + sync_id + self.dbname)
if ensure:
db, self.replica_uid = self.state.ensure_database(self.dbname)
+ elif cache and 'instance' in cache:
+ db = cache['instance']
else:
db = self.state.open_database(self.dbname)
+ db.init_caching(cache)
+ cache['instance'] = db
# validate the information the client has about server replica
db.validate_gen_and_trans_id(
last_known_generation, last_known_trans_id)
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-12 21:37:41 +0000