Added support so clients can detect if a server has potentially lost commits after multiple updates, like during bulk imports and so the replicator can detect lost commits on remote replications.

git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@733174 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 64 insertions, 14 deletions

diff --git a/src/couchdb/couch_db.erl b/src/couchdb/couch_db.erl
index c9546240..3011d744 100644
--- a/src/couchdb/couch_db.erl
+++ b/src/couchdb/couch_db.erl
@@ -64,8 +64,9 @@ create(DbName, Options) ->
 open(DbName, Options) ->
     couch_server:open(DbName, Options).
 
-ensure_full_commit(#db{update_pid=UpdatePid}) ->
-    gen_server:call(UpdatePid, full_commit, infinity).
+ensure_full_commit(#db{update_pid=UpdatePid,instance_start_time=StartTime}) ->
+    ok = gen_server:call(UpdatePid, full_commit, infinity),
+    {ok, StartTime}.
 
 close(#db{fd=Fd}) ->
     couch_file:drop_ref(Fd).
@@ -166,7 +167,8 @@ get_db_info(Db) ->
         compactor_pid=Compactor,
         update_seq=SeqNum,
         name=Name,
-        fulldocinfo_by_id_btree=FullDocBtree} = Db,
+        fulldocinfo_by_id_btree=FullDocBtree,
+        instance_start_time=StartTime} = Db,
     {ok, Size} = couch_file:bytes(Fd),
     {ok, {Count, DelCount}} = couch_btree:full_reduce(FullDocBtree),
     InfoList = [
@@ -176,7 +178,8 @@ get_db_info(Db) ->
         {update_seq, SeqNum},
         {purge_seq, couch_db:get_purge_seq(Db)},
         {compact_running, Compactor/=nil},
-        {disk_size, Size}
+        {disk_size, Size},
+        {instance_start_time, StartTime}
         ],
     {ok, InfoList}.
 
diff --git a/src/couchdb/couch_db.hrl b/src/couchdb/couch_db.hrl
index 830f2f9a..238025a2 100644
--- a/src/couchdb/couch_db.hrl
+++ b/src/couchdb/couch_db.hrl
@@ -120,6 +120,7 @@
     {main_pid=nil,
     update_pid=nil,
     compactor_pid=nil,
+    instance_start_time, % number of microsecs since jan 1 1970 as a binary string
     fd,
     header = #db_header{},
     summary_stream,
diff --git a/src/couchdb/couch_db_updater.erl b/src/couchdb/couch_db_updater.erl
index 94605a3c..cbeda223 100644
--- a/src/couchdb/couch_db_updater.erl
+++ b/src/couchdb/couch_db_updater.erl
@@ -279,6 +279,12 @@ init_db(DbName, Filepath, Fd, Header0) ->
     AdminsPtr ->
         {ok, Admins} = couch_file:pread_term(Fd, AdminsPtr)
     end,
+    
+    % convert start time tuple to microsecs and store as a binary string
+    {MegaSecs, Secs, MicroSecs} = now(),
+    StartTime = ?l2b(io_lib:format("~p",
+            [(MegaSecs*1000000*1000000) + (Secs*1000000) + MicroSecs])),
+    
     #db{
         update_pid=self(),
         fd=Fd,
@@ -291,7 +297,9 @@ init_db(DbName, Filepath, Fd, Header0) ->
         name = DbName,
         filepath = Filepath,
         admins = Admins,
-        admins_ptr = AdminsPtr}.
+        admins_ptr = AdminsPtr,
+        instance_start_time = StartTime
+        }.
 
 
 close_db(#db{fd=Fd,summary_stream=Ss}) ->
diff --git a/src/couchdb/couch_file.erl b/src/couchdb/couch_file.erl
index 9e60eb09..b29f45d2 100644
--- a/src/couchdb/couch_file.erl
+++ b/src/couchdb/couch_file.erl
@@ -170,7 +170,8 @@ close(Fd) ->
     Result.
     
 close_maybe(Fd) ->
-    gen_server:cast(Fd, {close_maybe, self()}).
+    catch unlink(Fd),
+    catch gen_server:cast(Fd, close_maybe).
 
 drop_ref(Fd) ->
     drop_ref(Fd, self()).
@@ -372,8 +373,7 @@ handle_call(num_refs, _From, Fd) ->
 
 handle_cast(close, Fd) ->
     {stop,normal,Fd};
-handle_cast({close_maybe, Pid}, Fd) ->
-    catch unlink(Pid),
+handle_cast(close_maybe, Fd) ->
     maybe_close_async(Fd);
 handle_cast({drop_ref, Pid}, Fd) ->
     case get(Pid) of
diff --git a/src/couchdb/couch_httpd_db.erl b/src/couchdb/couch_httpd_db.erl
index 5366da1e..0519061b 100644
--- a/src/couchdb/couch_httpd_db.erl
+++ b/src/couchdb/couch_httpd_db.erl
@@ -90,9 +90,10 @@ db_req(#httpd{path_parts=[_DbName]}=Req, _Db) ->
     send_method_not_allowed(Req, "DELETE,GET,HEAD,POST");
 
 db_req(#httpd{method='POST',path_parts=[_,<<"_ensure_full_commit">>]}=Req, Db) ->
-    ok = couch_db:ensure_full_commit(Db),
+    {ok, DbStartTime} = couch_db:ensure_full_commit(Db),
     send_json(Req, 201, {[
-            {ok, true}
+            {ok, true},
+            {instance_start_time, DbStartTime}
         ]});
     
 db_req(#httpd{path_parts=[_,<<"_ensure_full_commit">>]}=Req, _Db) ->
diff --git a/src/couchdb/couch_rep.erl b/src/couchdb/couch_rep.erl
index 881525f0..29f1fc80 100644
--- a/src/couchdb/couch_rep.erl
+++ b/src/couchdb/couch_rep.erl
@@ -69,7 +69,10 @@ replicate2(Source, DbSrc, Target, DbTgt, Options) ->
     RepRecKey = <<?LOCAL_DOC_PREFIX, HostNameBin/binary, 
             ":", Source/binary, ":", Target/binary>>,
     
-    StartTime = httpd_util:rfc1123_date(),
+    ReplicationStartTime = httpd_util:rfc1123_date(),
+    
+    {ok, SrcInstanceStartTime} = get_db_info(DbSrc),
+    {ok, TgtInstanceStartTime} = get_db_info(DbTgt),
     
     case proplists:get_value(full, Options, false)
         orelse proplists:get_value("full", Options, false) of
@@ -115,9 +118,28 @@ replicate2(Source, DbSrc, Target, DbTgt, Options) ->
         % nothing changed, don't record results
         {ok, {OldRepHistoryProps}};
     false ->
+        % commit changes to both src and tgt. The src because if changes
+        % we replicated are lost, we'll record the a seq number of ahead 
+        % of what was committed and therefore lose future changes with the
+        % same seq nums.
+        
+        {ok, SrcInstanceStartTime2} = ensure_full_commit(DbSrc),
+        {ok, TgtInstanceStartTime2} = ensure_full_commit(DbTgt),
+        
+        RecordSeqNum =
+        if SrcInstanceStartTime2 == SrcInstanceStartTime andalso
+                TgtInstanceStartTime2 == TgtInstanceStartTime ->
+            NewSeqNum;
+        true ->
+            ?LOG_INFO("A server has restarted sinced replication start. "
+                "Not recording the new sequence number to ensure the "
+                "replication is redone and documents reexamined.", []),
+            SeqNum
+        end,
+        
         HistEntries =[
             {
-                [{<<"start_time">>, list_to_binary(StartTime)},
+                [{<<"start_time">>, list_to_binary(ReplicationStartTime)},
                 {<<"end_time">>, list_to_binary(httpd_util:rfc1123_date())},
                 {<<"start_last_seq">>, SeqNum},
                 {<<"end_last_seq">>, NewSeqNum} | Stats]}
@@ -126,7 +148,7 @@ replicate2(Source, DbSrc, Target, DbTgt, Options) ->
         NewRepHistory =
             {
                 [{<<"session_id">>, couch_util:new_uuid()},
-                {<<"source_last_seq">>, NewSeqNum},
+                {<<"source_last_seq">>, RecordSeqNum},
                 {<<"history">>, lists:sublist(HistEntries, 50)}]},
 
         {ok, _} = update_doc(DbSrc, RepRecSrc#doc{body=NewRepHistory}, []),
@@ -276,6 +298,21 @@ close_db(#http_db{})->
 close_db(Db)->
     couch_db:close(Db).
 
+get_db_info(#http_db{uri=DbUrl, headers=Headers}) ->
+    {DbProps} = do_http_request(DbUrl, get, Headers),
+    {ok, [{list_to_existing_atom(?b2l(K)), V} || {K,V} <- DbProps]};
+get_db_info(Db) ->
+    couch_db:get_db_info(Db).
+
+
+ensure_full_commit(#http_db{uri=DbUrl, headers=Headers}) ->
+    {ResultProps} = do_http_request(DbUrl ++ "_ensure_full_commit", post, Headers, true),
+    true = proplists:get_value(<<"ok">>, ResultProps),
+    {ok, proplists:get_value(<<"instance_start_time">>, ResultProps)};
+ensure_full_commit(Db) ->
+    couch_db:ensure_full_commit(Db).
+    
+    
 get_doc_info_list(#http_db{uri=DbUrl, headers=Headers}, StartSeq) ->
     Url = DbUrl ++ "_all_docs_by_seq?limit=100&startkey=" 
         ++ integer_to_list(StartSeq),
diff --git a/src/couchdb/couch_view_group.erl b/src/couchdb/couch_view_group.erl
index 45e92dce..68c6c5cb 100644
--- a/src/couchdb/couch_view_group.erl
+++ b/src/couchdb/couch_view_group.erl
@@ -149,7 +149,7 @@ handle_info(delayed_commit, #group_state{db_name=DbName,group=Group}=State) ->
         {noreply, State#group_state{waiting_commit=false}};
     true ->
         % We can't commit the header because the database seq that's fully
-        % committed to disk is still behind us. It we committed now and the
+        % committed to disk is still behind us. If we committed now and the
         % database lost those changes our view could be forever out of sync
         % with the database. But a crash before we commit these changes, no big
         % deal, we only lose incremental changes since last committal.