{"action":"created","comment":{"url":"https://api.github.com/repos/apache/couchdb/pulls/comments/243377690","pull_request_review_id":187132901,"id":243377690,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDI0MzM3NzY5MA==","diff_hunk":"@@ -0,0 +1,216 @@\n+% Licensed under the Apache License, Version 2.0 (the \"License\"); you may not\n+% use this file except in compliance with the License. You may obtain a copy of\n+% the License at\n+%\n+%   http://www.apache.org/licenses/LICENSE-2.0\n+%\n+% Unless required by applicable law or agreed to in writing, software\n+% distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n+% License for the specific language governing permissions and limitations under\n+% the License.\n+\n+-module(fabric_streams).\n+\n+-export([\n+    start/2,\n+    start/4,\n+    cleanup/1\n+]).\n+\n+-include_lib(\"fabric/include/fabric.hrl\").\n+-include_lib(\"mem3/include/mem3.hrl\").\n+\n+\n+-define(WORKER_CLEANER, fabric_worker_cleaner).\n+\n+\n+start(Workers, Keypos) ->\n+    start(Workers, Keypos, undefined, undefined).\n+\n+start(Workers0, Keypos, StartFun, Replacements) ->\n+    Fun = fun handle_stream_start/3,\n+    Acc = #stream_acc{\n+        workers = fabric_dict:init(Workers0, waiting),\n+        start_fun = StartFun,\n+        replacements = Replacements\n+    },\n+    spawn_worker_cleaner(self(), Workers0),\n+    Timeout = fabric_util:request_timeout(),\n+    case rexi_utils:recv(Workers0, Keypos, Fun, Acc, Timeout, infinity) of\n+        {ok, #stream_acc{workers=Workers}} ->\n+            true = fabric_view:is_progress_possible(Workers),\n+            AckedWorkers = fabric_dict:fold(fun(Worker, From, WorkerAcc) ->\n+                rexi:stream_start(From),\n+                [Worker | WorkerAcc]\n+            end, [], Workers),\n+            {ok, AckedWorkers};\n+        Else ->\n+            Else\n+    end.\n+\n+\n+cleanup(Workers) ->\n+    % Stop the auxiliary cleaner process as we got to the point where cleanup\n+    % happesn in the regular fashion so we don't want to send 2x the number kill\n+    % messages\n+    case get(?WORKER_CLEANER) of\n+        CleanerPid when is_pid(CleanerPid) ->\n+            erase(?WORKER_CLEANER),\n+            exit(CleanerPid, kill);\n+        _ ->\n+            ok\n+    end,\n+    fabric_util:cleanup(Workers).\n+\n+\n+handle_stream_start({rexi_DOWN, _, {_, NodeRef}, _}, _, St) ->\n+    case fabric_util:remove_down_workers(St#stream_acc.workers, NodeRef) of\n+    {ok, Workers} ->\n+        {ok, St#stream_acc{workers=Workers}};\n+    error ->\n+        Reason = {nodedown, <<\"progress not possible\">>},\n+        {error, Reason}\n+    end;\n+\n+handle_stream_start({rexi_EXIT, Reason}, Worker, St) ->\n+    Workers = fabric_dict:erase(Worker, St#stream_acc.workers),\n+    Replacements = St#stream_acc.replacements,\n+    case {fabric_view:is_progress_possible(Workers), Reason} of\n+    {true, _} ->\n+        {ok, St#stream_acc{workers=Workers}};\n+    {false, {maintenance_mode, _Node}} when Replacements /= undefined ->\n+        % Check if we have replacements for this range\n+        % and start the new workers if so.\n+        case lists:keytake(Worker#shard.range, 1, Replacements) of\n+            {value, {_Range, WorkerReplacements}, NewReplacements} ->\n+                FinalWorkers = lists:foldl(fun(Repl, NewWorkers) ->\n+                    NewWorker = (St#stream_acc.start_fun)(Repl),\n+                    fabric_dict:store(NewWorker, waiting, NewWorkers)\n+                end, Workers, WorkerReplacements),\n+                % Assert that our replaced worker provides us\n+                % the oppurtunity to make progress.\n+                true = fabric_view:is_progress_possible(FinalWorkers),\n+                NewRefs = fabric_dict:fetch_keys(FinalWorkers),\n+                {new_refs, NewRefs, St#stream_acc{\n+                    workers=FinalWorkers,\n+                    replacements=NewReplacements\n+                }};\n+            false ->\n+                % If we progress isn't possible and we don't have any\n+                % replacements then we're dead in the water.\n+                Error = {nodedown, <<\"progress not possible\">>},\n+                {error, Error}\n+        end;\n+    {false, _} ->\n+        {error, fabric_util:error_info(Reason)}\n+    end;\n+\n+handle_stream_start(rexi_STREAM_INIT, {Worker, From}, St) ->\n+    case fabric_dict:lookup_element(Worker, St#stream_acc.workers) of\n+    undefined ->\n+        % This worker lost the race with other partition copies, terminate\n+        rexi:stream_cancel(From),\n+        {ok, St};\n+    waiting ->\n+        % Don't ack the worker yet so they don't start sending us\n+        % rows until we're ready\n+        Workers0 = fabric_dict:store(Worker, From, St#stream_acc.workers),\n+        Workers1 = fabric_view:remove_overlapping_shards(Worker, Workers0),\n+        case fabric_dict:any(waiting, Workers1) of\n+            true ->\n+                {ok, St#stream_acc{workers=Workers1}};\n+            false ->\n+                {stop, St#stream_acc{workers=Workers1}}\n+        end\n+    end;\n+\n+handle_stream_start({ok, ddoc_updated}, _, St) ->\n+    cleanup(St#stream_acc.workers),\n+    {stop, ddoc_updated};\n+\n+handle_stream_start(Else, _, _) ->\n+    exit({invalid_stream_start, Else}).\n+\n+\n+% Spawn an auxiliary rexi worker cleaner. This will be used in cases\n+% when the coordinator (request) process is forceably killed and doesn't\n+% get a chance to process its `after` fabric:clean/1 clause.\n+spawn_worker_cleaner(Coordinator, Workers) ->\n+    case get(?WORKER_CLEANER) of\n+        undefined ->\n+            Pid = spawn(fun() ->","path":"src/fabric/src/fabric_streams.erl","position":143,"original_position":142,"commit_id":"41757cd0f8ec48991c0c2fddcb6085b77e636e09","original_commit_id":"8ce8abdebf1df9d4a3751b72f043df7c9babee5a","user":{"login":"nickva","id":211822,"node_id":"MDQ6VXNlcjIxMTgyMg==","avatar_url":"https://avatars3.githubusercontent.com/u/211822?v=4","gravatar_id":"","url":"https://api.github.com/users/nickva","html_url":"https://github.com/nickva","followers_url":"https://api.github.com/users/nickva/followers","following_url":"https://api.github.com/users/nickva/following{/other_user}","gists_url":"https://api.github.com/users/nickva/gists{/gist_id}","starred_url":"https://api.github.com/users/nickva/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/nickva/subscriptions","organizations_url":"https://api.github.com/users/nickva/orgs","repos_url":"https://api.github.com/users/nickva/repos","events_url":"https://api.github.com/users/nickva/events{/privacy}","received_events_url":"https://api.github.com/users/nickva/received_events","type":"User","site_admin":false},"body":"Good call. Added a thing  for it","created_at":"2018-12-20T18:44:00Z","updated_at":"2018-12-20T18:44:00Z","html_url":"https://github.com/apache/couchdb/pull/1824#discussion_r243377690","pull_request_url":"https://api.github.com/repos/apache/couchdb/pulls/1824","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/couchdb/pulls/comments/243377690"},"html":{"href":"https://github.com/apache/couchdb/pull/1824#discussion_r243377690"},"pull_request":{"href":"https://api.github.com/repos/apache/couchdb/pulls/1824"}},"in_reply_to_id":243364963},"pull_request":{"url":"https://api.github.com/repos/apache/couchdb/pulls/1824","id":240207960,"node_id":"MDExOlB1bGxSZXF1ZXN0MjQwMjA3OTYw","html_url":"https://github.com/apache/couchdb/pull/1824","diff_url":"https://github.com/apache/couchdb/pull/1824.diff","patch_url":"https://github.com/apache/couchdb/pull/1824.patch","issue_url":"https://api.github.com/repos/apache/couchdb/issues/1824","number":1824,"state":"open","locked":false,"title":"Clean rexi stream workers when coordinator process is killed","user":{"login":"nickva","id":211822,"node_id":"MDQ6VXNlcjIxMTgyMg==","avatar_url":"https://avatars3.githubusercontent.com/u/211822?v=4","gravatar_id":"","url":"https://api.github.com/users/nickva","html_url":"https://github.com/nickva","followers_url":"https://api.github.com/users/nickva/followers","following_url":"https://api.github.com/users/nickva/following{/other_user}","gists_url":"https://api.github.com/users/nickva/gists{/gist_id}","starred_url":"https://api.github.com/users/nickva/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/nickva/subscriptions","organizations_url":"https://api.github.com/users/nickva/orgs","repos_url":"https://api.github.com/users/nickva/repos","events_url":"https://api.github.com/users/nickva/events{/privacy}","received_events_url":"https://api.github.com/users/nickva/received_events","type":"User","site_admin":false},"body":"Sometimes fabric coordinators end up getting brutally terminated [1], and in that\r\ncase they might never process their `after` clause where their remote rexi\r\nworkers are killed. Those workers are left lingering around keeping databases\r\nactive for up to 5 minutes at a time.\r\n\r\nTo prevent that from happening, let coordinators which use streams spawn an\r\nauxiliary cleaner process. This process will monitor the main coordinator and\r\nif it dies will ensure remote workers are killed, freeing resources\r\nimmediately. In order not to send 2x the number of kill messages during the\r\nnormal exit, fabric_util:cleanup() will stop the auxiliary process before\r\ncontinuing.\r\n\r\n[1] One instance is when the ddoc cache is refreshed:\r\n https://github.com/apache/couchdb/blob/master/src/ddoc_cache/src/ddoc_cache_entry.erl#L236\r\n\r\n","created_at":"2018-12-20T15:25:55Z","updated_at":"2018-12-20T18:44:00Z","closed_at":null,"merged_at":null,"merge_commit_sha":"e2880721fc8d2d5b733607a78e54345934dc1ce4","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/couchdb/pulls/1824/commits","review_comments_url":"https://api.github.com/repos/apache/couchdb/pulls/1824/comments","review_comment_url":"https://api.github.com/repos/apache/couchdb/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/couchdb/issues/1824/comments","statuses_url":"https://api.github.com/repos/apache/couchdb/statuses/41757cd0f8ec48991c0c2fddcb6085b77e636e09","head":{"label":"cloudant:clean-rexi-stream-workers","ref":"clean-rexi-stream-workers","sha":"41757cd0f8ec48991c0c2fddcb6085b77e636e09","user":{"login":"cloudant","id":44642,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ0NjQy","avatar_url":"https://avatars0.githubusercontent.com/u/44642?v=4","gravatar_id":"","url":"https://api.github.com/users/cloudant","html_url":"https://github.com/cloudant","followers_url":"https://api.github.com/users/cloudant/followers","following_url":"https://api.github.com/users/cloudant/following{/other_user}","gists_url":"https://api.github.com/users/cloudant/gists{/gist_id}","starred_url":"https://api.github.com/users/cloudant/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloudant/subscriptions","organizations_url":"https://api.github.com/users/cloudant/orgs","repos_url":"https://api.github.com/users/cloudant/repos","events_url":"https://api.github.com/users/cloudant/events{/privacy}","received_events_url":"https://api.github.com/users/cloudant/received_events","type":"Organization","site_admin":false},"repo":{"id":39005786,"node_id":"MDEwOlJlcG9zaXRvcnkzOTAwNTc4Ng==","name":"couchdb","full_name":"cloudant/couchdb","private":false,"owner":{"login":"cloudant","id":44642,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ0NjQy","avatar_url":"https://avatars0.githubusercontent.com/u/44642?v=4","gravatar_id":"","url":"https://api.github.com/users/cloudant","html_url":"https://github.com/cloudant","followers_url":"https://api.github.com/users/cloudant/followers","following_url":"https://api.github.com/users/cloudant/following{/other_user}","gists_url":"https://api.github.com/users/cloudant/gists{/gist_id}","starred_url":"https://api.github.com/users/cloudant/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloudant/subscriptions","organizations_url":"https://api.github.com/users/cloudant/orgs","repos_url":"https://api.github.com/users/cloudant/repos","events_url":"https://api.github.com/users/cloudant/events{/privacy}","received_events_url":"https://api.github.com/users/cloudant/received_events","type":"Organization","site_admin":false},"html_url":"https://github.com/cloudant/couchdb","description":"Mirror of Apache CouchDB","fork":true,"url":"https://api.github.com/repos/cloudant/couchdb","forks_url":"https://api.github.com/repos/cloudant/couchdb/forks","keys_url":"https://api.github.com/repos/cloudant/couchdb/keys{/key_id}","collaborators_url":"https://api.github.com/repos/cloudant/couchdb/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/cloudant/couchdb/teams","hooks_url":"https://api.github.com/repos/cloudant/couchdb/hooks","issue_events_url":"https://api.github.com/repos/cloudant/couchdb/issues/events{/number}","events_url":"https://api.github.com/repos/cloudant/couchdb/events","assignees_url":"https://api.github.com/repos/cloudant/couchdb/assignees{/user}","branches_url":"https://api.github.com/repos/cloudant/couchdb/branches{/branch}","tags_url":"https://api.github.com/repos/cloudant/couchdb/tags","blobs_url":"https://api.github.com/repos/cloudant/couchdb/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/cloudant/couchdb/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/cloudant/couchdb/git/refs{/sha}","trees_url":"https://api.github.com/repos/cloudant/couchdb/git/trees{/sha}","statuses_url":"https://api.github.com/repos/cloudant/couchdb/statuses/{sha}","languages_url":"https://api.github.com/repos/cloudant/couchdb/languages","stargazers_url":"https://api.github.com/repos/cloudant/couchdb/stargazers","contributors_url":"https://api.github.com/repos/cloudant/couchdb/contributors","subscribers_url":"https://api.github.com/repos/cloudant/couchdb/subscribers","subscription_url":"https://api.github.com/repos/cloudant/couchdb/subscription","commits_url":"https://api.github.com/repos/cloudant/couchdb/commits{/sha}","git_commits_url":"https://api.github.com/repos/cloudant/couchdb/git/commits{/sha}","comments_url":"https://api.github.com/repos/cloudant/couchdb/comments{/number}","issue_comment_url":"https://api.github.com/repos/cloudant/couchdb/issues/comments{/number}","contents_url":"https://api.github.com/repos/cloudant/couchdb/contents/{+path}","compare_url":"https://api.github.com/repos/cloudant/couchdb/compare/{base}...{head}","merges_url":"https://api.github.com/repos/cloudant/couchdb/merges","archive_url":"https://api.github.com/repos/cloudant/couchdb/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/cloudant/couchdb/downloads","issues_url":"https://api.github.com/repos/cloudant/couchdb/issues{/number}","pulls_url":"https://api.github.com/repos/cloudant/couchdb/pulls{/number}","milestones_url":"https://api.github.com/repos/cloudant/couchdb/milestones{/number}","notifications_url":"https://api.github.com/repos/cloudant/couchdb/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/cloudant/couchdb/labels{/name}","releases_url":"https://api.github.com/repos/cloudant/couchdb/releases{/id}","deployments_url":"https://api.github.com/repos/cloudant/couchdb/deployments","created_at":"2015-07-13T10:27:13Z","updated_at":"2018-05-21T06:03:17Z","pushed_at":"2018-12-20T18:24:37Z","git_url":"git://github.com/cloudant/couchdb.git","ssh_url":"git@github.com:cloudant/couchdb.git","clone_url":"https://github.com/cloudant/couchdb.git","svn_url":"https://github.com/cloudant/couchdb","homepage":null,"size":36880,"stargazers_count":7,"watchers_count":7,"language":"Erlang","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":4,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"other","name":"Other","spdx_id":"NOASSERTION","url":null,"node_id":"MDc6TGljZW5zZTA="},"forks":4,"open_issues":0,"watchers":7,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"f60f7a1c66a9b238ae66798b82058a4d9dc82731","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":206417,"node_id":"MDEwOlJlcG9zaXRvcnkyMDY0MTc=","name":"couchdb","full_name":"apache/couchdb","private":false,"owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"html_url":"https://github.com/apache/couchdb","description":"Apache CouchDB","fork":false,"url":"https://api.github.com/repos/apache/couchdb","forks_url":"https://api.github.com/repos/apache/couchdb/forks","keys_url":"https://api.github.com/repos/apache/couchdb/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/couchdb/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/couchdb/teams","hooks_url":"https://api.github.com/repos/apache/couchdb/hooks","issue_events_url":"https://api.github.com/repos/apache/couchdb/issues/events{/number}","events_url":"https://api.github.com/repos/apache/couchdb/events","assignees_url":"https://api.github.com/repos/apache/couchdb/assignees{/user}","branches_url":"https://api.github.com/repos/apache/couchdb/branches{/branch}","tags_url":"https://api.github.com/repos/apache/couchdb/tags","blobs_url":"https://api.github.com/repos/apache/couchdb/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/couchdb/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/couchdb/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/couchdb/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/couchdb/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/couchdb/languages","stargazers_url":"https://api.github.com/repos/apache/couchdb/stargazers","contributors_url":"https://api.github.com/repos/apache/couchdb/contributors","subscribers_url":"https://api.github.com/repos/apache/couchdb/subscribers","subscription_url":"https://api.github.com/repos/apache/couchdb/subscription","commits_url":"https://api.github.com/repos/apache/couchdb/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/couchdb/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/couchdb/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/couchdb/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/couchdb/contents/{+path}","compare_url":"https://api.github.com/repos/apache/couchdb/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/couchdb/merges","archive_url":"https://api.github.com/repos/apache/couchdb/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/couchdb/downloads","issues_url":"https://api.github.com/repos/apache/couchdb/issues{/number}","pulls_url":"https://api.github.com/repos/apache/couchdb/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/couchdb/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/couchdb/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/couchdb/labels{/name}","releases_url":"https://api.github.com/repos/apache/couchdb/releases{/id}","deployments_url":"https://api.github.com/repos/apache/couchdb/deployments","created_at":"2009-05-21T02:03:38Z","updated_at":"2018-12-20T16:13:48Z","pushed_at":"2018-12-20T18:24:38Z","git_url":"git://github.com/apache/couchdb.git","ssh_url":"git@github.com:apache/couchdb.git","clone_url":"https://github.com/apache/couchdb.git","svn_url":"https://github.com/apache/couchdb","homepage":"","size":30709,"stargazers_count":3977,"watchers_count":3977,"language":"Erlang","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":811,"mirror_url":null,"archived":false,"open_issues_count":169,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":811,"open_issues":169,"watchers":3977,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/couchdb/pulls/1824"},"html":{"href":"https://github.com/apache/couchdb/pull/1824"},"issue":{"href":"https://api.github.com/repos/apache/couchdb/issues/1824"},"comments":{"href":"https://api.github.com/repos/apache/couchdb/issues/1824/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/couchdb/pulls/1824/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/couchdb/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/couchdb/pulls/1824/commits"},"statuses":{"href":"https://api.github.com/repos/apache/couchdb/statuses/41757cd0f8ec48991c0c2fddcb6085b77e636e09"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"nickva"}}