mango: prevent occasional duplication of paginated text results

When an interleaved update happens to a `text` index while it is
being queried, the affected documents might appear duplicated in
the collected results.  That is because `search` results are
paginated where the boundaries might move due to the updates and
the document might show up once more on subsequent queries.

Mitigate the situation by tracking and losing duplicated documents
while the results are being streamed to the user.

Author: pgj

src/mango/src/mango_cursor_text.erl

@@ -37,7 +37,8 @@
     user_fun,
     user_acc,
     fields,
-    execution_stats
+    execution_stats,
+    documents_seen
 }).
 
 -spec create(Db, {Indexes, Trace}, Selector, Options) -> {ok, #cursor{}} | no_return() when
@@ -118,7 +119,8 @@ execute(Cursor, UserFun, UserAcc) ->
         user_fun = UserFun,
         user_acc = UserAcc,
         fields = Cursor#cursor.fields,
-        execution_stats = mango_execution_stats:log_start(Stats)
+        execution_stats = mango_execution_stats:log_start(Stats),
+        documents_seen = sets:new([{version, 2}])
     },
     try
         case Query of
@@ -197,26 +199,40 @@ handle_hit(CAcc0, Sort, Doc) ->
     #cacc{
         limit = Limit,
         skip = Skip,
-        execution_stats = Stats
+        execution_stats = Stats,
+        documents_seen = Seen
     } = CAcc0,
     CAcc1 = update_bookmark(CAcc0, Sort),
     Stats1 = mango_execution_stats:incr_docs_examined(Stats),
     couch_stats:increment_counter([mango, docs_examined]),
     CAcc2 = CAcc1#cacc{execution_stats = Stats1},
     case mango_selector:match(CAcc2#cacc.selector, Doc) of
-        true when Skip > 0 ->
-            CAcc2#cacc{skip = Skip - 1};
-        true when Limit == 0 ->
-            % We hit this case if the user spcified with a
-            % zero limit. Notice that in this case we need
-            % to return the bookmark from before this match
-            throw({stop, CAcc0});
-        true when Limit == 1 ->
-            NewCAcc = apply_user_fun(CAcc2, Doc),
-            throw({stop, NewCAcc});
-        true when Limit > 1 ->
-            NewCAcc = apply_user_fun(CAcc2, Doc),
-            NewCAcc#cacc{limit = Limit - 1};
+        true ->
+            DocId = mango_fields:extract(Doc, [<<"_id">>]),
+            case sets:is_element(DocId, Seen) of
+                true ->
+                    CAcc2;
+                false ->
+                    CAcc3 = CAcc2#cacc{
+                        documents_seen = sets:add_element(DocId, Seen)
+                    },
+                    if
+                        Skip > 0 ->
+                            CAcc3#cacc{skip = Skip - 1};
+                        Limit == 0 ->
+                            % We hit this case if the user specified
+                            % with a zero limit. Notice that in this
+                            % case we need to return the bookmark from
+                            % before this match.
+                            throw({stop, CAcc0});
+                        Limit == 1 ->
+                            CAcc4 = apply_user_fun(CAcc3, Doc),
+                            throw({stop, CAcc4});
+                        Limit > 1 ->
+                            CAcc4 = apply_user_fun(CAcc3, Doc),
+                            CAcc4#cacc{limit = Limit - 1}
+                    end
+            end;
         false ->
             CAcc2
     end.
@@ -522,7 +538,16 @@ execute_test_() ->
                 append_sort_type,
                 fun(RawField, selector) -> <<RawField/binary, "<type>">> end
             ),
