Remove failed view jobs

If a client notices that a job has failed we restart it. If a job failed
for a different design document id then we resubmit the build request.

3 files changed, 22 insertions, 8 deletions

diff --git a/src/couch_views/src/couch_views_jobs.erl b/src/couch_views/src/couch_views_jobs.erl
index 76cc56337..d0de44ea8 100644
--- a/src/couch_views/src/couch_views_jobs.erl
+++ b/src/couch_views/src/couch_views_jobs.erl
@@ -35,7 +35,7 @@ set_timeout() ->
 
 build_view(TxDb, Mrst, UpdateSeq) ->
     {ok, JobId} = build_view_async(TxDb, Mrst),
-    case wait_for_job(JobId, UpdateSeq) of
+    case wait_for_job(JobId, Mrst#mrst.idx_name, UpdateSeq) of
         ok -> ok;
         retry -> build_view(TxDb, Mrst, UpdateSeq)
     end.
@@ -77,10 +77,10 @@ ensure_correct_tx(#{tx := Tx} = TxDb) ->
     end.
 
 
-wait_for_job(JobId, UpdateSeq) ->
+wait_for_job(JobId, DDocId, UpdateSeq) ->
     case couch_jobs:subscribe(?INDEX_JOB_TYPE, JobId) of
         {ok, Subscription, _State, _Data} ->
-            wait_for_job(JobId, Subscription, UpdateSeq);
+            wait_for_job(JobId, Subscription, DDocId, UpdateSeq);
         {ok, finished, Data} ->
             case Data of
                 #{<<"view_seq">> := ViewSeq} when ViewSeq >= UpdateSeq ->
@@ -91,21 +91,35 @@ wait_for_job(JobId, UpdateSeq) ->
     end.
 
 
-wait_for_job(JobId, Subscription, UpdateSeq) ->
+wait_for_job(JobId, Subscription, DDocId, UpdateSeq) ->
     case wait(Subscription) of
+        {not_found, not_found} ->
+            erlang:error(index_not_found);
         {error, Error} ->
             erlang:error(Error);
+        {finished, #{<<"error">> := <<"ddoc_deleted">>} = Data} ->
+            case maps:get(<<"ddoc_id">>, Data) of
+                DDocId ->
+                    couch_jobs:remove(undefined, ?INDEX_JOB_TYPE, JobId),
+                    erlang:error({ddoc_deleted, maps:get(<<"reason">>, Data)});
+                _OtherDocId ->
+                    % A different design doc wiht the same signature
+                    % was deleted. Resubmit this job which will overwrite
+                    % the ddoc_id in the job.
+                    retry
+            end;
         {finished, #{<<"error">> := Error, <<"reason">> := Reason}} ->
+            couch_jobs:remove(undefined, ?INDEX_JOB_TYPE, JobId),
             erlang:error({binary_to_existing_atom(Error, latin1), Reason});
         {finished, #{<<"view_seq">> := ViewSeq}} when ViewSeq >= UpdateSeq ->
             ok;
         {finished, _} ->
-            wait_for_job(JobId, UpdateSeq);
+            wait_for_job(JobId, DDocId, UpdateSeq);
         {_State, #{<<"view_seq">> := ViewSeq}} when ViewSeq >= UpdateSeq ->
             couch_jobs:unsubscribe(Subscription),
             ok;
         {_, _} ->
-            wait_for_job(JobId, Subscription, UpdateSeq)
+            wait_for_job(JobId, Subscription, DDocId, UpdateSeq)
     end.
 
 
diff --git a/src/couch_views/test/couch_views_indexer_test.erl b/src/couch_views/test/couch_views_indexer_test.erl
index 8ddb64b9c..54f787da3 100644
--- a/src/couch_views/test/couch_views_indexer_test.erl
+++ b/src/couch_views/test/couch_views_indexer_test.erl
@@ -375,7 +375,7 @@ index_autoupdater_callback(Db) ->
     ?assertMatch([{ok, <<_/binary>>}], Result),
     [{ok, JobId}] = Result,
 
-    ?assertEqual(ok, couch_views_jobs:wait_for_job(JobId, DbSeq)).
+    ?assertEqual(ok, couch_views_jobs:wait_for_job(JobId, DDoc#doc.id, DbSeq)).
 
 index_budget_is_changing(Db) ->
     ok = meck:new(couch_rate, [passthrough]),
diff --git a/src/couch_views/test/couch_views_map_test.erl b/src/couch_views/test/couch_views_map_test.erl
index 7d1e94b2c..2b679f07c 100644
--- a/src/couch_views/test/couch_views_map_test.erl
+++ b/src/couch_views/test/couch_views_map_test.erl
@@ -409,7 +409,7 @@ should_map_update_is_lazy() ->
     {ok, Mrst} = couch_views_util:ddoc_to_mrst(DbName, DDoc),
     JobId = couch_views_jobs:job_id(Db, Mrst),
     UpdateSeq = fabric2_db:get_update_seq(Db),
-    ok = couch_views_jobs:wait_for_job(JobId, UpdateSeq),
+    ok = couch_views_jobs:wait_for_job(JobId, DDoc#doc.id, UpdateSeq),
 
     Args2 = #{
         start_key => 8,