You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2026/09/15 17:49:31 UTC
(couchdb) 01/01: Optimize and improve change sequence decoding
This is an automated email from the ASF dual-hosted git repository.
nickva pushed a commit to branch improve-sequences-and-bookmarks
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 702f2fc6db3ea434d04d2eba352de861031c188b
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Sat Sep 12 05:08:09 2026 -0400
Optimize and improve change sequence decoding
Previously the sequence regex checker `"^\"?([0-9]+-)?(?<opaque>.*?)\"?$"` was
2-3 times slower than a simple matching and string module calls. A Q64 sequence
used to take 120usec to decode, and the new code takes 50usec. Q8 used to take
25usec, and now it takes 10usec (there is some time spent just setting up the
re:run/1).
Also, while at it add some extra validation when decoding: range types, use the
recent `[safe]` flag. To ensure replaced nodes keep getting accepted, ensure to
intern all the deleted `_nodes` doc IDs at node startup. Since we fixed it here
also apply the same improvement to dreyfus bookmarks.
---
src/dreyfus/src/dreyfus_bookmark.erl | 18 ++-
src/dreyfus/test/eunit/dreyfus_bookmark_test.erl | 89 ++++++++++++
src/fabric/src/fabric_view_changes.erl | 175 +++++++++++++++++++++--
src/mem3/include/mem3.hrl | 2 +-
src/mem3/src/mem3_nodes.erl | 68 ++++++++-
5 files changed, 334 insertions(+), 18 deletions(-)
diff --git a/src/dreyfus/src/dreyfus_bookmark.erl b/src/dreyfus/src/dreyfus_bookmark.erl
index e33087a27..268a6b845 100644
--- a/src/dreyfus/src/dreyfus_bookmark.erl
+++ b/src/dreyfus/src/dreyfus_bookmark.erl
@@ -63,9 +63,25 @@ unpack(DbName, Packed) when is_binary(Packed) ->
{PlaceHolder, After}
end
end,
- binary_to_term(couch_util:decodeBase64Url(Packed))
+ unpack_entries(Packed)
).
+unpack_entries(Packed) ->
+ case binary_to_term(couch_util:decodeBase64Url(Packed), [safe]) of
+ Entries when is_list(Entries) ->
+ lists:foreach(fun validate_entry/1, Entries),
+ Entries;
+ _ ->
+ error(badarg)
+ end.
+
+validate_entry({Node, [B, E], _After}) when
+ is_atom(Node), is_integer(B), is_integer(E), B >= 0, B =< E, E =< ?RING_END
+->
+ ok;
+validate_entry(_) ->
+ error(badarg).
+
pack(nil) ->
null;
pack(Workers) ->
diff --git a/src/dreyfus/test/eunit/dreyfus_bookmark_test.erl b/src/dreyfus/test/eunit/dreyfus_bookmark_test.erl
new file mode 100644
index 000000000..ca5445162
--- /dev/null
+++ b/src/dreyfus/test/eunit/dreyfus_bookmark_test.erl
@@ -0,0 +1,89 @@
+% 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(dreyfus_bookmark_test).
+
+-include_lib("mem3/include/mem3.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+-define(DBNAME, <<"db">>).
+
+setup() ->
+ meck:new(mem3),
+ % One ([0,10]) is a known shard only
+ meck:expect(mem3, get_shard, fun
+ (?DBNAME, n1, [0, 10]) -> {ok, known_shard()};
+ (_, _, _) -> {error, not_found}
+ end).
+
+teardown(_) ->
+ meck:unload().
+
+known_shard() ->
+ #shard{
+ name = <<"shards/00000000-0000000a/db.1">>,
+ node = n1,
+ range = [0, 10],
+ dbname = ?DBNAME
+ }.
+
+placeholder(Node, Range) ->
+ #shard{node = Node, range = Range, dbname = ?DBNAME, _ = '_'}.
+
+dreyfus_bookmark_test_() ->
+ {
+ setup,
+ fun setup/0,
+ fun teardown/1,
+ [
+ fun t_roundtrip/0,
+ fun t_rejects_unknown_atoms/0,
+ fun t_rejects_malformed/0
+ ]
+ }.
+
+t_roundtrip() ->
+ Workers = [
+ {known_shard(), {1.5, <<"doc1">>}},
+ {#shard{node = n2, range = [11, ?RING_END]}, [<<"x">>, null]},
+ {#shard{node = n3, range = [11, ?RING_END]}, nil}
+ ],
+ Packed = dreyfus_bookmark:pack(Workers),
+ ?assertEqual(
+ [
+ {known_shard(), {1.5, <<"doc1">>}},
+ {placeholder(n2, [11, ?RING_END]), [<<"x">>, null]}
+ ],
+ dreyfus_bookmark:unpack(?DBNAME, Packed)
+ ).
+
+t_rejects_unknown_atoms() ->
+ %> couch_util:encodeBase64Url(term_to_binary([{'otherclust@baz', [0, 10], 42}])).
+ %> <<"g2wAAAABaAN3Dm90aGVyY2x1c3RAYmF6awACAAphKmo">>
+ Unknown = <<"g2wAAAABaAN3Dm90aGVyY2x1c3RAYmF6awACAAphKmo">>,
+ %> couch_util:encodeBase64Url(term_to_binary([{n1, [0, 10], 42}])).
+ %> <<"g2wAAAABaAN3Am4xawACAAphKmo">>
+ Known = <<"g2wAAAABaAN3Am4xawACAAphKmo">>,
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, Unknown)),
+ ?assertError(badarg, binary_to_existing_atom(<<"otherclust@baz">>, utf8)),
+ ?assertEqual([{known_shard(), 42}], dreyfus_bookmark:unpack(?DBNAME, Known)).
+
+t_rejects_malformed() ->
+ T = fun(Term) -> couch_util:encodeBase64Url(term_to_binary(Term)) end,
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, T({n1, [0, 10], 42}))),
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, T([{n1, [0, 10]}]))),
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, T([{n1, {0, 10}, 42}]))),
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, T([{n1, [0, 1 bsl 32], 42}]))),
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, T([{n1, [10, 0], 42}]))),
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, T([{<<"n1">>, [0, 10], 42}]))),
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, <<"not base64url junk!">>)),
+ ?assertError(badarg, dreyfus_bookmark:unpack(?DBNAME, couch_util:encodeBase64Url(<<2, 0>>))).
diff --git a/src/fabric/src/fabric_view_changes.erl b/src/fabric/src/fabric_view_changes.erl
index dd2386f5e..f02e787c7 100644
--- a/src/fabric/src/fabric_view_changes.erl
+++ b/src/fabric/src/fabric_view_changes.erl
@@ -29,6 +29,16 @@
-define(RFC3339_TIME, [_, _, _, _, $-, _, _, $-, _, _, $T, _, _, $:, _, _, $:, _, _, $Z]).
+-define(IN_RANGE(X, Min, Max), (is_integer(X) andalso X >= Min andalso X =< Max)).
+-define(IS_BOUND(X), ?IN_RANGE(X, 0, ?RING_END)).
+-define(IS_SEQ_NUM(X), ?IN_RANGE(X, 0, (1 bsl 64) - 1)).
+-define(IS_UUID(X), (is_binary(X) andalso ?IN_RANGE(byte_size(X), 1, 255))).
+% Bad sequence error to indicate the passed in sequence cannot be decoded. A
+% common reason is it references node names unknown on this node. This error
+% unlike a malformed base64 string will result in a rewind. That's done for
+% compatibilty reasons and to make replication self-healing.
+-define(BADSEQ, badseq).
+
go(DbName, Feed, Options, Callback, Acc0) when
Feed == "continuous" orelse
Feed == "longpoll" orelse Feed == "eventsource"
@@ -416,14 +426,84 @@ fake_packed_seq({Seq, _Uuid, _Node}) -> Seq;
fake_packed_seq({Seq, _Uuid}) -> Seq;
fake_packed_seq(Seq) -> Seq.
-unpack_seq_regex_match(Packed) ->
- Pattern = "^\"?([0-9]+-)?(?<opaque>.*?)\"?$",
- Options = [{capture, [opaque], binary}],
- {match, Match} = re:run(Packed, Pattern, Options),
- Match.
+% Strip quotes and the N- prefix
+unpack_seq_opaque(Packed) ->
+ Bin = iolist_to_binary(Packed),
+ strip_sum_prefix(strip_quote_suffix(strip_quote_prefix(Bin))).
+
+strip_quote_prefix(<<$", Rest/binary>>) ->
+ Rest;
+strip_quote_prefix(Bin) ->
+ Bin.
+
+strip_quote_suffix(Bin) when byte_size(Bin) >= 1 ->
+ case binary:last(Bin) of
+ $" -> binary:part(Bin, 0, byte_size(Bin) - 1);
+ _ -> Bin
+ end;
+strip_quote_suffix(Bin) ->
+ Bin.
+
+strip_sum_prefix(Bin) ->
+ case binary:split(Bin, <<"-">>) of
+ [Digits, Rest] ->
+ case all_digits(Digits) of
+ true -> Rest;
+ false -> Bin
+ end;
+ [Bin] ->
+ Bin
+ end.
+
+all_digits(<<>>) ->
+ false;
+all_digits(Bin) ->
+ all_digits1(Bin).
+
+all_digits1(<<D, Rest/binary>>) when D >= $0, D =< $9 ->
+ all_digits1(Rest);
+all_digits1(<<_, _/binary>>) ->
+ false;
+all_digits1(<<>>) ->
+ true.
+
+unpack_seq_term(Opaque) ->
+ case couch_util:decodeBase64Url(Opaque) of
+ <<131, _/binary>> = Bin ->
+ % Looks like a valid term_to_binary at least
+ SeqList =
+ try
+ binary_to_term(Bin, [safe])
+ catch
+ error:badarg -> error(?BADSEQ)
+ end,
+ validate_seq_list(SeqList);
+ _ ->
+ % Not even a valid term_to_binary result. Throw a badarg which
+ % becomes a 4xx error at the HTTP API level.
+ error(badarg)
+ end.
+
+validate_seq_list(SeqList) when is_list(SeqList) ->
+ lists:foreach(fun validate_seq_item/1, SeqList),
+ SeqList;
+validate_seq_list(_) ->
+ error(badarg).
+
+validate_seq_item({Node, [B, E], Val}) when is_atom(Node), ?IS_BOUND(B), ?IS_BOUND(E) ->
+ case valid_seq_val(Val) of
+ true -> ok;
+ false -> error(badarg)
+ end;
+validate_seq_item(_) ->
+ error(badarg).
-unpack_seq_decode_term(Opaque) ->
- binary_to_term(couch_util:decodeBase64Url(Opaque)).
+valid_seq_val({Seq, {split, Uuid}, Epoch}) ->
+ ?IS_SEQ_NUM(Seq) andalso ?IS_UUID(Uuid) andalso is_atom(Epoch);
+valid_seq_val({Seq, Uuid, Epoch}) ->
+ ?IS_SEQ_NUM(Seq) andalso ?IS_UUID(Uuid) andalso is_atom(Epoch);
+valid_seq_val(Seq) ->
+ ?IS_SEQ_NUM(Seq).
% This is used for testing and for remsh debugging
%
@@ -433,8 +513,8 @@ unpack_seq_decode_term(Opaque) ->
%
-spec decode_seq(binary()) -> [tuple()].
decode_seq(Packed) ->
- Opaque = unpack_seq_regex_match(Packed),
- unpack_seq_decode_term(Opaque).
+ Opaque = unpack_seq_opaque(Packed),
+ unpack_seq_term(Opaque).
% Returns fabric_dict with {Shard, Seq} entries
%
@@ -445,14 +525,20 @@ unpack_seqs(0, DbName) ->
unpack_seqs("0", DbName) ->
fabric_dict:init(mem3:shards(DbName), 0);
unpack_seqs(Packed, DbName) ->
- Opaque = unpack_seq_regex_match(Packed),
- do_unpack_seqs(Opaque, DbName).
+ Opaque = unpack_seq_opaque(Packed),
+ try
+ do_unpack_seqs(Opaque, DbName)
+ catch
+ error:?BADSEQ ->
+ couch_log:warning("~p cannot decode since seq db:~s, rewinding", [?MODULE, DbName]),
+ fabric_dict:init(mem3:shards(DbName), 0)
+ end.
do_unpack_seqs(Opaque, DbName) ->
% A preventative fix for FB 13533 to remove duplicate shards.
% This just picks each unique shard and keeps the largest seq
% value recorded.
- Decoded = unpack_seq_decode_term(Opaque),
+ Decoded = unpack_seq_term(Opaque),
DedupDict = lists:foldl(
fun({Node, [A, B], Seq}, Acc) ->
dict:append({Node, [A, B]}, Seq, Acc)
@@ -741,10 +827,13 @@ validate_start_seq("0") ->
ok;
validate_start_seq(Seq) when is_list(Seq) orelse is_binary(Seq) ->
try
- Opaque = unpack_seq_regex_match(Seq),
- unpack_seq_decode_term(Opaque),
+ Opaque = unpack_seq_opaque(Seq),
+ unpack_seq_term(Opaque),
ok
catch
+ error:?BADSEQ ->
+ % Rewound by unpack_seqs/2
+ ok;
_:_ ->
Reason = <<"Malformed sequence supplied in 'since' parameter.">>,
{error, {bad_request, Reason}}
@@ -773,7 +862,11 @@ unpack_seq_setup() ->
meck:new(fabric_view),
meck:expect(mem3, get_shard, fun(_, _, _) -> {ok, #shard{}} end),
meck:expect(mem3, shards, fun(_) -> [#shard{}] end),
- meck:expect(fabric_ring, is_progress_possible, fun(_) -> true end).
+ meck:expect(fabric_ring, is_progress_possible, fun(_) -> true end),
+ meck:new(couch_log),
+ meck:expect(couch_log, warning, fun(_, _) -> ok end),
+ % Some tests use this atom and it must exist already
+ list_to_atom("dev1@127.0.0.1").
unpack_seqs_test_() ->
{
@@ -788,6 +881,7 @@ unpack_seqs_test_() ->
?TDEF(t_no_numeric_prefix),
?TDEF(t_zero_seq_int),
?TDEF(t_zero_seq_string),
+ ?TDEF(t_badseq_rewinds),
?TDEF(t_fail_now),
?TDEF(t_fail_numeric_int),
?TDEF(t_fail_numeric_string),
@@ -841,6 +935,15 @@ t_zero_seq_int(_) ->
t_zero_seq_string(_) ->
assert_shards("0").
+t_badseq_rewinds(_) ->
+ %> term_to_binary('otherclust@baz').
+ %> <<131,119,14,111,116,104,101,114,99,108,117,115,116,64,98,97,122>>
+ Ext = <<131, 119, 14, 111, 116, 104, 101, 114, 99, 108, 117, 115, 116, 64, 98, 97, 122>>,
+ Seq = <<"42-", (couch_util:encodeBase64Url(Ext))/binary>>,
+ ?assertEqual(ok, validate_start_seq(Seq)),
+ ?assertEqual(fabric_dict:init([#shard{}], 0), unpack_seqs(Seq, <<"foo">>)),
+ ?assertError(badarg, binary_to_existing_atom(<<"otherclust@baz">>, utf8)).
+
t_fail_now(_) ->
% "now" should have been transformed into a sequence in get_start_seq/2
fail("now").
@@ -1067,4 +1170,46 @@ pack_split_seq_test() ->
DecodedSeq
).
+decode_seq_unknown_atom_test() ->
+ Uniq = integer_to_binary(erlang:unique_integer([positive])),
+ Name = <<"nonexistent", Uniq/binary, "@127.0.0.1">>,
+ % Create via external term [{Atom, [0, 10], 42}] so the atom is not created
+ Ext = fun(AtomExt) ->
+ <<131, 108, 1:32, 104, 3, AtomExt/binary, 108, 2:32, 97, 0, 97, 10, 106, 97, 42, 106>>
+ end,
+ BadSeq = Ext(<<119, (byte_size(Name)):8, Name/binary>>),
+ BadB64 = couch_util:encodeBase64Url(BadSeq),
+ % Distinct from badarg: unpack_seqs/2 rewinds these instead of failing
+ ?assertError(?BADSEQ, decode_seq(<<"42-", BadB64/binary>>)),
+ % With existing atom should be good
+ Ok = Ext(<<119, 2, "n1">>),
+ OkB64 = couch_util:encodeBase64Url(Ok),
+ ?assertEqual([{n1, [0, 10], 42}], decode_seq(<<"42-", OkB64/binary>>)),
+ % We didn't create a bad atom
+ ?assertError(badarg, binary_to_existing_atom(Name, utf8)).
+
+decode_bad_shape_test() ->
+ T = fun(Term) ->
+ B64 = couch_util:encodeBase64Url(term_to_binary(Term, [{minor_version, 1}])),
+ <<"1-", B64/binary>>
+ end,
+ SeqList = [
+ {n1, [0, 10], 0},
+ {n1, [11, 20], {7, <<"u">>, n1}},
+ {n2, [21, ?RING_END], {8, {split, <<"u">>}, n1}}
+ ],
+ ?assertEqual(SeqList, decode_seq(T(SeqList))),
+ ?assertError(badarg, decode_seq(T({n1, [0, 10], 0}))),
+ ?assertError(badarg, decode_seq(T(42))),
+ ?assertError(badarg, decode_seq(T([{n1, [0, 10]}]))),
+ ?assertError(badarg, decode_seq(T([{n1, [0, 10, 20], 0}]))),
+ ?assertError(badarg, decode_seq(T([{n1, {0, 10}, 0}]))),
+ ?assertError(badarg, decode_seq(T([{n1, [0, 10], {7, <<"u">>}}]))),
+ ?assertError(badarg, decode_seq(T([{n1, [0, 10], -1}]))),
+ ?assertError(badarg, decode_seq(T([{n1, [0, 1 bsl 32], 0}]))),
+ ?assertError(badarg, decode_seq(T([{n1, [-1, 10], 0}]))),
+ ?assertError(badarg, decode_seq(T([{n1, [0, 10], {7, u, n1}}]))),
+ ?assertError(badarg, decode_seq(T([{42, [0, 10], 0}]))),
+ ?assertError(badarg, decode_seq(T([{<<"n1">>, [0, 10], 0}]))).
+
-endif.
diff --git a/src/mem3/include/mem3.hrl b/src/mem3/include/mem3.hrl
index fa232e840..12cfa0ddc 100644
--- a/src/mem3/include/mem3.hrl
+++ b/src/mem3/include/mem3.hrl
@@ -12,7 +12,7 @@
% The last element in the ring
--define(RING_END, 2 bsl 31 - 1).
+-define(RING_END, (2 bsl 31 - 1)).
% type specification hacked to suppress dialyzer warning re: match spec
diff --git a/src/mem3/src/mem3_nodes.erl b/src/mem3/src/mem3_nodes.erl
index cf895dfab..2ef4fcf41 100644
--- a/src/mem3/src/mem3_nodes.erl
+++ b/src/mem3/src/mem3_nodes.erl
@@ -97,7 +97,7 @@ handle_info(_Info, State) ->
initialize_nodelist() ->
DbName = mem3_sync:nodes_db(),
{ok, Db} = mem3_util:ensure_exists(DbName),
- {ok, _} = couch_db:fold_docs(Db, fun first_fold/2, Db, []),
+ {ok, _} = couch_db:fold_docs(Db, fun first_fold/2, Db, [include_deleted]),
insert_if_missing(Db, [config:node_name() | mem3_seeds:get_seeds()]),
@@ -120,6 +120,11 @@ initialize_nodelist() ->
first_fold(#full_doc_info{id = <<"_design/", _/binary>>}, Acc) ->
{ok, Acc};
+first_fold(#full_doc_info{deleted = true, id = Id}, Acc) when byte_size(Id) =< 255 ->
+ % Turn deleted nodes names into atoms so binary_to_term/2 with [safe] can
+ % decode sequences with those atoms.
+ mem3_util:to_atom(Id),
+ {ok, Acc};
first_fold(#full_doc_info{deleted = true}, Acc) ->
{ok, Acc};
first_fold(#full_doc_info{id = Id} = DocInfo, Db) ->
@@ -221,3 +226,64 @@ save_to_db(Db, Node, Props) ->
{ok, _} = couch_db:update_doc(Db, Doc, []),
update_ets(Node, NodeInfo),
ok.
+
+-ifdef(TEST).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+tombstones_reach_first_fold_test_() ->
+ {
+ setup,
+ fun test_util:start_couch/0,
+ fun test_util:stop_couch/1,
+ fun(_) ->
+ [fun t_tombstones_reach_first_fold/0]
+ end
+ }.
+
+t_tombstones_reach_first_fold() ->
+ DbName = ?tempdb(),
+ {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
+ try
+ Id1 = <<"tombstoned@baz">>,
+ Id2 = <<"alsodeleted@baz">>,
+ lists:foreach(
+ fun(Id) ->
+ {ok, DbA} = couch_db:reopen(Db),
+ {ok, {Pos, Rev}} = couch_db:update_doc(DbA, #doc{id = Id}, []),
+ {ok, DbB} = couch_db:reopen(DbA),
+ Del = #doc{id = Id, revs = {Pos, [Rev]}, deleted = true},
+ {ok, _} = couch_db:update_doc(DbB, Del, [])
+ end,
+ [Id1, Id2]
+ ),
+ {ok, Db1} = couch_db:reopen(Db),
+ % Without include_deleted tombstones not visited
+ {ok, _} = couch_db:fold_docs(Db1, fun first_fold/2, Db1, []),
+ ?assertError(badarg, binary_to_existing_atom(Id1, utf8)),
+ % With include_deleted they are visited
+ {ok, _} = couch_db:fold_docs(Db1, fun first_fold/2, Db1, [include_deleted]),
+ ?assert(is_atom(binary_to_existing_atom(Id1, utf8))),
+ ?assert(is_atom(binary_to_existing_atom(Id2, utf8)))
+ after
+ couch_db:close(Db),
+ couch_server:delete(DbName, [?ADMIN_CTX])
+ end.
+
+first_fold_interns_deleted_node_names_test() ->
+ Id = <<"decom@baz">>,
+ % Atom doesn't exist. Folding tombstone must create it
+ ?assertError(badarg, binary_to_existing_atom(Id, utf8)),
+ ?assertEqual({ok, acc}, first_fold(#full_doc_info{id = Id, deleted = true}, acc)),
+ ?assert(is_atom(binary_to_existing_atom(Id, utf8))),
+ % Silly huge ids skipped without crashing (otherwise we hit a system limit
+ % and can never boot a node).
+ Big = binary:copy(<<"x">>, 256),
+ ?assertEqual({ok, acc}, first_fold(#full_doc_info{id = Big, deleted = true}, acc)),
+ ?assertError(badarg, binary_to_existing_atom(Big, utf8)),
+ % Deleted design docs are also skipped by the earlier clause
+ DDocId = <<"_design/deleted@baz">>,
+ ?assertEqual({ok, acc}, first_fold(#full_doc_info{id = DDocId, deleted = true}, acc)),
+ ?assertError(badarg, binary_to_existing_atom(DDocId, utf8)).
+
+-endif.