-            meck:expect(mango_fields, extract, fun({doc, N}, fields) -> {final_doc, N} end),
+            meck:expect(
+                mango_fields,
+                extract,
+                fun({doc, N}, Fields) ->
+                    case Fields of
+                        fields -> {final_doc, N};
+                        [<<"_id">>] -> N
+                    end
+                end
+            ),
             meck:expect(
                 foo,
                 add_key_only,
@@ -549,8 +574,8 @@ execute_test_() ->
             ?TDEF_FE(t_execute_more_results, 10),
             ?TDEF_FE(t_execute_unique_results, 10),
             ?TDEF_FE(t_execute_limit_cutoff, 10),
+            ?TDEF_FE(t_execute_limit_cutoff_unique, 10),
             ?TDEF_FE(t_execute_limit_zero, 10),
-            ?TDEF_FE(t_execute_limit_unique, 10),
             ?TDEF_FE(t_execute_skip, 10),
             ?TDEF_FE(t_execute_skip_unique, 10),
             ?TDEF_FE(t_execute_no_matches, 10),
@@ -753,54 +778,12 @@ t_execute_limit_cutoff(_) ->
     meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
     meck:expect(mango_selector, match, fun(selector, {doc, _N}) -> true end),
     ?assertEqual({ok, {acc, 5}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
-    ?assertEqual(2, meck:num_calls(couch_stats, increment_counter, 1)),
-    ?assertEqual(2, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
-    ?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
-
-t_execute_limit_zero(_) ->
-    Limit = 0,
-    Skip = 0,
-    IdxName = <<"index">>,
-    Options = [{partition, <<>>}, {sort, {[]}}, {bookmark, [bookmark, 0]}],
-    Cursor = #cursor{
-        db = db,
-        index = #idx{ddoc = <<"_design/ddoc">>, name = IdxName},
-        limit = Limit,
-        skip = Skip,
-        fields = fields,
-        selector = selector,
-        opts = Options,
-        execution_stats = stats
-    },
-    QueryArgs =
-        #index_query_args{
-            q = query,
-            partition = nil,
-            limit = Skip + Limit,
-            bookmark = [bookmark, 0],
-            sort = relevance,
-            raw_bookmark = true
-        },
-    Hit1 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 1}}]}},
-    Hit2 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 2}}]}},
-    Hit3 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 3}}]}},
-    Hits = [Hit1, Hit2, Hit3],
-    meck:expect(
-        dreyfus_fabric_search,
-        go,
-        [db_name, <<"ddoc">>, IdxName, QueryArgs],
-        meck:val({ok, [bookmark, 1], undefined, Hits, undefined, undefined})
-    ),
-    meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
-    meck:expect(mango_selector, match, fun(selector, {doc, _N}) -> true end),
-    ?assertEqual({ok, {acc, 3}}, execute(Cursor, fun foo:add_key_only/2, {acc, 0})),
-    ?assertEqual(1, meck:num_calls(couch_stats, increment_counter, 1)),
-    ?assertEqual(1, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
+    ?assertEqual(Limit, meck:num_calls(couch_stats, increment_counter, 1)),
+    ?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
     ?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
 
-t_execute_limit_unique(_) ->
-    Limit = 5,
-    UniqueHits = 3,
+t_execute_limit_cutoff_unique(_) ->
+    Limit = 2,
     Options = [{partition, partition}, {sort, {[]}}, {bookmark, []}],
     Cursor = #cursor{
         db = db,
@@ -835,17 +818,60 @@ t_execute_limit_unique(_) ->
                         Hit1 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 1}}]}},
                         Hit2 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 2}}]}},
                         Hit3 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 3}}]}},
-                        {[bookmark, 4], [Hit3, Hit2, Hit1]}
+                        {[bookmark, 4], [Hit3, Hit2, Hit1]};
+                    [bookmark, 7] ->
+                        {[bookmark, 8], []}
                 end,
             {ok, Bookmark, undefined, Hits, undefined, undefined}
         end
     ),
     meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
     meck:expect(mango_selector, match, fun(selector, {doc, _N}) -> true end),
-    ?assertEqual({ok, {acc, 6}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
+    ?assertEqual({ok, {acc, 5}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
     ?assertEqual(Limit, meck:num_calls(couch_stats, increment_counter, 1)),
     ?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
-    ?assertEqual(UniqueHits, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
+    ?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
+
+t_execute_limit_zero(_) ->
+    Limit = 0,
+    Skip = 0,
+    IdxName = <<"index">>,
+    Options = [{partition, <<>>}, {sort, {[]}}, {bookmark, [bookmark, 0]}],
+    Cursor = #cursor{
+        db = db,
+        index = #idx{ddoc = <<"_design/ddoc">>, name = IdxName},
+        limit = Limit,
+        skip = Skip,
+        fields = fields,
+        selector = selector,
+        opts = Options,
+        execution_stats = stats
+    },
+    QueryArgs =
+        #index_query_args{
+            q = query,
+            partition = nil,
+            limit = Skip + Limit,
+            bookmark = [bookmark, 0],
+            sort = relevance,
+            raw_bookmark = true
+        },
+    Hit1 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 1}}]}},
+    Hit2 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 2}}]}},
+    Hit3 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 3}}]}},
+    Hits = [Hit1, Hit2, Hit3],
+    meck:expect(
+        dreyfus_fabric_search,
+        go,
+        [db_name, <<"ddoc">>, IdxName, QueryArgs],
+        meck:val({ok, [bookmark, 1], undefined, Hits, undefined, undefined})
+    ),
+    meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
+    meck:expect(mango_selector, match, fun(selector, {doc, _N}) -> true end),
+    ?assertEqual({ok, {acc, 3}}, execute(Cursor, fun foo:add_key_only/2, {acc, 0})),
+    ?assertEqual(1, meck:num_calls(couch_stats, increment_counter, 1)),
+    ?assertEqual(1, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
+    ?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
 
 t_execute_skip(_) ->
     UniqueHits = 3,
@@ -898,12 +924,13 @@ t_execute_skip(_) ->
 t_execute_skip_unique(_) ->
     UniqueHits = 3,
     AllHits = 6,
+    Skip = 2,
     Options = [{partition, partition}, {sort, {[]}}, {bookmark, []}],
     Cursor = #cursor{
         db = db,
         index = #idx{ddoc = <<"ddoc">>, name = <<"index">>},
         limit = 10,
-        skip = 2,
+        skip = Skip,
         fields = fields,
         selector = selector,
         opts = Options,
@@ -941,10 +968,12 @@ t_execute_skip_unique(_) ->
     ),
     meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
     meck:expect(mango_selector, match, fun(selector, {doc, _N}) -> true end),
