Implement the ability to downgrade CouchDB versions

Currently, CouchDB on-disk header record allows appending new fields to it in
such a way that newer versions can upgrade themselves from the old versions
easily if the ?LATEST_DISK_VERSION stays the same. However, it was not possible
to perform a downgrade back to an old version in case something went wrong.

While in general it may not be safe to downgrade in such cases, it may be
possible for some features, so allow for such an option and make it
configurable.

This is essentially a counterpart to this feature:

```
% As long the changes are limited to new header fields (with inline
% defaults) added to the end of the record, then there is no need to increment
% the disk revision number.
```

The release notes of future releases may indicate which version are downgrade
safe. Then, to perform a downgrade users would enable the downgrade flag,
downgrade, and then reset it back to default (prohibit = true).

Implementation-wise, it's pretty basic, if the size of the new tuple is larger
than the old one, it's a downgrade. Then, only use as many tuple fields as
current (aka the "old" version) knows about, everything else is discarded.

Author: nickva

rel/overlay/etc/default.ini

@@ -120,6 +120,16 @@ view_index_dir = {{view_index_dir}}
 ; this setting.
 ;cfile_skip_ioq = false
 
+; CouchDB will not normally allow a database downgrade to a previous version if
+; the new version added new fields to the database file header structure. You
+; can disable this protection by setting this to false. We recommend
+; re-enabling the protection after you have performed the downgrade but it is
+; not mandatory to do so.
+; Note that some future versions of CouchDB might not support downgrade at all,
+; whatever value this is set to. In those cases CouchDB will refuse to
+; downgrade or even open the databases in question.
+;prohibit_downgrade = true
+
 [purge]
 ; Allowed maximum number of documents in one purge request
 ;max_document_id_number = 100

src/couch/src/couch_bt_engine_header.erl

@@ -204,17 +204,16 @@ upgrade_tuple(Old) when is_record(Old, db_header) ->
     Old;
 upgrade_tuple(Old) when is_tuple(Old) ->
     NewSize = record_info(size, db_header),
-    if
-        tuple_size(Old) < NewSize -> ok;
-        true -> erlang:error({invalid_header_size, Old})
-    end,
-    {_, New} = lists:foldl(
-        fun(Val, {Idx, Hdr}) ->
-            {Idx + 1, setelement(Idx, Hdr, Val)}
+    Upgrade = tuple_size(Old) < NewSize,
+    ProhibitDowngrade = config:get_boolean("couchdb", "prohibit_downgrade", true),
+    OldKVs =
+        case {Upgrade, ProhibitDowngrade} of
+            {true, AnyBool} when is_boolean(AnyBool) -> tuple_to_list(Old);
+            {false, true} -> error({invalid_header_size, Old});
+            {false, false} -> lists:sublist(tuple_to_list(Old), NewSize)
         end,
-        {1, #db_header{}},
-        tuple_to_list(Old)
-    ),
+    FoldFun = fun(Val, {Idx, Hdr}) -> {Idx + 1, setelement(Idx, Hdr, Val)} end,
+    {_, New} = lists:foldl(FoldFun, {1, #db_header{}}, OldKVs),
     if
         is_record(New, db_header) -> ok;
         true -> erlang:error({invalid_header_extension, {Old, New}})
@@ -338,7 +337,7 @@ latest(_Else) ->
     undefined.
 
 -ifdef(TEST).
--include_lib("eunit/include/eunit.hrl").
+-include_lib("couch/include/couch_eunit.hrl").
 
 mk_header(Vsn) ->
     {
@@ -478,4 +477,61 @@ get_epochs_from_old_header_test() ->
     Vsn5Header = mk_header(5),
     ?assertEqual(undefined, epochs(Vsn5Header)).
 
+tuple_uprade_test_() ->
+    {
+        foreach,
+        fun() ->
+            Ctx = test_util:start_couch(),
+            config:set("couchdb", "prohibit_downgrade", "true", false),
+            Ctx
+        end,
+        fun(Ctx) ->
+            config:delete("couchdb", "prohibit_downgrade", false),
+            test_util:stop_couch(Ctx)
+        end,
+        [
+            ?TDEF_FE(t_upgrade_tuple_same_size),
+            ?TDEF_FE(t_upgrade_tuple),
+            ?TDEF_FE(t_downgrade_default),
+            ?TDEF_FE(t_downgrade_allowed)
+        ]
+    }.
+
+t_upgrade_tuple_same_size(_) ->
+    Hdr = #db_header{disk_version = ?LATEST_DISK_VERSION},
+    Hdr1 = upgrade_tuple(Hdr),
+    ?assertEqual(Hdr, Hdr1).
+
+t_upgrade_tuple(_) ->
+    Hdr = {db_header, ?LATEST_DISK_VERSION, 101},
+    Hdr1 = upgrade_tuple(Hdr),
+    ?assertMatch(
+        #db_header{
+            disk_version = ?LATEST_DISK_VERSION,
+            update_seq = 101,
+            purge_infos_limit = 1000
+        },
+        Hdr1
+    ).
+
+t_downgrade_default(_) ->
+    Junk = lists:duplicate(50, x),
+    Hdr = list_to_tuple([db_header, ?LATEST_DISK_VERSION] ++ Junk),
+    % Not allowed by default
+    ?assertError({invalid_header_size, _}, upgrade_tuple(Hdr)).
+
+t_downgrade_allowed(_) ->
+    Junk = lists:duplicate(50, x),
+    Hdr = list_to_tuple([db_header, ?LATEST_DISK_VERSION, 42] ++ Junk),
+    config:set("couchdb", "prohibit_downgrade", "false", false),
+    Hdr1 = upgrade_tuple(Hdr),
+    ?assert(is_record(Hdr1, db_header)),
+    ?assertMatch(
+        #db_header{
+            disk_version = ?LATEST_DISK_VERSION,
+            update_seq = 42
+        },
+        Hdr1
+    ).
+
 -endif.