Merge lp:~pedronis/u1db/sync-autocreate into lp:u1db

[Samuele Pedroni]

support sync(autocreate=True) by supporting sending an ensure=true flag over the sync_exchange POST

support sync(autocreate=True) by supporting sending an ensure=true flag over the sync_exchange POST, mediated by passing a ensure_callback to sync_exchange() that will gather the newly created db replica uid

[Lucio Torre (lucio.torre)]

review: Approve

=== modified file 'include/u1db/u1db_internal.h'
--- include/u1db/u1db_internal.h	2012-08-15 12:39:01 +0000
+++ include/u1db/u1db_internal.h	2012-09-25 20:06:22 +0000
@@ -48,6 +48,8 @@
 
 typedef int (*u1db__trace_callback)(void *context, const char *state);
 
+typedef void (*u1db__ensure_callback)(void *context, const char *replica_uid);
+
 struct _u1db_sync_target {
     void *trace_context;
     u1db__trace_callback trace_cb;
@@ -119,11 +121,16 @@
      * @param cb        After sending the requested documents, we read the
      *                  response stream. For each document in the stream, we
      *                  will trigger a callback.
+     * @param: ensure_callback  If set the target may create
+     *                          the target db if not yet existent,
+     *                          the callback can then be used to inform of
+     *                          the created db replica uid.
      */
     int (*sync_exchange_doc_ids)(u1db_sync_target *st, u1database *source_db,
             int n_doc_ids, const char **doc_ids, int *generations,
             const char **trans_ids, int *target_gen, char **target_trans_id,
-            void *context, u1db_doc_gen_callback cb);
+            void *context, u1db_doc_gen_callback cb,
+            u1db__ensure_callback ensure_callback);
 
     /**
      * Same as sync_exchange, only using document objects.
@@ -133,7 +140,8 @@
                          u1db_document **docs, int *generations,
                          const char **trans_ids, int *target_gen,
                          char **target_trans_id, void *context,
-                         u1db_doc_gen_callback cb);
+                         u1db_doc_gen_callback cb,
+                         u1db__ensure_callback ensure_callback);
     /**
      * Create a sync_exchange state object.
      *

=== modified file 'src/u1db_http_sync_target.c'
--- src/u1db_http_sync_target.c	2012-07-26 16:24:39 +0000
+++ src/u1db_http_sync_target.c	2012-09-25 20:06:22 +0000
@@ -62,7 +62,8 @@
                                  u1db_document **docs, int *generations,
                                  const char **trans_ids, int *target_gen,
                                  char **target_trans_id, void *context,
-                                 u1db_doc_gen_callback cb);
+                                 u1db_doc_gen_callback cb,
+                                 u1db__ensure_callback ensure_callback);
 static int st_http_sync_exchange_doc_ids(u1db_sync_target *st,
                                          u1database *source_db, int n_doc_ids,
                                          const char **doc_ids,
@@ -70,7 +71,8 @@
                                          const char **trans_ids,
                                          int *target_gen,
                                          char **target_trans_id, void *context,
-                                         u1db_doc_gen_callback cb);
+                                         u1db_doc_gen_callback cb,
+                                         u1db__ensure_callback ensure_callback);
 static void st_http_finalize_sync_exchange(u1db_sync_target *st,
                                u1db_sync_exchange **exchange);
 static int st_http_set_trace_hook(u1db_sync_target *st,
@@ -794,7 +796,7 @@
 
 static int
 init_temp_file(char tmpname[], FILE **temp_fd, int target_gen,
-               char *target_trans_id)
+               char *target_trans_id, int ensure)
 {
     int status = U1DB_OK;
     *temp_fd = make_tempfile(tmpname);
@@ -809,8 +811,16 @@
     // determine Content-Length before we start uploading the data.
     fprintf(
         *temp_fd,
-        "[\r\n{\"last_known_generation\": %d, \"last_known_trans_id\": \"%s\"}",
+        "[\r\n{\"last_known_generation\": %d, \"last_known_trans_id\": \"%s\"",
         target_gen, target_trans_id);
+    if (ensure) {
+        fprintf(
+            *temp_fd,
+            ", \"ensure\": true");
+    }
+    fprintf(
+        *temp_fd,
+        "}");
 finish:
     return status;
 }
@@ -866,7 +876,7 @@
     }
     if (status != U1DB_OK) { goto finish; }
     if (http_code != 200 && http_code != 201) {
-        printf("broken 0\n");
+        //printf("broken 0\n");
         status = U1DB_BROKEN_SYNC_STREAM;
         goto finish;
     }
@@ -883,7 +893,8 @@
 
 static int
 process_response(u1db_sync_target *st, void *context, u1db_doc_gen_callback cb,
-                 char *response, int *target_gen, char **target_trans_id)
+                 u1db__ensure_callback ensure_callback, char *response,
+                 int *target_gen, char **target_trans_id)
 {
     int status = U1DB_OK;
     int i, doc_count;
@@ -893,37 +904,38 @@
     int gen;
     const char *trans_id = NULL;
     u1db_document *doc;
+    const char *replica_uid = NULL;
 
     json = json_tokener_parse(response);
     if (json == NULL || !json_object_is_type(json, json_type_array)) {
-        printf("broken 1, response: %s\n", response);
+        //printf("broken 1, response: %s\n", response);
         status = U1DB_BROKEN_SYNC_STREAM;
         goto finish;
     }
     doc_count = json_object_array_length(json);
     if (doc_count < 1) {
         // the first response is the new_generation info, so it must exist
-        printf("broken 2\n");
+        //printf("broken 2\n");
         status = U1DB_BROKEN_SYNC_STREAM;
         goto finish;
     }
     obj = json_object_array_get_idx(json, 0);
     attr = json_object_object_get(obj, "new_generation");
     if (attr == NULL) {
-        printf("broken 3\n");
+        //printf("broken 3\n");
         status = U1DB_BROKEN_SYNC_STREAM;
         goto finish;
     }
     *target_gen = json_object_get_int(attr);
     attr = json_object_object_get(obj, "new_transaction_id");
     if (attr == NULL) {
-        printf("broken 4\n");
+        //printf("broken 4\n");
         status = U1DB_BROKEN_SYNC_STREAM;
         goto finish;
     }
     tmp = json_object_get_string(attr);
     if (tmp == NULL) {
-        printf("broken 5\n");
+        //printf("broken 5\n");
         status = U1DB_BROKEN_SYNC_STREAM;
         goto finish;
     }
@@ -933,6 +945,23 @@
         goto finish;
     }
 
+    if (ensure_callback) {
+        attr = json_object_object_get(obj, "replica_uid");
+        if (attr != NULL) {
+            tmp = json_object_get_string(attr);
+            if (tmp == NULL) {
+                status = U1DB_BROKEN_SYNC_STREAM;
+                goto finish;
+            }
+            replica_uid = strdup(tmp);
+            if (*target_trans_id == NULL) {
+                status = U1DB_NOMEM;
+                goto finish;
+            }
+            ensure_callback(context, replica_uid);
+        }
+    }
+
     for (i = 1; i < doc_count; ++i) {
         obj = json_object_array_get_idx(json, i);
         attr = json_object_object_get(obj, "id");
@@ -989,7 +1018,8 @@
                       int n_docs, u1db_document **docs, int *generations,
                       const char **trans_ids, int *target_gen,
                       char **target_trans_id, void *context,
-                      u1db_doc_gen_callback cb)
+                      u1db_doc_gen_callback cb,
+                      u1db__ensure_callback ensure_callback)
 {
     int status, i;
     FILE *temp_fd = NULL;
@@ -1004,7 +1034,8 @@
     if (n_docs > 0 && (docs == NULL || generations == NULL)) {
         return U1DB_INVALID_PARAMETER;
     }
-    status = init_temp_file(tmpname, &temp_fd, *target_gen, *target_trans_id);
+    status = init_temp_file(tmpname, &temp_fd, *target_gen, *target_trans_id,
+                            ensure_callback != NULL);
     if (status != U1DB_OK) { goto finish; }
     for (i = 0; i < n_docs; ++i) {
         status = doc_to_tempfile(
@@ -1013,7 +1044,8 @@
     }
     status = finalize_and_send_temp_file(st, temp_fd, source_replica_uid, &req);
     if (status != U1DB_OK) { goto finish; }
-    status = process_response(st, context, cb, req.body_buffer, target_gen,
+    status = process_response(st, context, cb, ensure_callback,
+                              req.body_buffer, target_gen,
                               target_trans_id);
 finish:
     cleanup_temp_files(tmpname, temp_fd, &req);
@@ -1058,7 +1090,8 @@
                               int n_doc_ids, const char **doc_ids,
                               int *generations, const char **trans_ids,
                               int *target_gen, char **target_trans_id,
-                              void *context, u1db_doc_gen_callback cb)
+                              void *context, u1db_doc_gen_callback cb,
+                              u1db__ensure_callback ensure_callback)
 {
     int status;
     FILE *temp_fd = NULL;
@@ -1077,7 +1110,8 @@
     }
     status = u1db_get_replica_uid(source_db, &source_replica_uid);
     if (status != U1DB_OK) { goto finish; }
-    status = init_temp_file(tmpname, &temp_fd, *target_gen, *target_trans_id);
+    status = init_temp_file(tmpname, &temp_fd, *target_gen, *target_trans_id,
+                            ensure_callback != NULL);
     if (status != U1DB_OK) { goto finish; }
     state.num = n_doc_ids;
     state.generations = generations;
@@ -1088,7 +1122,8 @@
     if (status != U1DB_OK) { goto finish; }
     status = finalize_and_send_temp_file(st, temp_fd, source_replica_uid, &req);
     if (status != U1DB_OK) { goto finish; }
-    status = process_response(st, context, cb, req.body_buffer, target_gen,
+    status = process_response(st, context, cb, ensure_callback,
+                              req.body_buffer, target_gen,
                               target_trans_id);
 finish:
     cleanup_temp_files(tmpname, temp_fd, &req);

=== modified file 'src/u1db_sync_target.c'
--- src/u1db_sync_target.c	2012-08-15 12:57:57 +0000
+++ src/u1db_sync_target.c	2012-09-25 20:06:22 +0000
@@ -35,13 +35,15 @@
                             u1db_document **docs, int *generations,
                             const char **trans_ids, int *target_gen,
                             char **target_trans_id, void *context,
-                            u1db_doc_gen_callback cb);
+                            u1db_doc_gen_callback cb,
+                            u1db__ensure_callback ensure_callback);
 static int st_sync_exchange_doc_ids(u1db_sync_target *st,
                                     u1database *source_db, int n_doc_ids,
                                     const char **doc_ids, int *generations,
                                     const char **trans_ids, int *target_gen,
                                     char **target_trans_id, void *context,
-                                    u1db_doc_gen_callback cb);
+                                    u1db_doc_gen_callback cb,
+                                    u1db__ensure_callback ensure_callback);
 static int st_get_sync_exchange(u1db_sync_target *st,
                          const char *source_replica_uid,
                          int source_gen,
@@ -477,6 +479,12 @@
     return status;
 }
 
+static void ensure_callback(void *context, const char *replica_uid) {
+   struct _return_doc_state *state;
+   state = (struct _return_doc_state *)context;
+   state->target_uid = replica_uid;
+}
+
 
 static int
 get_and_insert_docs(u1database *source_db, u1db_sync_exchange *se,
@@ -503,7 +511,8 @@
                  int n_docs, u1db_document **docs, int *generations,
                  const char **trans_ids, int *target_gen,
                  char **target_trans_id, void *context,
-                 u1db_doc_gen_callback cb)
+                 u1db_doc_gen_callback cb,
+                 u1db__ensure_callback ensure_callback)
 {
     int status, i;
 
@@ -544,7 +553,8 @@
 st_sync_exchange_doc_ids(u1db_sync_target *st, u1database *source_db,
         int n_doc_ids, const char **doc_ids, int *generations,
         const char **trans_ids, int *target_gen, char **target_trans_id,
-        void *context, u1db_doc_gen_callback cb)
+        void *context, u1db_doc_gen_callback cb,
+        u1db__ensure_callback ensure_callback)
 {
     int status;
     const char *source_replica_uid = NULL;
@@ -598,6 +608,7 @@
     char *target_trans_id = NULL;
     int target_gen, local_gen;
     int local_gen_known_by_target, target_gen_known_by_local;
+    u1db__ensure_callback ensure= NULL;
 
     // fprintf(stderr, "Starting\n");
     if (db == NULL || target == NULL || local_gen_before_sync == NULL) {
@@ -612,14 +623,38 @@
     status = target->get_sync_info(
         target, local_uid, &target_uid, &target_gen, &target_trans_id,
         &local_gen_known_by_target, &local_trans_id_known_by_target);
-    if (status != U1DB_OK) { goto finish; }
+    if (status == 404) { // database does not exist
+        target_uid = NULL;
+        target_gen = 0;
+        local_gen_known_by_target = 0;
+        target_trans_id = strdup("");
+        if (target_trans_id == NULL) {
+            status = U1DB_NOMEM;
+            goto finish;
+        }
+        local_trans_id_known_by_target = strdup("");
+        if (local_trans_id_known_by_target == NULL) {
+            status = U1DB_NOMEM;
+            goto finish;
+        }
+        ensure = ensure_callback;
+    } else if (status != U1DB_OK) { goto finish; }
     status = u1db_validate_gen_and_trans_id(
         db, local_gen_known_by_target, local_trans_id_known_by_target);
     if (status != U1DB_OK) { goto finish; }
-    status = u1db__get_replica_gen_and_trans_id(
-        db, target_uid, &target_gen_known_by_local,
-        &target_trans_id_known_by_local);
-    if (status != U1DB_OK) { goto finish; }
+    if (target_uid == NULL) {
+        target_gen_known_by_local = 0;
+        target_trans_id_known_by_local = strdup("");
+        if (target_trans_id_known_by_local == NULL) {
+            status = U1DB_NOMEM;
+            goto finish;
+        }
+    } else {
+        status = u1db__get_replica_gen_and_trans_id(
+            db, target_uid, &target_gen_known_by_local,
+            &target_trans_id_known_by_local);
+        if (status != U1DB_OK) { goto finish; }
+    }
     local_target_trans_id = target_trans_id_known_by_local;
     local_gen = local_gen_known_by_target;
 
@@ -649,7 +684,14 @@
         (const char**)to_send_state.doc_ids_to_return,
         to_send_state.gen_for_doc_ids, to_send_state.trans_ids_for_doc_ids,
         &target_gen_known_by_local, &target_trans_id_known_by_local,
-        &return_doc_state, return_doc_to_insert_from_target);
+        &return_doc_state, return_doc_to_insert_from_target, ensure);
+    if (ensure != NULL) {
+        if (return_doc_state.target_uid != NULL) {
+            target_uid = return_doc_state.target_uid;
+        } else {
+            ensure = NULL;
+        }
+    }
     if (status != U1DB_OK) { goto finish; }
     if (local_trans_id != NULL) {
         free(local_trans_id);
@@ -671,6 +713,9 @@
         if (status != U1DB_OK) { goto finish; }
     }
 finish:
+    if (ensure != NULL && target_uid != NULL) {
+        free((void *)target_uid);
+    }
     if (local_trans_id != NULL) {
         free(local_trans_id);
     }

=== modified file 'u1db/__init__.py'
--- u1db/__init__.py	2012-09-24 14:12:27 +0000
+++ u1db/__init__.py	2012-09-25 20:06:22 +0000
@@ -303,7 +303,7 @@
         """Release any resources associated with this database."""
         raise NotImplementedError(self.close)
 
-    def sync(self, url, creds=None):
+    def sync(self, url, creds=None, autocreate=True):
         """Synchronize documents with remote replica exposed at url.
 
         :param url: the url of the target replica to sync with.
@@ -316,6 +316,7 @@
                  'token_key': ...,
                  'token_secret': ...
                 }}
+        :param autocreate: ask the target to create the db if non-existent.
         :return: local_gen_before_sync The local generation before the
             synchronisation was performed. This is useful to pass into
             whatschanged, if an application wants to know which documents were
@@ -323,7 +324,8 @@
         """
         from u1db.sync import Synchronizer
         from u1db.remote.http_target import HTTPSyncTarget
-        return Synchronizer(self, HTTPSyncTarget(url, creds=creds)).sync()
+        return Synchronizer(self, HTTPSyncTarget(url, creds=creds)).sync(
+            autocreate=autocreate)
 
     def _get_replica_gen_and_trans_id(self, other_replica_uid):
         """Return the last known generation and transaction id for the other db
@@ -633,7 +635,7 @@
 
     def sync_exchange(self, docs_by_generation, source_replica_uid,
                       last_known_generation, last_known_trans_id,
-                      return_doc_cb):
+                      return_doc_cb, ensure_callback=None):
         """Incorporate the documents sent from the source replica.
 
         This is not meant to be called by client code directly, but is used as
@@ -665,6 +667,9 @@
             be invoked in turn with Documents that have changed since
             last_known_generation together with the generation of
             their last change.
+        :param: ensure_callback(replica_uid): if set the target may create
+            the target db if not yet existent, the callback can then
+            be used to inform of the created db replica uid.
         :return: new_generation - After applying docs_by_generation, this is
             the current generation for this replica
         """

=== modified file 'u1db/remote/http_app.py'
--- u1db/remote/http_app.py	2012-09-04 16:24:31 +0000
+++ u1db/remote/http_app.py	2012-09-25 20:06:22 +0000
@@ -344,12 +344,16 @@
     def __init__(self, dbname, source_replica_uid, state, responder):
         self.source_replica_uid = source_replica_uid
         self.responder = responder
-        self.db = state.open_database(dbname)
-        self.target = self.db.get_sync_target()
+        self.state = state
+        self.dbname = dbname
+        self.replica_uid = None
+
+    def get_target(self):
+        return self.state.open_database(self.dbname).get_sync_target()
 
     @http_method()
     def get(self):
-        result = self.target.get_sync_info(self.source_replica_uid)
+        result = self.get_target().get_sync_info(self.source_replica_uid)
         self.responder.send_response_json(
             target_replica_uid=result[0], target_replica_generation=result[1],
             target_replica_transaction_id=result[2],
@@ -360,19 +364,25 @@
     @http_method(generation=int,
                  content_as_args=True, no_query=True)
     def put(self, generation, transaction_id):
-        self.target.record_sync_info(self.source_replica_uid, generation,
-                                     transaction_id)
+        self.get_target().record_sync_info(self.source_replica_uid,
+                                           generation,
+                                           transaction_id)
         self.responder.send_response_json(ok=True)
 
     # Implements the same logic as LocalSyncTarget.sync_exchange
 
     @http_method(last_known_generation=int, last_known_trans_id=none_or_str,
                  content_as_args=True)
-    def post_args(self, last_known_generation, last_known_trans_id=None):
-        self.db.validate_gen_and_trans_id(
+    def post_args(self, last_known_generation, last_known_trans_id=None,
+                  ensure=False):
+        if ensure:
+            db, self.replica_uid = self.state.ensure_database(self.dbname)
+        else:
+            db = self.state.open_database(self.dbname)
+        db.validate_gen_and_trans_id(
             last_known_generation, last_known_trans_id)
         self.sync_exch = self.sync_exchange_class(
-            self.db, self.source_replica_uid, last_known_generation)
+            db, self.source_replica_uid, last_known_generation)
 
     @http_method(content_as_args=True)
     def post_stream_entry(self, id, rev, content, gen, trans_id):
@@ -390,8 +400,11 @@
         self.responder.content_type = 'application/x-u1db-sync-stream'
         self.responder.start_response(200)
         self.responder.start_stream(),
-        self.responder.stream_entry({"new_generation": new_gen,
-                     "new_transaction_id": self.sync_exch.new_trans_id})
+        header = {"new_generation": new_gen,
+                     "new_transaction_id": self.sync_exch.new_trans_id}
+        if self.replica_uid is not None:
+            header['replica_uid'] = self.replica_uid
+        self.responder.stream_entry(header)
         self.sync_exch.return_docs(send_doc)
         self.responder.end_stream()
         self.responder.finish_response()

=== modified file 'u1db/remote/http_target.py'
--- u1db/remote/http_target.py	2012-09-20 18:46:10 +0000
+++ u1db/remote/http_target.py	2012-09-25 20:06:22 +0000
@@ -57,7 +57,7 @@
                               {'generation': source_replica_generation,
                                'transaction_id': source_transaction_id})
 
-    def _parse_sync_stream(self, data, return_doc_cb):
+    def _parse_sync_stream(self, data, return_doc_cb, ensure_callback=None):
         parts = data.splitlines()  # one at a time
         if not parts or parts[0] != '[':
             raise BrokenSyncStream
@@ -66,6 +66,8 @@
         if data:
             line, comma = utils.check_and_strip_comma(data[0])
             res = json.loads(line)
+            if ensure_callback and 'replica_uid' in res:
+                ensure_callback(res['replica_uid'])
             for entry in data[1:]:
                 if not comma:  # missing in between comma
                     raise BrokenSyncStream
@@ -88,7 +90,7 @@
 
     def sync_exchange(self, docs_by_generations, source_replica_uid,
                       last_known_generation, last_known_trans_id,
-                      return_doc_cb):
+                      return_doc_cb, ensure_callback=None):
         self._ensure_connection()
         if self._trace_hook:  # for tests
             self._trace_hook('sync_exchange')
@@ -108,7 +110,8 @@
         comma = ''
         size += prepare(
             last_known_generation=last_known_generation,
-            last_known_trans_id=last_known_trans_id)
+            last_known_trans_id=last_known_trans_id,
+            ensure=ensure_callback is not None)
         comma = ','
         for doc, gen, trans_id in docs_by_generations:
             size += prepare(id=doc.doc_id, rev=doc.rev, content=doc.get_json(),
@@ -121,7 +124,7 @@
             self._conn.send(entry)
         entries = None
         data, _ = self._response()
-        res = self._parse_sync_stream(data, return_doc_cb)
+        res = self._parse_sync_stream(data, return_doc_cb, ensure_callback)
         data = None
         return res['new_generation'], res['new_transaction_id']
 

=== modified file 'u1db/remote/server_state.py'
--- u1db/remote/server_state.py	2012-05-15 11:11:06 +0000
+++ u1db/remote/server_state.py	2012-09-25 20:06:22 +0000
@@ -56,8 +56,9 @@
         """Ensure database at the given location."""
         from u1db.backends import sqlite_backend
         full_path = self._relpath(path)
-        return sqlite_backend.SQLiteDatabase.open_database(full_path,
-                                                           create=True)
+        db = sqlite_backend.SQLiteDatabase.open_database(full_path,
+                                                         create=True)
+        return db, db._replica_uid
 
     def delete_database(self, path):
         """Delete database at the given location."""

=== modified file 'u1db/sync.py'
--- u1db/sync.py	2012-07-26 19:14:26 +0000
+++ u1db/sync.py	2012-09-25 20:06:22 +0000
@@ -90,14 +90,26 @@
             self.sync_target.record_sync_info(
                 self.source._replica_uid, cur_gen, trans_id)
 
-    def sync(self, callback=None):
+    def sync(self, callback=None, autocreate=False):
         """Synchronize documents between source and target."""
         sync_target = self.sync_target
         # get target identifier, its current generation,
         # and its last-seen database generation for this source
-        (self.target_replica_uid, target_gen, target_trans_id, target_my_gen,
-         target_my_trans_id) = sync_target.get_sync_info(
-             self.source._replica_uid)
+        try:
+            (self.target_replica_uid, target_gen, target_trans_id,
+             target_my_gen, target_my_trans_id) = sync_target.get_sync_info(
+                self.source._replica_uid)
+        except errors.DatabaseDoesNotExist:
+            if not autocreate:
+                raise
+            # will try to ask sync_exchange() to create the db
+            self.target_replica_uid = None
+            target_gen, target_trans_id = 0, ''
+            target_my_gen, target_my_trans_id = 0, ''
+            def ensure_callback(replica_uid):
+                self.target_replica_uid = replica_uid
+        else:
+            ensure_callback = None
         # validate the generation and transaction id the target knows about us
         self.source.validate_gen_and_trans_id(
             target_my_gen, target_my_trans_id)
@@ -105,7 +117,10 @@
         my_gen, _, changes = self.source.whats_changed(target_my_gen)
 
         # this source last-seen database generation for the target
-        target_last_known_gen, target_last_known_trans_id = \
+        if self.target_replica_uid is None:
+            target_last_known_gen, target_last_known_trans_id = 0, ''
+        else:
+            target_last_known_gen, target_last_known_trans_id = \
             self.source._get_replica_gen_and_trans_id(self.target_replica_uid)
         if not changes and target_last_known_gen == target_gen:
             if target_trans_id != target_last_known_trans_id:
@@ -125,7 +140,7 @@
         new_gen, new_trans_id = sync_target.sync_exchange(
             docs_by_generation, self.source._replica_uid,
             target_last_known_gen, target_last_known_trans_id,
-            self._insert_doc_from_target)
+            self._insert_doc_from_target, ensure_callback=ensure_callback)
         # record target synced-up-to generation including applying what we sent
         self.source._set_replica_gen_and_trans_id(
             self.target_replica_uid, new_gen, new_trans_id)
@@ -269,7 +284,7 @@
 
     def sync_exchange(self, docs_by_generations, source_replica_uid,
                       last_known_generation, last_known_trans_id,
-                      return_doc_cb):
+                      return_doc_cb, ensure_callback=None):
         self._db.validate_gen_and_trans_id(
             last_known_generation, last_known_trans_id)
         sync_exch = SyncExchange(

=== modified file 'u1db/tests/__init__.py'
--- u1db/tests/__init__.py	2012-09-24 09:36:54 +0000
+++ u1db/tests/__init__.py	2012-09-25 20:06:22 +0000
@@ -293,9 +293,10 @@
 
     def ensure_database(self, path):
         try:
-            return self.open_database(path)
+            db =  self.open_database(path)
         except errors.DatabaseDoesNotExist:
-            return self._create_database(path)
+            db = self._create_database(path)
+        return db, db._replica_uid
 
     def _copy_database(self, db):
         # DO NOT COPY OR REUSE THIS CODE OUTSIDE TESTS: COPYING U1DB DATABASES

=== modified file 'u1db/tests/c_backend_wrapper.pyx'
--- u1db/tests/c_backend_wrapper.pyx	2012-09-24 14:12:27 +0000
+++ u1db/tests/c_backend_wrapper.pyx	2012-09-25 20:06:22 +0000
@@ -205,14 +205,16 @@
                              u1db_document **docs, int *generations,
                              const_char_ptr *trans_ids,
                              int *target_gen, char **target_trans_id,
-                             void *context, u1db_doc_gen_callback cb) nogil
+                             void *context, u1db_doc_gen_callback cb,
+                             void *ensure_callback) nogil
         int (*sync_exchange_doc_ids)(u1db_sync_target *st,
                                      u1database *source_db, int n_doc_ids,
                                      const_char_ptr *doc_ids, int *generations,
                                      const_char_ptr *trans_ids,
                                      int *target_gen, char **target_trans_id,
                                      void *context,
-                                     u1db_doc_gen_callback cb) nogil
+                                     u1db_doc_gen_callback cb,
+                                     void *ensure_callback) nogil
         int (*get_sync_exchange)(u1db_sync_target *st,
                                  char *source_replica_uid,
                                  int last_known_source_gen,
@@ -811,7 +813,7 @@
                 status = self._st.sync_exchange_doc_ids(self._st, sdb._db,
                     num_doc_ids, doc_ids, generations, trans_ids,
                     &target_gen, &target_trans_id,
-                    <void*>return_doc_cb, return_doc_cb_wrapper)
+                    <void*>return_doc_cb, return_doc_cb_wrapper, NULL)
             handle_status("sync_exchange_doc_ids", status)
             if target_trans_id != NULL:
                 res_trans_id = target_trans_id
@@ -828,7 +830,7 @@
 
     def sync_exchange(self, docs_by_generations, source_replica_uid,
                       last_known_generation, last_known_trans_id,
-                      return_doc_cb):
+                      return_doc_cb, ensure_callback=None):
         cdef CDocument cur_doc
         cdef u1db_document **docs = NULL
         cdef int *generations = NULL
@@ -864,7 +866,7 @@
                 status = self._st.sync_exchange(
                     self._st, c_source_replica_uid, count, docs, generations,
                     trans_ids, &target_gen, &target_trans_id,
-                    <void *>return_doc_cb, return_doc_cb_wrapper)
+                    <void *>return_doc_cb, return_doc_cb_wrapper, NULL)
             handle_status("sync_exchange", status)
         finally:
             if docs != NULL:

=== modified file 'u1db/tests/test_http_app.py'
--- u1db/tests/test_http_app.py	2012-09-04 16:28:21 +0000
+++ u1db/tests/test_http_app.py	2012-09-25 20:06:22 +0000
@@ -839,6 +839,38 @@
         self.assertEqual('', bits[3])
         self.assertEqual([('replica', 10), ('replica', 11)], gens)
 
+    def test_sync_exchange_send_ensure(self):
+        entries = {
+            10: {'id': 'doc-here', 'rev': 'replica:1', 'content':
+                 '{"value": "here"}', 'gen': 10, 'trans_id': 'T-sid'},
+            11: {'id': 'doc-here2', 'rev': 'replica:1', 'content':
+                 '{"value": "here2"}', 'gen': 11, 'trans_id': 'T-sed'}
+            }
+
+        args = dict(last_known_generation=0, ensure=True)
+        body = ("[\r\n" +
+                "%s,\r\n" % json.dumps(args) +
+                "%s,\r\n" % json.dumps(entries[10]) +
+                "%s\r\n" % json.dumps(entries[11]) +
+                "]\r\n")
+        resp = self.app.post('/dbnew/sync-from/replica',
+                            params=body,
+                            headers={'content-type':
+                                     'application/x-u1db-sync-stream'})
+        self.assertEqual(200, resp.status)
+        self.assertEqual('application/x-u1db-sync-stream',
+                         resp.header('content-type'))
+        bits = resp.body.split('\r\n')
+        self.assertEqual('[', bits[0])
+        dbnew = self.state.open_database("dbnew")
+        last_trans_id = dbnew._get_transaction_log()[-1][1]
+        self.assertEqual({'new_generation': 2,
+                          'new_transaction_id': last_trans_id,
+                          'replica_uid': dbnew._replica_uid},
+                         json.loads(bits[1]))
+        self.assertEqual(']', bits[2])
+        self.assertEqual('', bits[3])
+
     def test_sync_exchange_send_entry_too_large(self):
         self.patch(http_app.SyncResource, 'max_request_size', 20000)
         self.patch(http_app.SyncResource, 'max_entry_size', 10000)

=== modified file 'u1db/tests/test_remote_sync_target.py'
--- u1db/tests/test_remote_sync_target.py	2012-09-24 09:36:54 +0000
+++ u1db/tests/test_remote_sync_target.py	2012-09-25 20:06:22 +0000
@@ -286,5 +286,29 @@
             (doc.doc_id, doc.rev, '{"value": "there"}', 1),
             other_changes[0][:-1])
 
+    def test_sync_exchange_send_ensure_callback(self):
+        self.startServer()
+        remote_target = self.getSyncTarget('test')
+        other_docs = []
+        replica_uid_box = []
+
+        def receive_doc(doc):
+            other_docs.append((doc.doc_id, doc.rev, doc.get_json()))
+
+        def ensure_cb(replica_uid):
+            replica_uid_box.append(replica_uid)
+
+        doc = self.make_document('doc-here', 'replica:1', '{"value": "here"}')
+        new_gen, trans_id = remote_target.sync_exchange(
+            [(doc, 10, 'T-sid')], 'replica', last_known_generation=0,
+            last_known_trans_id=None, return_doc_cb=receive_doc,
+            ensure_callback=ensure_cb)
+        self.assertEqual(1, new_gen)
+        db = self.request_state.open_database('test')
+        self.assertEqual(1, len(replica_uid_box))
+        self.assertEqual(db._replica_uid, replica_uid_box[0])
+        self.assertGetDoc(
+            db, 'doc-here', 'replica:1', '{"value": "here"}', False)
+
 
 load_tests = tests.load_with_scenarios

=== modified file 'u1db/tests/test_server_state.py'
--- u1db/tests/test_server_state.py	2012-05-16 08:39:54 +0000
+++ u1db/tests/test_server_state.py	2012-09-25 20:06:22 +0000
@@ -70,8 +70,9 @@
         self.state.set_workingdir(tempdir)
         path = tempdir + '/test.db'
         self.assertFalse(os.path.exists(path))
-        db = self.state.ensure_database('test.db')
+        db, replica_uid = self.state.ensure_database('test.db')
         self.assertIsInstance(db, sqlite_backend.SQLitePartialExpandDatabase)
+        self.assertEqual(db._replica_uid, replica_uid)
         self.assertTrue(os.path.exists(path))
         db2 = self.state.open_database('test.db')
         self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase)
@@ -80,7 +81,7 @@
         tempdir = self.createTempDir()
         self.state.set_workingdir(tempdir)
         path = tempdir + '/test.db'
-        db = self.state.ensure_database('test.db')
+        db, _ = self.state.ensure_database('test.db')
         db.close()
         self.state.delete_database('test.db')
         self.assertFalse(os.path.exists(path))

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2012-09-24 14:12:27 +0000
+++ u1db/tests/test_sync.py	2012-09-25 20:06:22 +0000
@@ -1178,9 +1178,9 @@
 
     oauth = False
 
-    def do_sync(self):
+    def do_sync(self, target_name):
         if self.oauth:
-            path = '~/test2.db'
+            path = '~/' + target_name
             extra = dict(creds={'oauth': {
                 'consumer_key': tests.consumer1.key,
                 'consumer_secret': tests.consumer1.secret,
@@ -1188,7 +1188,7 @@
                 'token_secret': tests.token1.secret
                 }})
         else:
-            path = 'test2.db'
+            path = target_name
             extra = {}
         target_url = self.getURL(path)
         return self.db.sync(target_url, **extra)
@@ -1202,7 +1202,7 @@
     def test_db_sync(self):
         doc1 = self.db.create_doc_from_json(tests.simple_doc)
         doc2 = self.db2.create_doc_from_json(tests.nested_doc)
-        local_gen_before_sync = self.do_sync()
+        local_gen_before_sync = self.do_sync('test2.db')
         gen, _, changes = self.db.whats_changed(local_gen_before_sync)
         self.assertEqual(1, len(changes))
         self.assertEqual(doc2.doc_id, changes[0][0])
@@ -1212,6 +1212,22 @@
         self.assertGetDoc(self.db, doc2.doc_id, doc2.rev, tests.nested_doc,
                           False)
 
+    def test_db_sync_autocreate(self):
+        doc1 = self.db.create_doc_from_json(tests.simple_doc)
+        local_gen_before_sync = self.do_sync('test3.db')
+        gen, _, changes = self.db.whats_changed(local_gen_before_sync)
+        self.assertEqual(0, gen - local_gen_before_sync)
+        db3 = self.request_state.open_database('test3.db')
+        gen, _, changes = db3.whats_changed()
+        self.assertEqual(1, len(changes))
+        self.assertEqual(doc1.doc_id, changes[0][0])
+        self.assertGetDoc(db3, doc1.doc_id, doc1.rev, tests.simple_doc,
+                          False)
+        t_gen, _ = self.db._get_replica_gen_and_trans_id('test3.db')
+        s_gen, _ = db3._get_replica_gen_and_trans_id('test1')
+        self.assertEqual(1, t_gen)
+        self.assertEqual(1, s_gen)
+
 
 class TestRemoteSyncIntegration(tests.TestCaseWithServer):
     """Integration tests for the most common sync scenario local -> remote"""