Check reader list in get_db, allow custom ctx in get_security

Supplying the user's context in fabric:get_security/2 offers a clean
way to check the database reader list early on in request processing.
Subsequent calls to open a database shard can then use open_int.

BugzID: 12038

src/fabric.erl

@@ -17,10 +17,12 @@
 -include_lib("mem3/include/mem3.hrl").
 -include_lib("couch/include/couch_db.hrl").
 
+-define(ADMIN_CTX, {user_ctx, #user_ctx{roles = [<<"_admin">>]}}).
+
 % DBs
 -export([all_dbs/0, all_dbs/1, create_db/1, create_db/2, delete_db/1,
     delete_db/2, get_db_info/1, get_doc_count/1, set_revs_limit/3,
-    set_security/3, get_revs_limit/1, get_security/1]).
+    set_security/3, get_revs_limit/1, get_security/1, get_security/2]).
 
 % Documents
 -export([open_doc/3, open_revs/4, get_missing_revs/2, get_missing_revs/3,
@@ -123,18 +125,21 @@ set_revs_limit(DbName, Limit, Options) when is_integer(Limit), Limit > 0 ->
 %% @doc retrieves the maximum number of document revisions
 -spec get_revs_limit(dbname()) -> pos_integer() | no_return().
 get_revs_limit(DbName) ->
-    {ok, Db} = fabric_util:get_db(dbname(DbName)),
+    {ok, Db} = fabric_util:get_db(dbname(DbName), [?ADMIN_CTX]),
     try couch_db:get_revs_limit(Db) after catch couch_db:close(Db) end.
 
 %% @doc sets the readers/writers/admin permissions for a database
 -spec set_security(dbname(), SecObj::json_obj(), [option()]) -> ok.
 set_security(DbName, SecObj, Options) ->
     fabric_db_meta:set_security(dbname(DbName), SecObj, opts(Options)).
 
+get_security(DbName) ->
+    get_security(DbName, [?ADMIN_CTX]).
+
 %% @doc retrieve the security object for a database
 -spec get_security(dbname()) -> json_obj() | no_return().
-get_security(DbName) ->
-    {ok, Db} = fabric_util:get_db(dbname(DbName)),
+get_security(DbName, Options) ->
+    {ok, Db} = fabric_util:get_db(dbname(DbName), opts(Options)),
     try couch_db:get_security(Db) after catch couch_db:close(Db) end.
 
 % doc operations

src/fabric_util.erl

@@ -14,7 +14,7 @@
 
 -module(fabric_util).
 
--export([submit_jobs/3, cleanup/1, recv/4, get_db/1, error_info/1,
+-export([submit_jobs/3, cleanup/1, recv/4, get_db/1, get_db/2, error_info/1,
         update_counter/3, remove_ancestors/2, kv/2]).
 
 -include("fabric.hrl").
@@ -40,21 +40,30 @@ recv(Workers, Keypos, Fun, Acc0) ->
 
 
 get_db(DbName) ->
-    % sort shards so we try the local ones first
+    get_db(DbName, []).
+
+get_db(DbName, Options) ->
+    % prefer local shards
     {Local, Remote} = lists:partition(fun(S) -> S#shard.node =:= node() end,
         mem3:shards(DbName)),
-    get_shard(Local ++ Remote, 100).
+    % suppress shards from down nodes
+    Nodes = erlang:nodes(),
+    Live = [S || #shard{node = N} = S <- Remote, lists:member(N, Nodes)],
+    % sort the live remote shards so that we don't repeatedly try the same node
+    get_shard(Local ++ lists:keysort(#shard.name, Live), Options, 100).
 
-get_shard([], _Timeout) ->
+get_shard([], _Opts, _Timeout) ->
     erlang:error({internal_server_error, "No DB shards could be opened."});
-get_shard([#shard{node = Node, name = Name} | Rest], Timeout) ->
-    case rpc:call(Node, couch_db, open_int, [Name, [{timeout, Timeout}]]) of
+get_shard([#shard{node = Node, name = Name} | Rest], Opts, Timeout) ->
+    case rpc:call(Node, couch_db, open, [Name, [{timeout, Timeout} | Opts]]) of
     {ok, Db} ->
         {ok, Db};
+    {unauthorized, _} = Error ->
+        throw(Error);
     {badrpc, {'EXIT', {timeout, _}}} ->
-        get_shard(Rest, 2*Timeout);
+        get_shard(Rest, Opts, 2*Timeout);
     _Else ->
-        get_shard(Rest, Timeout)
+        get_shard(Rest, Opts, Timeout)
     end.
 
 error_info({{<<"reduce_overflow_error">>, _} = Error, _Stack}) ->