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/10/12 19:09:45 UTC
(couchdb) branch changes-limit-last-seq created (now 6e86275d7)
This is an automated email from the ASF dual-hosted git repository.
nickva pushed a change to branch changes-limit-last-seq
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at 6e86275d7 Stop changes w/ limit from jumping over rows
This branch includes the following new commits:
new 6e86275d7 Stop changes w/ limit from jumping over rows
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(couchdb) 01/01: Stop changes w/ limit from jumping over rows
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nickva pushed a commit to branch changes-limit-last-seq
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 6e86275d7153d725d2a8b33cf70430fd573bf935
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Mon Oct 12 12:45:09 2026 -0400
Stop changes w/ limit from jumping over rows
Previously, when we reached limit=0 in the `no_pass` clause we kept advancing
the sequence even though we weren't actually sending rows back to the user any
longer (we get extra rows sent back since we have a pending ack buffer in
place). That means we could have potentially skipped over data. It's pretty
corner-case-y (limit set and a filter) but also a bit data-loss-y.
To fix it, we should do what we do for the change and complete clauses above.
Add tests for complete and and regular changes as well while at it.
---
src/fabric/src/fabric_view_changes.erl | 89 ++++++++++++++++++++++++++++++++--
1 file changed, 86 insertions(+), 3 deletions(-)
diff --git a/src/fabric/src/fabric_view_changes.erl b/src/fabric/src/fabric_view_changes.erl
index f02e787c7..e9c60a4d5 100644
--- a/src/fabric/src/fabric_view_changes.erl
+++ b/src/fabric/src/fabric_view_changes.erl
@@ -287,7 +287,11 @@ handle_message({complete, Props}, Worker, #collector{limit = 0} = State) ->
end,
maybe_stop(State#collector{offset = O1});
handle_message({no_pass, [_ | _] = Props}, {Worker, From}, #collector{limit = 0} = State) ->
- #collector{counters = S0, offset = O0} = State,
+ % We aren't sending more rows back, but keep receiving rows until we have
+ % pending updates from every worker. Just like for change and complete
+ % clauses we have to leave the worker's sequence as is. If we keep
+ % advancing then our last_seq could have undelivered changes in it.
+ #collector{offset = O0} = State,
O1 =
case fabric_dict:lookup_element(Worker, O0) of
null ->
@@ -295,9 +299,8 @@ handle_message({no_pass, [_ | _] = Props}, {Worker, From}, #collector{limit = 0}
_ ->
O0
end,
- S1 = fabric_dict:store(Worker, couch_util:get_value(seq, Props), S0),
rexi:stream_ack(From),
- maybe_stop(State#collector{counters = S1, offset = O1});
+ maybe_stop(State#collector{offset = O1});
handle_message({change, Props}, {Worker, From}, St) ->
#collector{
callback = Callback,
@@ -1212,4 +1215,84 @@ decode_bad_shape_test() ->
?assertError(badarg, decode_seq(T([{42, [0, 10], 0}]))),
?assertError(badarg, decode_seq(T([{<<"n1">>, [0, 10], 0}]))).
+limit_0_keeps_seq_test() ->
+ {A, B, #collector{counters = Seqs0} = St0} = limit_0_state(),
+ % Worker a sends us seq=11 after limit reached. We have to drop it and
+ % sequence must stay at 10.
+ {ok, St1} = handle_message({change, chprops(11, n1, 1)}, {A, self()}, St0),
+ ?assertEqual(Seqs0, St1#collector.counters),
+ ?assertEqual([{A, 2}, {B, null}], St1#collector.offset),
+ ?assertEqual(0, St1#collector.limit),
+ % Haven't heard from b yet. We wait for it to get the pending count but we
+ % must not update the sequence.
+ {stop, St2} = handle_message({change, chprops(4, n2, 5)}, {B, self()}, St1),
+ ?assertEqual(Seqs0, St2#collector.counters),
+ ?assertEqual([{A, 2}, {B, 6}], St2#collector.offset),
+ ?assertEqual(8, pending_count(St2#collector.offset)).
+
+limit_0_no_pass_keeps_seq_test() ->
+ {A, B, #collector{counters = Seqs0} = St0} = limit_0_state(),
+ % Worker a sends row seq=11. It gets dropped. Then sends a no_pass at
+ % seq=12.
+ {ok, St1} = handle_message({change, chprops(11, n1, 1)}, {A, self()}, St0),
+ {ok, St2} = handle_message({no_pass, no_pass_props(12, n1, 0)}, {A, self()}, St1),
+ % We do not advance the sequence to 12 since we never delivered 11. It
+ % could be a considered a data loss if we do and user checkpoints.
+ ?assertEqual(Seqs0, St2#collector.counters),
+ ?assertEqual([{A, 2}, {B, null}], St2#collector.offset),
+ % Worker b returns a no_pass. After this every worker has a pending count
+ % reported so we stop.
+ {stop, St3} = handle_message({no_pass, no_pass_props(4, n2, 5)}, {B, self()}, St2),
+ % This is the main assert: seq didn't change, stays at Seq0
+ ?assertEqual(Seqs0, St3#collector.counters),
+ ?assertEqual([{A, 2}, {B, 5}], St3#collector.offset),
+ ?assertEqual(7, pending_count(St3#collector.offset)),
+ % Last seq check
+ ?assertEqual(
+ [{n1, [0, 10], nseq(10, n1)}, {n2, [11, ?RING_END], nseq(3, n2)}],
+ decode_seq(pack_seqs(St3#collector.counters))
+ ).
+
+limit_0_complete_keeps_seq_test() ->
+ {A, B, #collector{counters = Seqs0} = St0} = limit_0_state(),
+ {ok, St1} = handle_message({change, chprops(11, n1, 0)}, {A, self()}, St0),
+ {ok, St2} = handle_message({complete, complete_props(11, n1, 0)}, A, St1),
+ ?assertEqual(Seqs0, St2#collector.counters),
+ ?assertEqual([{A, 2}, {B, null}], St2#collector.offset),
+ {stop, St3} = handle_message({complete, complete_props(5, n2, 0)}, B, St2),
+ ?assertEqual(Seqs0, St3#collector.counters),
+ ?assertEqual([{A, 2}, {B, 0}], St3#collector.offset),
+ ?assertEqual(2, pending_count(St3#collector.offset)).
+
+% Worker A makes limit = 0. Worker B hasn't returned yet so we don't know its
+% pending state.
+limit_0_state() ->
+ A = mk_shard("n1", 0, 10),
+ B = mk_shard("n2", 11, ?RING_END),
+ State = #collector{
+ limit = 0,
+ counters = orddict:from_list([{A, nseq(10, n1)}, {B, nseq(3, n2)}]),
+ offset = orddict:from_list([{A, 2}, {B, null}])
+ },
+ {A, B, State}.
+
+nseq(Seq, Node) ->
+ {Seq, <<"uuid">>, Node}.
+
+chprops(Seq, Node, Pending) ->
+ DocId = integer_to_binary(Seq),
+ [
+ {pending, Pending},
+ {seq, nseq(Seq, Node)},
+ {id, <<"doc", DocId/binary>>},
+ {changes, [{[{~"rev", ~"1-abc"}]}]},
+ {deleted, false}
+ ].
+
+no_pass_props(Seq, Node, Pending) ->
+ [{pending, Pending}, {seq, nseq(Seq, Node)}].
+
+complete_props(Seq, Node, Pending) ->
+ [{seq, nseq(Seq, Node)}, {pending, Pending}].
+
-endif.