-    ?assertEqual({ok, {acc, 6}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
+    ?assertEqual({ok, {acc, 4}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
     ?assertEqual(AllHits, meck:num_calls(couch_stats, increment_counter, 1)),
     ?assertEqual(AllHits, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
-    ?assertEqual(UniqueHits, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
+    ?assertEqual(
+        UniqueHits - Skip, meck:num_calls(mango_execution_stats, incr_results_returned, 1)
+    ).
 
 t_execute_no_matches(_) ->
     UniqueHits = 3,

src/mango/test/24-text-paginated-test.py

@@ -0,0 +1,108 @@
+# 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.
+
+
+import mango
+import unittest
+import time
+
+
+@unittest.skipUnless(mango.has_text_service(), "requires text service")
+class PaginatedResultsTest(mango.DbPerClass):
+    # Great enough to make faster systems busy while running the
+    # query.
+    NUM_DOCS = 10000
+    UPDATES = 25
+
+    def setUp(self):
+        self.db.recreate()
+        self.db.create_text_index(
+            analyzer="keyword",
+            default_field={},
+            selector={},
+            fields=[
+                {"name": "_id", "type": "string"},
+                {"name": "name", "type": "string"},
+            ],
+            index_array_lengths=True,
+        )
+        docs = [
+            {"_id": f"{doc_id:08X}", "name": mango.random_string(32)}
+            for doc_id in range(self.NUM_DOCS)
+        ]
+        self.db.save_docs(docs)
+
+    def test_query_with_lot_of_results(self):
+        # 200 is the maximum for `text` searches.
+        docs = self.db.find(selector={"_id": {"$lte": f"{1000:08X}"}}, limit=200)
+        assert len(docs) == 200
+
+    def do_query(self, delay, find_args):
+        time.sleep(delay)
+        return self.db.find(*find_args)
+
+    def do_updates(self, pause, doc_id):
+        for i in range(self.UPDATES):
+            doc = self.db.open_doc(doc_id)
+            updated_doc = {
+                "_id": doc_id,
+                "_rev": doc["_rev"],
+                "update": i,
+                "name": "foobar",
+            }
+            self.db.save_doc(updated_doc)
+            time.sleep(pause)
+
+    def test_no_duplicates_on_interleaved_updates(self):
+        # Give ~500 ms head start for the updates before running the
+        # query.
+        query = mango.Concurrently(self.do_query, (0.5, ({"name": "foobar"},)))
+        # Keep updating the target document in every 200 ms.
+        updates = mango.Concurrently(self.do_updates, (0.2, f"{2:08X}"))
+        docs = query.get_result()
+        updates.join()
+        assert len(docs) == 1
+
+    def test_no_duplicates_on_interleaved_updates_heavy(self):
+        query = mango.Concurrently(self.do_query, (0.5, ({"name": "foobar"},)))
+        updates = [
+            mango.Concurrently(self.do_updates, (0.05, f"{2:08X}")),
+            mango.Concurrently(self.do_updates, (0.2, f"{3:08X}")),
+            mango.Concurrently(self.do_updates, (0.3, f"{4:08X}")),
+            mango.Concurrently(self.do_updates, (0.15, f"{5:08X}")),
+            mango.Concurrently(self.do_updates, (0.1, f"{6:08X}")),
+        ]
+        docs = query.get_result()
+        for ref in updates:
+            ref.join()
+        ids = list(map(lambda d: d["_id"], docs))
+        assert sorted(ids) == [
+            f"{2:08X}",
+            f"{3:08X}",
+            f"{4:08X}",
+            f"{5:08X}",
+            f"{6:08X}",
+        ]
+
+    def test_no_duplicates_on_interleaved_updates_with_limit_skip(self):
+        query = mango.Concurrently(self.do_query, (0.5, ({"name": "foobar"}, 1, 3)))
+        updates = [
+            mango.Concurrently(self.do_updates, (0.05, f"{2:08X}")),
+            mango.Concurrently(self.do_updates, (0.2, f"{3:08X}")),
+            mango.Concurrently(self.do_updates, (0.3, f"{4:08X}")),
+            mango.Concurrently(self.do_updates, (0.15, f"{5:08X}")),
+            mango.Concurrently(self.do_updates, (0.1, f"{6:08X}")),
+        ]
+        docs = query.get_result()
+        for ref in updates:
+            ref.join()
+        assert len(docs) == 1