This is an automated email from the ASF dual-hosted git repository. rnewson pushed a commit to branch replicator-quorum-ops in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit f58ad1b9e07fe24bbe389975a5112624d86fbab5 Author: Robert Newson <[email protected]> AuthorDate: Wed Jun 10 09:04:42 2026 +0100 use fabric to read/write _replicator docs closes https://github.com/apache/couchdb/issues/6029 --- src/couch_replicator/src/couch_replicator.app.src | 1 + src/couch_replicator/src/couch_replicator_docs.erl | 47 ++++++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/src/couch_replicator/src/couch_replicator.app.src b/src/couch_replicator/src/couch_replicator.app.src index 4f68c5664..8739ea6de 100644 --- a/src/couch_replicator/src/couch_replicator.app.src +++ b/src/couch_replicator/src/couch_replicator.app.src @@ -26,6 +26,7 @@ kernel, stdlib, couch_log, + fabric, mem3, config, couch, diff --git a/src/couch_replicator/src/couch_replicator_docs.erl b/src/couch_replicator/src/couch_replicator_docs.erl index 8f134aaab..b4a7781e1 100644 --- a/src/couch_replicator/src/couch_replicator_docs.erl +++ b/src/couch_replicator/src/couch_replicator_docs.erl @@ -196,6 +196,12 @@ update_rep_doc(RepDbName, #doc{body = {RepDocBody}} = RepDoc, KVs, _Try) -> save_rep_doc(RepDbName, RepDoc#doc{body = {NewRepDocBody}}) end. +open_rep_doc(<<"shards/", _/binary>>=ShardDbName, DocId) -> + DbName = mem3:dbname(ShardDbName), + ioq:maybe_set_io_priority({system, DbName}), + defer_call(fun() -> + fabric:open_doc(DbName, DocId, [?CTX]) + end); open_rep_doc(DbName, DocId) -> ioq:maybe_set_io_priority({system, DbName}), case couch_db:open_int(DbName, [?CTX, sys_db]) of @@ -209,6 +215,22 @@ open_rep_doc(DbName, DocId) -> Else end. +save_rep_doc(<<"shards/", _/binary>>=ShardDbName, Doc) -> + DbName = mem3:dbname(ShardDbName), + ioq:maybe_set_io_priority({system, DbName}), + defer_call(fun() -> + try + fabric:update_doc(DbName, Doc, [?CTX]) + catch + % User can accidentally write a VDU which prevents _replicator from + % updating replication documents. Avoid crashing replicator and thus + % preventing all other replication jobs on the node from running. + throw:{forbidden, Reason} -> + Msg = "~p VDU function preventing doc update to ~s ~s ~p", + couch_log:error(Msg, [?MODULE, DbName, Doc#doc.id, Reason]), + {ok, forbidden} + end + end); save_rep_doc(DbName, Doc) -> ioq:maybe_set_io_priority({system, DbName}), {ok, Db} = couch_db:open_int(DbName, [?CTX, sys_db]), @@ -226,6 +248,31 @@ save_rep_doc(DbName, Doc) -> couch_db:close(Db) end. +defer_call(Fun) -> + {Pid, Ref} = spawn_monitor(fun() -> + try Fun() of + Resp -> + exit({exit_ok, Resp}) + catch + throw:Reason -> + exit({exit_throw, Reason}); + error:Reason -> + exit({exit_error, Reason}); + exit:Reason -> + exit({exit_exit, Reason}) + end + end), + receive + {'DOWN', Ref, process, Pid, {exit_ok, Ret}} -> + Ret; + {'DOWN', Ref, process, Pid, {exit_throw, Reason}} -> + throw(Reason); + {'DOWN', Ref, process_, Pid, {exit_error, Reason}} -> + error(Reason); + {'DOWN', Ref, process, Pid, {exit_exit, Reason}} -> + exit(Reason) + end. + -spec before_doc_update(#doc{}, Db :: any(), couch_db:update_type()) -> #doc{}. before_doc_update(#doc{id = <<?DESIGN_DOC_PREFIX, _/binary>>} = Doc, _Db, _UpdateType) -> Doc;
