ensure we decrement clients_requesting_changes

ensure we decrement clients_requesting_changes even if the mochiweb process
mediating the change request is killed by the client disconnect detector.

Author: rnewson

src/chttpd/src/chttpd_db.erl

@@ -151,11 +151,8 @@ handle_changes_req1(#httpd{} = Req, Db) ->
                 mochi = Req,
                 threshold = Max
             },
-            try
-                fabric:changes(Db, fun changes_callback/2, Acc0, ChangesArgs)
-            after
-                couch_stats:decrement_counter([couchdb, httpd, clients_requesting_changes])
-            end;
+            couch_changes_mon:decrement_clients_requesting_changes_on_exit(),
+            fabric:changes(Db, fun changes_callback/2, Acc0, ChangesArgs);
         _ ->
             Msg = <<"Supported `feed` types: normal, continuous, live, longpoll, eventsource">>,
             throw({bad_request, Msg})

src/couch/src/couch_changes_mon.erl

@@ -0,0 +1,47 @@
+% 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(couch_changes_mon).
+
+-behaviour(gen_server).
+
+-export([decrement_clients_requesting_changes_on_exit/0]).
+
+-export([
+    start_link/0,
+    init/1,
+    handle_call/3,
+    handle_cast/2,
+    handle_info/2
+]).
+
+decrement_clients_requesting_changes_on_exit() ->
+    gen_server:cast(?MODULE, {monitor_me, self()}).
+
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+init(_) ->
+    {ok, nil}.
+
+handle_call(_Msg, _From, State) ->
+    {reply, error, State}.
+
+handle_cast({monitor_me, Pid}, State) ->
+    monitor(process, Pid),
+    {noreply, State}.
+
+handle_info({'DOWN', _MonitorRef, process, _Pid, _Info}, State) ->
+    couch_stats:decrement_counter([couchdb, httpd, clients_requesting_changes]),
+    {noreply, State};
+handle_info(_Msg, State) ->
+    {noreply, State}.

src/couch/src/couch_httpd_db.erl

@@ -189,13 +189,8 @@ handle_changes_req1(Req, Db, ChangesArgs, ChangesFun) ->
     couch_stats:increment_counter(
         [couchdb, httpd, clients_requesting_changes]
     ),
-    try
-        WrapperFun(ChangesFun)
-    after
-        couch_stats:decrement_counter(
-            [couchdb, httpd, clients_requesting_changes]
-        )
-    end.
+    couch_changes_mon:decrement_clients_requesting_changes_on_exit(),
+    WrapperFun(ChangesFun).
 
 handle_compact_req(#httpd{method = 'POST'} = Req, Db) ->
     case Req#httpd.path_parts of

src/couch/src/couch_secondary_sup.erl

@@ -27,7 +27,8 @@ init([]) ->
             {query_servers, {couch_proc_manager, start_link, []}},
             {vhosts, {couch_httpd_vhost, start_link, []}},
             {uuids, {couch_uuids, start, []}},
-            {disk_manager, {couch_disk_monitor, start_link, []}}
+            {disk_manager, {couch_disk_monitor, start_link, []}},
+            {couch_changes_mon, {couch_changes_mon, start_link, []}}
         ] ++ couch_index_servers(),
 
     MaybeHttp =