Add Global timeout for map/reduce queries

This makes the global timeout for a map/reduce and all_docs request
configurable via the config. It separates the config into global queries
and partition queries so that it is possible to make the global timeout
is for partitioned queries

5 files changed, 46 insertions, 4 deletions

diff --git a/src/fabric/src/fabric_util.erl b/src/fabric/src/fabric_util.erl
index d65d3c81d..1ba1d17ea 100644
--- a/src/fabric/src/fabric_util.erl
+++ b/src/fabric/src/fabric_util.erl
@@ -15,7 +15,7 @@
 -export([submit_jobs/3, submit_jobs/4, cleanup/1, recv/4, get_db/1, get_db/2, error_info/1,
         update_counter/3, remove_ancestors/2, create_monitors/1, kv/2,
         remove_down_workers/2, doc_id_and_rev/1]).
--export([request_timeout/0, attachments_timeout/0, all_docs_timeout/0]).
+-export([request_timeout/0, attachments_timeout/0, all_docs_timeout/0, view_timeout/1]).
 -export([log_timeout/2, remove_done_workers/2]).
 -export([is_users_db/1, is_replicator_db/1]).
 -export([open_cluster_db/1, open_cluster_db/2]).
@@ -65,6 +65,13 @@ all_docs_timeout() ->
 attachments_timeout() ->
     timeout("attachments", "600000").
 
+view_timeout(Args) ->
+    PartitionQuery = couch_mrview_util:get_extra(Args, partition, false),
+    case PartitionQuery of
+        false -> timeout("view", "infinity");
+        _ -> timeout("partition_view", "infinity")
+    end.
+
 timeout(Type, Default) ->
     case config:get("fabric", Type ++ "_timeout", Default) of
         "infinity" -> infinity;
diff --git a/src/fabric/src/fabric_view_all_docs.erl b/src/fabric/src/fabric_view_all_docs.erl
index 4b412a683..9049eaa90 100644
--- a/src/fabric/src/fabric_view_all_docs.erl
+++ b/src/fabric/src/fabric_view_all_docs.erl
@@ -127,7 +127,7 @@ go(DbName, _Options, Workers, QueryArgs, Callback, Acc0) ->
         update_seq = case UpdateSeq of true -> []; false -> nil end
     },
     case rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
-        State, infinity, 5000) of
+        State, fabric_util:view_timeout(QueryArgs), 5000) of
     {ok, NewState} ->
         {ok, NewState#collector.user_acc};
     {timeout, NewState} ->
diff --git a/src/fabric/src/fabric_view_map.erl b/src/fabric/src/fabric_view_map.erl
index b3d768a51..1bbc6d201 100644
--- a/src/fabric/src/fabric_view_map.erl
+++ b/src/fabric/src/fabric_view_map.erl
@@ -81,7 +81,7 @@ go(DbName, Workers, {map, View, _}, Args, Callback, Acc0) ->
         update_seq = case UpdateSeq of true -> []; false -> nil end
     },
     case rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
-        State, infinity, 1000 * 60 * 60) of
+        State, fabric_util:view_timeout(Args), 1000 * 60 * 60) of
     {ok, NewState} ->
         {ok, NewState#collector.user_acc};
     {timeout, NewState} ->
diff --git a/src/fabric/src/fabric_view_reduce.erl b/src/fabric/src/fabric_view_reduce.erl
index f52061a4c..b157c37e5 100644
--- a/src/fabric/src/fabric_view_reduce.erl
+++ b/src/fabric/src/fabric_view_reduce.erl
@@ -87,7 +87,7 @@ go2(DbName, Workers, {red, {_, Lang, View}, _}=VInfo, Args, Callback, Acc0) ->
         update_seq = case UpdateSeq of true -> []; false -> nil end
     },
     try rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
-        State, infinity, 1000 * 60 * 60) of
+        State, fabric_util:view_timeout(Args), 1000 * 60 * 60) of
     {ok, NewState} ->
         {ok, NewState#collector.user_acc};
     {timeout, NewState} ->
diff --git a/test/elixir/test/partition_all_docs_test.exs b/test/elixir/test/partition_all_docs_test.exs
index 0941daf59..308c1b439 100644
--- a/test/elixir/test/partition_all_docs_test.exs
+++ b/test/elixir/test/partition_all_docs_test.exs
@@ -115,4 +115,39 @@ defmodule PartitionAllDocsTest do
     assert length(ids) == 1
     assert ids == ["foo:22"]
   end
+
+  # This test is timing based so it could be a little flaky.
+  # If that turns out to be the case we should probably just skip it
+  test "partition _all_docs with timeout", context do
+    on_exit(fn ->
+      resp = Couch.get("/_membership")
+      %{:body => body} = resp
+
+      Enum.each(body["all_nodes"], fn node ->
+        resp = Couch.put("/_node/#{node}/_config/fabric/partition_view_timeout", body: "\"3600000\"")
+        assert resp.status_code == 200
+      end)
+    end)
+
+    resp = Couch.get("/_membership")
+    %{:body => body} = resp
+
+     Enum.each(body["all_nodes"], fn node ->
+      resp = Couch.put("/_node/#{node}/_config/fabric/partition_view_timeout", body: "\"1\"")
+      assert resp.status_code == 200
+    end)
+
+    db_name = context[:db_name]
+    create_partition_docs(db_name)
+
+    retry_until(fn ->
+      url = "/#{db_name}/_partition/foo/_all_docs"
+      case Couch.get(url) do
+        %{:body => %{"reason" => reason}} ->
+          Regex.match?(~r/not be processed in a reasonable amount of time./, reason)
+        _ -> 
+          false
+      end
+    end)
+  end
 end