Include shard uuids in db_info update sequences

This means `update_seq` values from `GET $db` `last_seq` returned from ` GET
$db/_changes?since=now&limit=` will be more resilient to change feed rewinds.
Besides, those sequences will now be more consistent and users won't have to
wonder why one opaque sequence works slightly differently than another opaque
update sequence.

Previously, when the sequences were returned only as numeric values, it was
impossible to calculate replacements and change feeds had to always rewind back
to 0 for those ranges. With uuids and epochs in play, it is possible to figure
out that some shards might have moved to new nodes or find internal replication
checkpoints to avoid streaming changes feeds from 0 on those ranges.

Some replication Elixir tests decode update sequences, so those were updated to
handle the new uuid and epoch format as well.

Fixes: https://github.com/apache/couchdb/issues/3787

Co-author: Adam Kocoloski kocolosk@apache.org

4 files changed, 97 insertions, 3 deletions

diff --git a/src/fabric/src/fabric_db_info.erl b/src/fabric/src/fabric_db_info.erl
index 40da678e5..586f282c2 100644
--- a/src/fabric/src/fabric_db_info.erl
+++ b/src/fabric/src/fabric_db_info.erl
@@ -77,7 +77,7 @@ handle_message(Reason, Shard, {Counters, Resps, CInfo}) ->
 
 build_final_response(CInfo, DbName, Responses) ->
     AccF = fabric_dict:fold(fun(Shard, Info, {Seqs, PSeqs, Infos}) ->
-        Seq = couch_util:get_value(update_seq, Info),
+        Seq = build_seq(Shard, Info),
         PSeq = couch_util:get_value(purge_seq, Info),
         {[{Shard, Seq} | Seqs], [{Shard, PSeq} | PSeqs], [Info | Infos]}
     end, {[], [], []}, Responses),
@@ -89,6 +89,13 @@ build_final_response(CInfo, DbName, Responses) ->
     [{db_name, DbName}] ++ Sequences ++ MergedInfos.
 
 
+build_seq(#shard{node = Node}, Info) when is_list(Info) ->
+    Seq = couch_util:get_value(update_seq, Info),
+    Uuid = couch_util:get_value(uuid, Info),
+    PrefixLen = fabric_util:get_uuid_prefix_len(),
+    {Seq, binary:part(Uuid, {0, PrefixLen}), Node}.
+
+
 merge_results(Info) ->
     Dict = lists:foldl(fun({K,V},D0) -> orddict:append(K,V,D0) end,
         orddict:new(), Info),
diff --git a/src/fabric/src/fabric_view_changes.erl b/src/fabric/src/fabric_view_changes.erl
index beeaecee1..eea6a72bb 100644
--- a/src/fabric/src/fabric_view_changes.erl
+++ b/src/fabric/src/fabric_view_changes.erl
@@ -18,6 +18,9 @@
 %% exported for upgrade purposes.
 -export([keep_sending_changes/8]).
 
+%% exported for testing and remsh debugging
+-export([decode_seq/1]).
+
 -include_lib("fabric/include/fabric.hrl").
 -include_lib("mem3/include/mem3.hrl").
 -include_lib("couch/include/couch_db.hrl").
@@ -410,6 +413,22 @@ unpack_seq_decode_term(Opaque) ->
     binary_to_term(couch_util:decodeBase64Url(Opaque)).
 
 
+% This is used for testing and for remsh debugging
+%
+% Return the unpacked list of sequences from a raw update seq string. The input
+% string is expected to include the N- prefix. The result looks like:
+%  [{Node, Range, {SeqNum, Uuid, EpochNode}}, ...]
+%
+-spec decode_seq(binary()) -> [tuple()].
+decode_seq(Packed) ->
+    Opaque = unpack_seq_regex_match(Packed),
+    unpack_seq_decode_term(Opaque).
+
+
+% Returns fabric_dict with {Shard, Seq} entries
+%
+-spec unpack_seqs(pos_integer() | list() | binary(), binary()) ->
+        orddict:orddict().
 unpack_seqs(0, DbName) ->
     fabric_dict:init(mem3:shards(DbName), 0);
 
diff --git a/src/fabric/test/eunit/fabric_db_info_tests.erl b/src/fabric/test/eunit/fabric_db_info_tests.erl
new file mode 100644
index 000000000..ccdafe3ae
--- /dev/null
+++ b/src/fabric/test/eunit/fabric_db_info_tests.erl
@@ -0,0 +1,68 @@
+% 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(fabric_db_info_tests).
+
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("mem3/include/mem3.hrl").
+
+
+-define(TDEF(A), {atom_to_list(A), fun A/0}).
+
+
+main_test_() ->
+    {
+        setup,
+        fun setup/0,
+        fun teardown/1,
+        [
+            ?TDEF(t_update_seq_has_uuids)
+        ]
+    }.
+
+
+setup() ->
+    test_util:start_couch([fabric]).
+
+
+teardown(Ctx) ->
+    meck:unload(),
+    test_util:stop_couch(Ctx).
+
+
+t_update_seq_has_uuids() ->
+    DbName = ?tempdb(),
+    ok = fabric:create_db(DbName, [{q, 1}, {n, 1}]),
+
+    {ok, Info} = fabric:get_db_info(DbName),
+    UpdateSeq = couch_util:get_value(update_seq, Info),
+    UnpackedSeq = fabric_view_changes:decode_seq(UpdateSeq),
+
+    ?assertMatch([{_, _, _}], UnpackedSeq),
+    [{Node, Range, Seq}] = UnpackedSeq,
+    ?assert(is_atom(Node)),
+    ?assertMatch([_, _], Range),
+    ?assertMatch({_, _, _}, Seq),
+    {SeqNum, SeqUuid, EpochNode} = Seq,
+    ?assert(is_integer(SeqNum)),
+    ?assert(is_binary(SeqUuid)),
+    ?assert(is_atom(EpochNode)),
+
+    {ok, UuidMap} = fabric:db_uuids(DbName),
+    PrefixLen = fabric_util:get_uuid_prefix_len(),
+    Uuids = [binary:part(Uuid, {0, PrefixLen}) || Uuid <- maps:keys(UuidMap)],
+    [UuidFromShard] = Uuids,
+    ?assertEqual(UuidFromShard, SeqUuid),
+
+    ok = fabric:delete_db(DbName, []).
diff --git a/test/elixir/test/replication_test.exs b/test/elixir/test/replication_test.exs
index 12057d75b..b2e30ab95 100644
--- a/test/elixir/test/replication_test.exs
+++ b/test/elixir/test/replication_test.exs
@@ -1753,8 +1753,8 @@ defmodule ReplicationTest do
   def cmp_json(lhs, rhs), do: lhs == rhs
 
   def seq_to_shards(seq) do
-    for {_node, range, update_seq} <- decode_seq(seq) do
-      {range, update_seq}
+    for {_node, range, {seq_num, uuid, epoch}} <- decode_seq(seq) do
+      {range, seq_num}
     end
   end