| % 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. |
| |
| -module(couch_db_updater). |
| -behaviour(gen_server). |
| -vsn(1). |
| |
| -export([add_sizes/3, upgrade_sizes/1]). |
| -export([init/1,terminate/2,handle_call/3,handle_cast/2,code_change/3,handle_info/2]). |
| |
| -include_lib("couch/include/couch_db.hrl"). |
| -include("couch_db_int.hrl"). |
| |
| -define(IDLE_LIMIT_DEFAULT, 61000). |
| -define(DEFAULT_MAX_PARTITION_SIZE, 16#280000000). % 10 GiB |
| |
| |
| -record(merge_acc, { |
| revs_limit, |
| merge_conflicts, |
| add_infos = [], |
| rem_seqs = [], |
| cur_seq, |
| full_partitions = [] |
| }). |
| |
| |
| init({Engine, DbName, FilePath, Options0}) -> |
| erlang:put(io_priority, {db_update, DbName}), |
| update_idle_limit_from_config(), |
| DefaultSecObj = default_security_object(DbName), |
| Options = [{default_security_object, DefaultSecObj} | Options0], |
| try |
| {ok, EngineState} = couch_db_engine:init(Engine, FilePath, Options), |
| Db = init_db(DbName, FilePath, EngineState, Options), |
| case lists:member(sys_db, Options) of |
| false -> |
| couch_stats_process_tracker:track([couchdb, open_databases]); |
| true -> |
| ok |
| end, |
| % Don't load validation funs here because the fabric query is |
| % liable to race conditions. Instead see |
| % couch_db:validate_doc_update, which loads them lazily. |
| NewDb = Db#db{main_pid = self()}, |
| proc_lib:init_ack({ok, NewDb}), |
| gen_server:enter_loop(?MODULE, [], NewDb, idle_limit()) |
| catch |
| throw:InitError -> |
| proc_lib:init_ack(InitError) |
| end. |
| |
| |
| terminate(Reason, Db) -> |
| couch_util:shutdown_sync(Db#db.compactor_pid), |
| couch_db_engine:terminate(Reason, Db), |
| ok. |
| |
| handle_call(get_db, _From, Db) -> |
| {reply, {ok, Db}, Db, idle_limit()}; |
| handle_call(full_commit, _From, #db{waiting_delayed_commit=nil}=Db) -> |
| {reply, ok, Db, idle_limit()}; % no data waiting, return ok immediately |
| handle_call(full_commit, _From, Db) -> |
| {reply, ok, commit_data(Db), idle_limit()}; |
| handle_call({full_commit, RequiredSeq}, _From, Db) |
| when RequiredSeq =< Db#db.committed_update_seq -> |
| {reply, ok, Db, idle_limit()}; |
| handle_call({full_commit, _}, _, Db) -> |
| {reply, ok, commit_data(Db), idle_limit()}; % commit the data and return ok |
| handle_call(start_compact, _From, Db) -> |
| {noreply, NewDb, _Timeout} = handle_cast(start_compact, Db), |
| {reply, {ok, NewDb#db.compactor_pid}, NewDb, idle_limit()}; |
| handle_call(compactor_pid, _From, #db{compactor_pid = Pid} = Db) -> |
| {reply, Pid, Db, idle_limit()}; |
| handle_call(cancel_compact, _From, #db{compactor_pid = nil} = Db) -> |
| {reply, ok, Db, idle_limit()}; |
| handle_call(cancel_compact, _From, #db{compactor_pid = Pid} = Db) -> |
| unlink(Pid), |
| exit(Pid, kill), |
| couch_server:delete_compaction_files(Db#db.name), |
| Db2 = Db#db{compactor_pid = nil}, |
| ok = gen_server:call(couch_server, {db_updated, Db2}, infinity), |
| {reply, ok, Db2, idle_limit()}; |
| |
| handle_call({set_security, NewSec}, _From, #db{} = Db) -> |
| {ok, NewDb} = couch_db_engine:set_security(Db, NewSec), |
| NewSecDb = commit_data(NewDb#db{ |
| security = NewSec |
| }), |
| ok = gen_server:call(couch_server, {db_updated, NewSecDb}, infinity), |
| {reply, ok, NewSecDb, idle_limit()}; |
| |
| handle_call({set_revs_limit, Limit}, _From, Db) -> |
| {ok, Db2} = couch_db_engine:set_revs_limit(Db, Limit), |
| Db3 = commit_data(Db2), |
| ok = gen_server:call(couch_server, {db_updated, Db3}, infinity), |
| {reply, ok, Db3, idle_limit()}; |
| |
| handle_call({set_purge_infos_limit, Limit}, _From, Db) -> |
| {ok, Db2} = couch_db_engine:set_purge_infos_limit(Db, Limit), |
| ok = gen_server:call(couch_server, {db_updated, Db2}, infinity), |
| {reply, ok, Db2, idle_limit()}; |
| |
| handle_call({purge_docs, [], _}, _From, Db) -> |
| {reply, {ok, []}, Db, idle_limit()}; |
| |
| handle_call({purge_docs, PurgeReqs0, Options}, _From, Db) -> |
| % Filter out any previously applied updates during |
| % internal replication |
| IsRepl = lists:member(replicated_changes, Options), |
| PurgeReqs = if not IsRepl -> PurgeReqs0; true -> |
| UUIDs = [UUID || {UUID, _Id, _Revs} <- PurgeReqs0], |
| PurgeInfos = couch_db_engine:load_purge_infos(Db, UUIDs), |
| lists:flatmap(fun |
| ({not_found, PReq}) -> [PReq]; |
| ({{_, _, _, _}, _}) -> [] |
| end, lists:zip(PurgeInfos, PurgeReqs0)) |
| end, |
| {ok, NewDb, Replies} = purge_docs(Db, PurgeReqs), |
| {reply, {ok, Replies}, NewDb, idle_limit()}; |
| |
| handle_call(Msg, From, Db) -> |
| case couch_db_engine:handle_db_updater_call(Msg, From, Db) of |
| {reply, Resp, NewDb} -> |
| {reply, Resp, NewDb, idle_limit()}; |
| Else -> |
| Else |
| end. |
| |
| |
| handle_cast({load_validation_funs, ValidationFuns}, Db) -> |
| Db2 = Db#db{validate_doc_funs = ValidationFuns}, |
| ok = gen_server:call(couch_server, {db_updated, Db2}, infinity), |
| {noreply, Db2, idle_limit()}; |
| handle_cast(start_compact, Db) -> |
| case Db#db.compactor_pid of |
| nil -> |
| % For now we only support compacting to the same |
| % storage engine. After the first round of patches |
| % we'll add a field that sets the target engine |
| % type to compact to with a new copy compactor. |
| UpdateSeq = couch_db_engine:get_update_seq(Db), |
| Args = [Db#db.name, UpdateSeq], |
| couch_log:info("Starting compaction for db \"~s\" at ~p", Args), |
| {ok, Db2} = couch_db_engine:start_compaction(Db), |
| ok = gen_server:call(couch_server, {db_updated, Db2}, infinity), |
| {noreply, Db2, idle_limit()}; |
| _ -> |
| % compact currently running, this is a no-op |
| {noreply, Db, idle_limit()} |
| end; |
| handle_cast({compact_done, _Engine, CompactInfo}, #db{} = OldDb) -> |
| {ok, NewDb} = couch_db_engine:finish_compaction(OldDb, CompactInfo), |
| {noreply, NewDb}; |
| |
| handle_cast(wakeup, Db) -> |
| {noreply, Db, idle_limit()}; |
| |
| handle_cast(Msg, #db{name = Name} = Db) -> |
| couch_log:error("Database `~s` updater received unexpected cast: ~p", |
| [Name, Msg]), |
| {stop, Msg, Db}. |
| |
| |
| handle_info({update_docs, Client, GroupedDocs, NonRepDocs, MergeConflicts, |
| FullCommit}, Db) -> |
| GroupedDocs2 = sort_and_tag_grouped_docs(Client, GroupedDocs), |
| if NonRepDocs == [] -> |
| {GroupedDocs3, Clients, FullCommit2} = collect_updates(GroupedDocs2, |
| [Client], MergeConflicts, FullCommit); |
| true -> |
| GroupedDocs3 = GroupedDocs2, |
| FullCommit2 = FullCommit, |
| Clients = [Client] |
| end, |
| NonRepDocs2 = [{Client, NRDoc} || NRDoc <- NonRepDocs], |
| try update_docs_int(Db, GroupedDocs3, NonRepDocs2, MergeConflicts, |
| FullCommit2) of |
| {ok, Db2, UpdatedDDocIds} -> |
| ok = gen_server:call(couch_server, {db_updated, Db2}, infinity), |
| case {couch_db:get_update_seq(Db), couch_db:get_update_seq(Db2)} of |
| {Seq, Seq} -> ok; |
| _ -> couch_event:notify(Db2#db.name, updated) |
| end, |
| if NonRepDocs2 /= [] -> |
| couch_event:notify(Db2#db.name, local_updated); |
| true -> ok |
| end, |
| [catch(ClientPid ! {done, self()}) || ClientPid <- Clients], |
| Db3 = case length(UpdatedDDocIds) > 0 of |
| true -> |
| % Ken and ddoc_cache are the only things that |
| % use the unspecified ddoc_updated message. We |
| % should update them to use the new message per |
| % ddoc. |
| lists:foreach(fun(DDocId) -> |
| couch_event:notify(Db2#db.name, {ddoc_updated, DDocId}) |
| end, UpdatedDDocIds), |
| couch_event:notify(Db2#db.name, ddoc_updated), |
| ddoc_cache:refresh(Db2#db.name, UpdatedDDocIds), |
| refresh_validate_doc_funs(Db2); |
| false -> |
| Db2 |
| end, |
| {noreply, Db3, hibernate_if_no_idle_limit()} |
| catch |
| throw: retry -> |
| [catch(ClientPid ! {retry, self()}) || ClientPid <- Clients], |
| {noreply, Db, hibernate_if_no_idle_limit()} |
| end; |
| handle_info(delayed_commit, #db{waiting_delayed_commit=nil}=Db) -> |
| %no outstanding delayed commits, ignore |
| {noreply, Db, idle_limit()}; |
| handle_info(delayed_commit, Db) -> |
| case commit_data(Db) of |
| Db -> |
| {noreply, Db, idle_limit()}; |
| Db2 -> |
| ok = gen_server:call(couch_server, {db_updated, Db2}, infinity), |
| {noreply, Db2, idle_limit()} |
| end; |
| handle_info({'EXIT', _Pid, normal}, Db) -> |
| {noreply, Db, idle_limit()}; |
| handle_info({'EXIT', _Pid, Reason}, Db) -> |
| {stop, Reason, Db}; |
| handle_info(timeout, #db{name=DbName} = Db) -> |
| IdleLimitMSec = update_idle_limit_from_config(), |
| case couch_db:is_idle(Db) of |
| true -> |
| LastActivity = couch_db_engine:last_activity(Db), |
| DtMSec = timer:now_diff(os:timestamp(), LastActivity) div 1000, |
| MSecSinceLastActivity = max(0, DtMSec), |
| case MSecSinceLastActivity > IdleLimitMSec of |
| true -> |
| ok = couch_server:close_db_if_idle(DbName); |
| false -> |
| ok |
| end; |
| false -> |
| ok |
| end, |
| % Send a message to wake up and then hibernate. Hibernation here is done to |
| % force a thorough garbage collection. |
| gen_server:cast(self(), wakeup), |
| {noreply, Db, hibernate}; |
| |
| handle_info(Msg, Db) -> |
| case couch_db_engine:handle_db_updater_info(Msg, Db) of |
| {noreply, NewDb} -> |
| {noreply, NewDb, idle_limit()}; |
| Else -> |
| Else |
| end. |
| |
| |
| code_change(_OldVsn, State, _Extra) -> |
| {ok, State}. |
| |
| sort_and_tag_grouped_docs(Client, GroupedDocs) -> |
| % These groups should already be sorted but sometimes clients misbehave. |
| % The merge_updates function will fail and the database can end up with |
| % duplicate documents if the incoming groups are not sorted, so as a sanity |
| % check we sort them again here. See COUCHDB-2735. |
| Cmp = fun([#doc{id=A}|_], [#doc{id=B}|_]) -> A < B end, |
| lists:map(fun(DocGroup) -> |
| [{Client, maybe_tag_doc(D)} || D <- DocGroup] |
| end, lists:sort(Cmp, GroupedDocs)). |
| |
| maybe_tag_doc(#doc{id=Id, revs={Pos,[_Rev|PrevRevs]}, meta=Meta0}=Doc) -> |
| case lists:keymember(ref, 1, Meta0) of |
| true -> |
| Doc; |
| false -> |
| Key = {Id, {Pos-1, PrevRevs}}, |
| Doc#doc{meta=[{ref, Key} | Meta0]} |
| end. |
| |
| merge_updates([[{_,#doc{id=X}}|_]=A|RestA], [[{_,#doc{id=X}}|_]=B|RestB]) -> |
| [A++B | merge_updates(RestA, RestB)]; |
| merge_updates([[{_,#doc{id=X}}|_]|_]=A, [[{_,#doc{id=Y}}|_]|_]=B) when X < Y -> |
| [hd(A) | merge_updates(tl(A), B)]; |
| merge_updates([[{_,#doc{id=X}}|_]|_]=A, [[{_,#doc{id=Y}}|_]|_]=B) when X > Y -> |
| [hd(B) | merge_updates(A, tl(B))]; |
| merge_updates([], RestB) -> |
| RestB; |
| merge_updates(RestA, []) -> |
| RestA. |
| |
| collect_updates(GroupedDocsAcc, ClientsAcc, MergeConflicts, FullCommit) -> |
| receive |
| % Only collect updates with the same MergeConflicts flag and without |
| % local docs. It's easier to just avoid multiple _local doc |
| % updaters than deal with their possible conflicts, and local docs |
| % writes are relatively rare. Can be optmized later if really needed. |
| {update_docs, Client, GroupedDocs, [], MergeConflicts, FullCommit2} -> |
| GroupedDocs2 = sort_and_tag_grouped_docs(Client, GroupedDocs), |
| GroupedDocsAcc2 = |
| merge_updates(GroupedDocsAcc, GroupedDocs2), |
| collect_updates(GroupedDocsAcc2, [Client | ClientsAcc], |
| MergeConflicts, (FullCommit or FullCommit2)) |
| after 0 -> |
| {GroupedDocsAcc, ClientsAcc, FullCommit} |
| end. |
| |
| |
| init_db(DbName, FilePath, EngineState, Options) -> |
| % convert start time tuple to microsecs and store as a binary string |
| {MegaSecs, Secs, MicroSecs} = os:timestamp(), |
| StartTime = ?l2b(io_lib:format("~p", |
| [(MegaSecs*1000000*1000000) + (Secs*1000000) + MicroSecs])), |
| |
| BDU = couch_util:get_value(before_doc_update, Options, nil), |
| ADR = couch_util:get_value(after_doc_read, Options, nil), |
| |
| NonCreateOpts = [Opt || Opt <- Options, Opt /= create], |
| |
| InitDb = #db{ |
| name = DbName, |
| filepath = FilePath, |
| engine = EngineState, |
| instance_start_time = StartTime, |
| options = NonCreateOpts, |
| before_doc_update = BDU, |
| after_doc_read = ADR |
| }, |
| |
| DbProps = couch_db_engine:get_props(InitDb), |
| |
| InitDb#db{ |
| committed_update_seq = couch_db_engine:get_update_seq(InitDb), |
| security = couch_db_engine:get_security(InitDb), |
| options = lists:keystore(props, 1, NonCreateOpts, {props, DbProps}) |
| }. |
| |
| |
| refresh_validate_doc_funs(#db{name = <<"shards/", _/binary>> = Name} = Db) -> |
| spawn(fabric, reset_validation_funs, [mem3:dbname(Name)]), |
| Db#db{validate_doc_funs = undefined}; |
| refresh_validate_doc_funs(Db0) -> |
| Db = Db0#db{user_ctx=?ADMIN_USER}, |
| {ok, DesignDocs} = couch_db:get_design_docs(Db), |
| ProcessDocFuns = lists:flatmap( |
| fun(DesignDocInfo) -> |
| {ok, DesignDoc} = couch_db:open_doc_int( |
| Db, DesignDocInfo, [ejson_body]), |
| case couch_doc:get_validate_doc_fun(DesignDoc) of |
| nil -> []; |
| Fun -> [Fun] |
| end |
| end, DesignDocs), |
| Db#db{validate_doc_funs=ProcessDocFuns}. |
| |
| % rev tree functions |
| |
| flush_trees(_Db, [], AccFlushedTrees) -> |
| {ok, lists:reverse(AccFlushedTrees)}; |
| flush_trees(#db{} = Db, |
| [InfoUnflushed | RestUnflushed], AccFlushed) -> |
| #full_doc_info{update_seq=UpdateSeq, rev_tree=Unflushed} = InfoUnflushed, |
| {Flushed, FinalAcc} = couch_key_tree:mapfold( |
| fun(_Rev, Value, Type, SizesAcc) -> |
| case Value of |
| % This node is a document summary that needs to be |
| % flushed to disk. |
| #doc{} = Doc -> |
| check_doc_atts(Db, Doc), |
| ExternalSize = get_meta_body_size(Value#doc.meta), |
| {size_info, AttSizeInfo} = |
| lists:keyfind(size_info, 1, Doc#doc.meta), |
| {ok, NewDoc, WrittenSize} = |
| couch_db_engine:write_doc_body(Db, Doc), |
| Leaf = #leaf{ |
| deleted = Doc#doc.deleted, |
| ptr = NewDoc#doc.body, |
| seq = UpdateSeq, |
| sizes = #size_info{ |
| active = WrittenSize, |
| external = ExternalSize |
| }, |
| atts = AttSizeInfo |
| }, |
| {Leaf, add_sizes(Type, Leaf, SizesAcc)}; |
| #leaf{} -> |
| {Value, add_sizes(Type, Value, SizesAcc)}; |
| _ -> |
| {Value, SizesAcc} |
| end |
| end, {0, 0, []}, Unflushed), |
| {FinalAS, FinalES, FinalAtts} = FinalAcc, |
| TotalAttSize = lists:foldl(fun({_, S}, A) -> S + A end, 0, FinalAtts), |
| NewInfo = InfoUnflushed#full_doc_info{ |
| rev_tree = Flushed, |
| sizes = #size_info{ |
| active = FinalAS + TotalAttSize, |
| external = FinalES + TotalAttSize |
| } |
| }, |
| flush_trees(Db, RestUnflushed, [NewInfo | AccFlushed]). |
| |
| |
| check_doc_atts(Db, Doc) -> |
| {atts_stream, Stream} = lists:keyfind(atts_stream, 1, Doc#doc.meta), |
| % Make sure that the attachments were written to the currently |
| % active attachment stream. If compaction swaps during a write |
| % request we may have to rewrite our attachment bodies. |
| if Stream == nil -> ok; true -> |
| case couch_db:is_active_stream(Db, Stream) of |
| true -> |
| ok; |
| false -> |
| % Stream where the attachments were written to is |
| % no longer the current attachment stream. This |
| % can happen when a database is switched at |
| % compaction time. |
| couch_log:debug("Stream where the attachments were" |
| " written has changed." |
| " Possibly retrying.", []), |
| throw(retry) |
| end |
| end. |
| |
| |
| add_sizes(Type, #leaf{sizes=Sizes, atts=AttSizes}, Acc) -> |
| % Maybe upgrade from disk_size only |
| #size_info{ |
| active = ActiveSize, |
| external = ExternalSize |
| } = upgrade_sizes(Sizes), |
| {ASAcc, ESAcc, AttsAcc} = Acc, |
| NewASAcc = ActiveSize + ASAcc, |
| NewESAcc = ESAcc + if Type == leaf -> ExternalSize; true -> 0 end, |
| NewAttsAcc = lists:umerge(AttSizes, AttsAcc), |
| {NewASAcc, NewESAcc, NewAttsAcc}. |
| |
| |
| upgrade_sizes(#size_info{}=SI) -> |
| SI; |
| upgrade_sizes({D, E}) -> |
| #size_info{active=D, external=E}; |
| upgrade_sizes(S) when is_integer(S) -> |
| #size_info{active=S, external=0}. |
| |
| |
| send_result(Client, Doc, NewResult) -> |
| % used to send a result to the client |
| catch(Client ! {result, self(), {doc_tag(Doc), NewResult}}). |
| |
| doc_tag(#doc{meta=Meta}) -> |
| case lists:keyfind(ref, 1, Meta) of |
| {ref, Ref} -> Ref; |
| false -> throw(no_doc_tag); |
| Else -> throw({invalid_doc_tag, Else}) |
| end. |
| |
| merge_rev_trees([], [], Acc) -> |
| {ok, Acc#merge_acc{ |
| add_infos = lists:reverse(Acc#merge_acc.add_infos) |
| }}; |
| merge_rev_trees([NewDocs | RestDocsList], [OldDocInfo | RestOldInfo], Acc) -> |
| #merge_acc{ |
| revs_limit = Limit, |
| merge_conflicts = MergeConflicts, |
| full_partitions = FullPartitions |
| } = Acc, |
| |
| % Track doc ids so we can debug large revision trees |
| erlang:put(last_id_merged, OldDocInfo#full_doc_info.id), |
| NewDocInfo0 = lists:foldl(fun({Client, NewDoc}, OldInfoAcc) -> |
| NewInfo = merge_rev_tree(OldInfoAcc, NewDoc, Client, MergeConflicts), |
| case is_overflowed(NewInfo, OldInfoAcc, FullPartitions) of |
| true when not MergeConflicts -> |
| DocId = NewInfo#full_doc_info.id, |
| send_result(Client, NewDoc, {partition_overflow, DocId}), |
| OldInfoAcc; |
| _ -> |
| NewInfo |
| end |
| end, OldDocInfo, NewDocs), |
| NewDocInfo1 = maybe_stem_full_doc_info(NewDocInfo0, Limit), |
| % When MergeConflicts is false, we updated #full_doc_info.deleted on every |
| % iteration of merge_rev_tree. However, merge_rev_tree does not update |
| % #full_doc_info.deleted when MergeConflicts is true, since we don't need |
| % to know whether the doc is deleted between iterations. Since we still |
| % need to know if the doc is deleted after the merge happens, we have to |
| % set it here. |
| NewDocInfo2 = case MergeConflicts of |
| true -> |
| NewDocInfo1#full_doc_info{ |
| deleted = couch_doc:is_deleted(NewDocInfo1) |
| }; |
| false -> |
| NewDocInfo1 |
| end, |
| if NewDocInfo2 == OldDocInfo -> |
| % nothing changed |
| merge_rev_trees(RestDocsList, RestOldInfo, Acc); |
| true -> |
| % We have updated the document, give it a new update_seq. Its |
| % important to note that the update_seq on OldDocInfo should |
| % be identical to the value on NewDocInfo1. |
| OldSeq = OldDocInfo#full_doc_info.update_seq, |
| NewDocInfo3 = NewDocInfo2#full_doc_info{ |
| update_seq = Acc#merge_acc.cur_seq + 1 |
| }, |
| RemoveSeqs = case OldSeq of |
| 0 -> Acc#merge_acc.rem_seqs; |
| _ -> [OldSeq | Acc#merge_acc.rem_seqs] |
| end, |
| NewAcc = Acc#merge_acc{ |
| add_infos = [NewDocInfo3 | Acc#merge_acc.add_infos], |
| rem_seqs = RemoveSeqs, |
| cur_seq = Acc#merge_acc.cur_seq + 1 |
| }, |
| merge_rev_trees(RestDocsList, RestOldInfo, NewAcc) |
| end. |
| |
| merge_rev_tree(OldInfo, NewDoc, Client, false) |
| when OldInfo#full_doc_info.deleted -> |
| % We're recreating a document that was previously |
| % deleted. To check that this is a recreation from |
| % the root we assert that the new document has a |
| % revision depth of 1 (this is to avoid recreating a |
| % doc from a previous internal revision) and is also |
| % not deleted. To avoid expanding the revision tree |
| % unnecessarily we create a new revision based on |
| % the winning deleted revision. |
| |
| {RevDepth, _} = NewDoc#doc.revs, |
| NewDeleted = NewDoc#doc.deleted, |
| case RevDepth == 1 andalso not NewDeleted of |
| true -> |
| % Update the new doc based on revisions in OldInfo |
| #doc_info{revs=[WinningRev | _]} = couch_doc:to_doc_info(OldInfo), |
| #rev_info{rev={OldPos, OldRev}} = WinningRev, |
| Body = case couch_util:get_value(comp_body, NewDoc#doc.meta) of |
| CompBody when is_binary(CompBody) -> |
| couch_compress:decompress(CompBody); |
| _ -> |
| NewDoc#doc.body |
| end, |
| RevIdDoc = NewDoc#doc{ |
| revs = {OldPos, [OldRev]}, |
| body = Body |
| }, |
| NewRevId = couch_db:new_revid(RevIdDoc), |
| NewDoc2 = NewDoc#doc{revs={OldPos + 1, [NewRevId, OldRev]}}, |
| |
| % Merge our modified new doc into the tree |
| #full_doc_info{rev_tree=OldTree} = OldInfo, |
| NewTree0 = couch_doc:to_path(NewDoc2), |
| case couch_key_tree:merge(OldTree, NewTree0) of |
| {NewTree1, new_leaf} -> |
| % We changed the revision id so inform the caller |
| send_result(Client, NewDoc, {ok, {OldPos+1, NewRevId}}), |
| OldInfo#full_doc_info{ |
| rev_tree = NewTree1, |
| deleted = false |
| }; |
| _ -> |
| throw(doc_recreation_failed) |
| end; |
| _ -> |
| send_result(Client, NewDoc, conflict), |
| OldInfo |
| end; |
| merge_rev_tree(OldInfo, NewDoc, Client, false) -> |
| % We're attempting to merge a new revision into an |
| % undeleted document. To not be a conflict we require |
| % that the merge results in extending a branch. |
| |
| OldTree = OldInfo#full_doc_info.rev_tree, |
| NewTree0 = couch_doc:to_path(NewDoc), |
| NewDeleted = NewDoc#doc.deleted, |
| case couch_key_tree:merge(OldTree, NewTree0) of |
| {NewTree, new_leaf} when not NewDeleted -> |
| OldInfo#full_doc_info{ |
| rev_tree = NewTree, |
| deleted = false |
| }; |
| {NewTree, new_leaf} when NewDeleted -> |
| % We have to check if we just deleted this |
| % document completely or if it was a conflict |
| % resolution. |
| OldInfo#full_doc_info{ |
| rev_tree = NewTree, |
| deleted = couch_doc:is_deleted(NewTree) |
| }; |
| _ -> |
| send_result(Client, NewDoc, conflict), |
| OldInfo |
| end; |
| merge_rev_tree(OldInfo, NewDoc, _Client, true) -> |
| % We're merging in revisions without caring about |
| % conflicts. Most likely this is a replication update. |
| OldTree = OldInfo#full_doc_info.rev_tree, |
| NewTree0 = couch_doc:to_path(NewDoc), |
| {NewTree, _} = couch_key_tree:merge(OldTree, NewTree0), |
| OldInfo#full_doc_info{rev_tree = NewTree}. |
| |
| is_overflowed(_New, _Old, []) -> |
| false; |
| is_overflowed(Old, Old, _FullPartitions) -> |
| false; |
| is_overflowed(New, Old, FullPartitions) -> |
| case New#full_doc_info.id of |
| <<"_design/", _/binary>> -> |
| false; |
| DDocId -> |
| Partition = couch_partition:from_docid(DDocId), |
| case lists:member(Partition, FullPartitions) of |
| true -> |
| estimate_size(New) > estimate_size(Old); |
| false -> |
| false |
| end |
| end. |
| |
| maybe_stem_full_doc_info(#full_doc_info{rev_tree = Tree} = Info, Limit) -> |
| case config:get_boolean("couchdb", "stem_interactive_updates", true) of |
| true -> |
| Stemmed = couch_key_tree:stem(Tree, Limit), |
| Info#full_doc_info{rev_tree = Stemmed}; |
| false -> |
| Info |
| end. |
| |
| update_docs_int(Db, DocsList, LocalDocs, MergeConflicts, FullCommit) -> |
| UpdateSeq = couch_db_engine:get_update_seq(Db), |
| RevsLimit = couch_db_engine:get_revs_limit(Db), |
| |
| Ids = [Id || [{_Client, #doc{id=Id}}|_] <- DocsList], |
| % lookup up the old documents, if they exist. |
| OldDocLookups = couch_db_engine:open_docs(Db, Ids), |
| OldDocInfos = lists:zipwith(fun |
| (_Id, #full_doc_info{} = FDI) -> |
| FDI; |
| (Id, not_found) -> |
| #full_doc_info{id=Id} |
| end, Ids, OldDocLookups), |
| |
| %% Get the list of full partitions |
| FullPartitions = case couch_db:is_partitioned(Db) of |
| true -> |
| case max_partition_size() of |
| N when N =< 0 -> |
| []; |
| Max -> |
| Partitions = lists:usort(lists:flatmap(fun(Id) -> |
| case couch_partition:extract(Id) of |
| undefined -> []; |
| {Partition, _} -> [Partition] |
| end |
| end, Ids)), |
| [P || P <- Partitions, partition_size(Db, P) >= Max] |
| end; |
| false -> |
| [] |
| end, |
| |
| % Merge the new docs into the revision trees. |
| AccIn = #merge_acc{ |
| revs_limit = RevsLimit, |
| merge_conflicts = MergeConflicts, |
| add_infos = [], |
| rem_seqs = [], |
| cur_seq = UpdateSeq, |
| full_partitions = FullPartitions |
| }, |
| {ok, AccOut} = merge_rev_trees(DocsList, OldDocInfos, AccIn), |
| #merge_acc{ |
| add_infos = NewFullDocInfos, |
| rem_seqs = RemSeqs |
| } = AccOut, |
| |
| % Write out the document summaries (the bodies are stored in the nodes of |
| % the trees, the attachments are already written to disk) |
| {ok, IndexFDIs} = flush_trees(Db, NewFullDocInfos, []), |
| Pairs = pair_write_info(OldDocLookups, IndexFDIs), |
| LocalDocs2 = update_local_doc_revs(LocalDocs), |
| |
| {ok, Db1} = couch_db_engine:write_doc_infos(Db, Pairs, LocalDocs2), |
| |
| WriteCount = length(IndexFDIs), |
| couch_stats:increment_counter([couchdb, document_inserts], |
| WriteCount - length(RemSeqs)), |
| couch_stats:increment_counter([couchdb, document_writes], WriteCount), |
| couch_stats:increment_counter( |
| [couchdb, local_document_writes], |
| length(LocalDocs2) |
| ), |
| |
| % Check if we just updated any design documents, and update the validation |
| % funs if we did. |
| UpdatedDDocIds = lists:flatmap(fun |
| (<<"_design/", _/binary>> = Id) -> [Id]; |
| (_) -> [] |
| end, Ids), |
| |
| {ok, commit_data(Db1, not FullCommit), UpdatedDDocIds}. |
| |
| |
| update_local_doc_revs(Docs) -> |
| lists:foldl(fun({Client, Doc}, Acc) -> |
| case increment_local_doc_revs(Doc) of |
| {ok, #doc{revs = {0, [NewRev]}} = NewDoc} -> |
| send_result(Client, Doc, {ok, {0, integer_to_binary(NewRev)}}), |
| [NewDoc | Acc]; |
| {error, Error} -> |
| send_result(Client, Doc, {error, Error}), |
| Acc |
| end |
| end, [], Docs). |
| |
| |
| increment_local_doc_revs(#doc{deleted = true} = Doc) -> |
| {ok, Doc#doc{revs = {0, [0]}}}; |
| increment_local_doc_revs(#doc{revs = {0, []}} = Doc) -> |
| {ok, Doc#doc{revs = {0, [1]}}}; |
| increment_local_doc_revs(#doc{revs = {0, [RevStr | _]}} = Doc) -> |
| try |
| PrevRev = binary_to_integer(RevStr), |
| {ok, Doc#doc{revs = {0, [PrevRev + 1]}}} |
| catch error:badarg -> |
| {error, <<"Invalid rev format">>} |
| end; |
| increment_local_doc_revs(#doc{}) -> |
| {error, <<"Invalid rev format">>}. |
| |
| max_partition_size() -> |
| config:get_integer("couchdb", "max_partition_size", |
| ?DEFAULT_MAX_PARTITION_SIZE). |
| |
| partition_size(Db, Partition) -> |
| {ok, Info} = couch_db:get_partition_info(Db, Partition), |
| Sizes = couch_util:get_value(sizes, Info), |
| couch_util:get_value(external, Sizes). |
| |
| estimate_size(#full_doc_info{} = FDI) -> |
| #full_doc_info{rev_tree = RevTree} = FDI, |
| Fun = fun |
| (_Rev, Value, leaf, SizesAcc) -> |
| case Value of |
| #doc{} = Doc -> |
| ExternalSize = get_meta_body_size(Value#doc.meta), |
| {size_info, AttSizeInfo} = |
| lists:keyfind(size_info, 1, Doc#doc.meta), |
| Leaf = #leaf{ |
| sizes = #size_info{ |
| external = ExternalSize |
| }, |
| atts = AttSizeInfo |
| }, |
| add_sizes(leaf, Leaf, SizesAcc); |
| #leaf{} -> |
| add_sizes(leaf, Value, SizesAcc) |
| end; |
| (_Rev, _Value, branch, SizesAcc) -> |
| SizesAcc |
| end, |
| {_, FinalES, FinalAtts} = couch_key_tree:fold(Fun, {0, 0, []}, RevTree), |
| TotalAttSize = lists:foldl(fun({_, S}, A) -> S + A end, 0, FinalAtts), |
| FinalES + TotalAttSize. |
| |
| purge_docs(Db, []) -> |
| {ok, Db, []}; |
| |
| purge_docs(Db, PurgeReqs) -> |
| Ids = lists:usort(lists:map(fun({_UUID, Id, _Revs}) -> Id end, PurgeReqs)), |
| FDIs = couch_db_engine:open_docs(Db, Ids), |
| USeq = couch_db_engine:get_update_seq(Db), |
| |
| IdFDIs = lists:zip(Ids, FDIs), |
| {NewIdFDIs, Replies} = apply_purge_reqs(PurgeReqs, IdFDIs, USeq, []), |
| |
| Pairs = lists:flatmap(fun({DocId, OldFDI}) -> |
| {DocId, NewFDI} = lists:keyfind(DocId, 1, NewIdFDIs), |
| case {OldFDI, NewFDI} of |
| {not_found, not_found} -> |
| []; |
| {#full_doc_info{} = A, #full_doc_info{} = A} -> |
| []; |
| {#full_doc_info{}, _} -> |
| [{OldFDI, NewFDI}] |
| end |
| end, IdFDIs), |
| |
| PSeq = couch_db_engine:get_purge_seq(Db), |
| {RevPInfos, _} = lists:foldl(fun({UUID, DocId, Revs}, {PIAcc, PSeqAcc}) -> |
| Info = {PSeqAcc + 1, UUID, DocId, Revs}, |
| {[Info | PIAcc], PSeqAcc + 1} |
| end, {[], PSeq}, PurgeReqs), |
| PInfos = lists:reverse(RevPInfos), |
| |
| {ok, Db1} = couch_db_engine:purge_docs(Db, Pairs, PInfos), |
| Db2 = commit_data(Db1), |
| ok = gen_server:call(couch_server, {db_updated, Db2}, infinity), |
| couch_event:notify(Db2#db.name, updated), |
| {ok, Db2, Replies}. |
| |
| |
| apply_purge_reqs([], IdFDIs, _USeq, Replies) -> |
| {IdFDIs, lists:reverse(Replies)}; |
| |
| apply_purge_reqs([Req | RestReqs], IdFDIs, USeq, Replies) -> |
| {_UUID, DocId, Revs} = Req, |
| {value, {_, FDI0}, RestIdFDIs} = lists:keytake(DocId, 1, IdFDIs), |
| {NewFDI, RemovedRevs, NewUSeq} = case FDI0 of |
| #full_doc_info{rev_tree = Tree} -> |
| case couch_key_tree:remove_leafs(Tree, Revs) of |
| {_, []} -> |
| % No change |
| {FDI0, [], USeq}; |
| {[], Removed} -> |
| % Completely purged |
| {not_found, Removed, USeq}; |
| {NewTree, Removed} -> |
| % Its possible to purge the #leaf{} that contains |
| % the update_seq where this doc sits in the |
| % update_seq sequence. Rather than do a bunch of |
| % complicated checks we just re-label every #leaf{} |
| % and reinsert it into the update_seq sequence. |
| {NewTree2, NewUpdateSeq} = couch_key_tree:mapfold(fun |
| (_RevId, Leaf, leaf, SeqAcc) -> |
| {Leaf#leaf{seq = SeqAcc + 1}, |
| SeqAcc + 1}; |
| (_RevId, Value, _Type, SeqAcc) -> |
| {Value, SeqAcc} |
| end, USeq, NewTree), |
| |
| FDI1 = FDI0#full_doc_info{ |
| update_seq = NewUpdateSeq, |
| rev_tree = NewTree2 |
| }, |
| {FDI1, Removed, NewUpdateSeq} |
| end; |
| not_found -> |
| % Not found means nothing to change |
| {not_found, [], USeq} |
| end, |
| NewIdFDIs = [{DocId, NewFDI} | RestIdFDIs], |
| NewReplies = [{ok, RemovedRevs} | Replies], |
| apply_purge_reqs(RestReqs, NewIdFDIs, NewUSeq, NewReplies). |
| |
| |
| commit_data(Db) -> |
| commit_data(Db, false). |
| |
| commit_data(#db{waiting_delayed_commit=nil} = Db, true) -> |
| TRef = erlang:send_after(1000,self(),delayed_commit), |
| Db#db{waiting_delayed_commit=TRef}; |
| commit_data(Db, true) -> |
| Db; |
| commit_data(Db, _) -> |
| #db{ |
| waiting_delayed_commit = Timer |
| } = Db, |
| if is_reference(Timer) -> erlang:cancel_timer(Timer); true -> ok end, |
| {ok, Db1} = couch_db_engine:commit_data(Db), |
| Db1#db{ |
| waiting_delayed_commit = nil, |
| committed_update_seq = couch_db_engine:get_update_seq(Db) |
| }. |
| |
| |
| pair_write_info(Old, New) -> |
| lists:map(fun(FDI) -> |
| case lists:keyfind(FDI#full_doc_info.id, #full_doc_info.id, Old) of |
| #full_doc_info{} = OldFDI -> {OldFDI, FDI}; |
| false -> {not_found, FDI} |
| end |
| end, New). |
| |
| |
| get_meta_body_size(Meta) -> |
| {ejson_size, ExternalSize} = lists:keyfind(ejson_size, 1, Meta), |
| ExternalSize. |
| |
| |
| default_security_object(<<"shards/", _/binary>>) -> |
| case config:get("couchdb", "default_security", "everyone") of |
| "admin_only" -> |
| [{<<"members">>,{[{<<"roles">>,[<<"_admin">>]}]}}, |
| {<<"admins">>,{[{<<"roles">>,[<<"_admin">>]}]}}]; |
| Everyone when Everyone == "everyone"; Everyone == "admin_local" -> |
| [] |
| end; |
| default_security_object(_DbName) -> |
| case config:get("couchdb", "default_security", "everyone") of |
| Admin when Admin == "admin_only"; Admin == "admin_local" -> |
| [{<<"members">>,{[{<<"roles">>,[<<"_admin">>]}]}}, |
| {<<"admins">>,{[{<<"roles">>,[<<"_admin">>]}]}}]; |
| "everyone" -> |
| [] |
| end. |
| |
| % These functions rely on using the process dictionary. This is |
| % usually frowned upon however in this case it is done to avoid |
| % changing to a different server state record. Once PSE (Pluggable |
| % Storage Engine) code lands this should be moved to the #db{} record. |
| update_idle_limit_from_config() -> |
| Default = integer_to_list(?IDLE_LIMIT_DEFAULT), |
| IdleLimit = case config:get("couchdb", "idle_check_timeout", Default) of |
| "infinity" -> |
| infinity; |
| Milliseconds -> |
| list_to_integer(Milliseconds) |
| end, |
| put(idle_limit, IdleLimit), |
| IdleLimit. |
| |
| idle_limit() -> |
| get(idle_limit). |
| |
| hibernate_if_no_idle_limit() -> |
| case idle_limit() of |
| infinity -> |
| hibernate; |
| Timeout when is_integer(Timeout) -> |
| Timeout |
| end. |
| |
| |
| -ifdef(TEST). |
| -include_lib("eunit/include/eunit.hrl"). |
| |
| |
| update_local_doc_revs_test_() -> |
| {inparallel, [ |
| {"Test local doc with valid rev", fun t_good_local_doc/0}, |
| {"Test local doc with invalid rev", fun t_bad_local_doc/0}, |
| {"Test deleted local doc", fun t_dead_local_doc/0} |
| ]}. |
| |
| |
| t_good_local_doc() -> |
| Doc = #doc{ |
| id = <<"_local/alice">>, |
| revs = {0, [<<"1">>]}, |
| meta = [{ref, make_ref()}] |
| }, |
| [NewDoc] = update_local_doc_revs([{self(), Doc}]), |
| ?assertEqual({0, [2]}, NewDoc#doc.revs), |
| {ok, Result} = receive_result(Doc), |
| ?assertEqual({ok,{0,<<"2">>}}, Result). |
| |
| |
| t_bad_local_doc() -> |
| lists:foreach(fun(BadRevs) -> |
| Doc = #doc{ |
| id = <<"_local/alice">>, |
| revs = BadRevs, |
| meta = [{ref, make_ref()}] |
| }, |
| NewDocs = update_local_doc_revs([{self(), Doc}]), |
| ?assertEqual([], NewDocs), |
| {ok, Result} = receive_result(Doc), |
| ?assertEqual({error,<<"Invalid rev format">>}, Result) |
| end, [{0, [<<"a">>]}, {1, [<<"1">>]}]). |
| |
| |
| |
| t_dead_local_doc() -> |
| Doc = #doc{ |
| id = <<"_local/alice">>, |
| revs = {0, [<<"122">>]}, |
| deleted = true, |
| meta = [{ref, make_ref()}] |
| }, |
| [NewDoc] = update_local_doc_revs([{self(), Doc}]), |
| ?assertEqual({0, [0]}, NewDoc#doc.revs), |
| {ok, Result} = receive_result(Doc), |
| ?assertEqual({ok,{0,<<"0">>}}, Result). |
| |
| |
| receive_result(#doc{meta = Meta}) -> |
| Ref = couch_util:get_value(ref, Meta), |
| receive |
| {result, _, {Ref, Result}} -> {ok, Result} |
| end. |
| |
| -endif. |
