summaryrefslogtreecommitdiff
path: root/src/couchdb
diff options
context:
space:
mode:
Diffstat (limited to 'src/couchdb')
-rw-r--r--src/couchdb/couch_db.erl11
-rw-r--r--src/couchdb/couch_db.hrl1
-rw-r--r--src/couchdb/couch_db_updater.erl10
-rw-r--r--src/couchdb/couch_file.erl6
-rw-r--r--src/couchdb/couch_httpd_db.erl5
-rw-r--r--src/couchdb/couch_rep.erl43
-rw-r--r--src/couchdb/couch_view_group.erl2
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.
generated by cgit v1.2.3 (git 2.25.1) at 2024-07-09 04:13:21 +0000