[10/N] Clustered Purge: Clustered HTTP APICOUCHDB-3326-clustered-purge-pr4-implementation

The HTTP API for clustered purge is fairly straightforward. It is
designed to match the general shape of the single node API. The only
major caveat here is that the purge sequence is now hardcoded as null
since the purge sequence would now otherwise be an opaque blob similar
to the update_seq blobs.

Its important to note that there is as yet no API invented for
traversing the history of purge requests in any shape or form as that
would mostly invalidate the entire purpose of using purge to remove any
trace of a document from a database at the HTTP level. Although there
will still be traces in individual shard files until all database
components have processed the purge and compaction has run (while
allowing for up to purge_infos_limit requests to remain available in
perpetuity).

COUCHDB-3326

Co-authored-by: Mayya Sharipova <mayyas@ca.ibm.com>
Co-authored-by: jiangphcn <jiangph@cn.ibm.com>

4 files changed, 315 insertions, 32 deletions

diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index ed0adead9..b9652bc0f 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -496,24 +496,20 @@ db_req(#httpd{path_parts=[_, <<"_bulk_get">>]}=Req, _Db) ->
 
 
 db_req(#httpd{method='POST',path_parts=[_,<<"_purge">>]}=Req, Db) ->
+    couch_stats:increment_counter([couchdb, httpd, purge_requests]),
     chttpd:validate_ctype(Req, "application/json"),
+    W = chttpd:qs_value(Req, "w", integer_to_list(mem3:quorum(Db))),
+    Options = [{user_ctx, Req#httpd.user_ctx}, {w, W}],
     {IdsRevs} = chttpd:json_body_obj(Req),
     IdsRevs2 = [{Id, couch_doc:parse_revs(Revs)} || {Id, Revs} <- IdsRevs],
-    case fabric:purge_docs(Db, IdsRevs2) of
-    {ok, PurgeSeq, PurgedIdsRevs} ->
-        PurgedIdsRevs2 = [{Id, couch_doc:revs_to_strs(Revs)} || {Id, Revs}
-            <- PurgedIdsRevs],
-        send_json(Req, 200, {[
-            {<<"purge_seq">>, PurgeSeq},
-            {<<"purged">>, {PurgedIdsRevs2}}
-        ]});
-    Error ->
-        throw(Error)
-    end;
+    {ok, Results} = fabric:purge_docs(Db, IdsRevs2, Options),
+    {Code, Json} = purge_results_to_json(IdsRevs2, Results),
+    send_json(Req, Code, {[{<<"purge_seq">>, null}, {<<"purged">>, {Json}}]});
 
 db_req(#httpd{path_parts=[_,<<"_purge">>]}=Req, _Db) ->
     send_method_not_allowed(Req, "POST");
 
+
 db_req(#httpd{method='GET',path_parts=[_,OP]}=Req, Db) when ?IS_ALL_DOCS(OP) ->
     case chttpd:qs_json_value(Req, "keys", nil) of
     Keys when is_list(Keys) ->
@@ -623,6 +619,20 @@ db_req(#httpd{method='GET',path_parts=[_,<<"_revs_limit">>]}=Req, Db) ->
 db_req(#httpd{path_parts=[_,<<"_revs_limit">>]}=Req, _Db) ->
     send_method_not_allowed(Req, "PUT,GET");
 
+db_req(#httpd{method='PUT',path_parts=[_,<<"_purged_infos_limit">>]}=Req, Db) ->
+    Limit = chttpd:json_body(Req),
+    Options = [{user_ctx, Req#httpd.user_ctx}],
+    case chttpd:json_body(Req) of
+        Limit when is_integer(Limit), Limit > 0 ->
+            ok = fabric:set_purge_infos_limit(Db, Limit, Options),
+            send_json(Req, {[{<<"ok">>, true}]});
+        _->
+            throw({bad_request, "`purge_infos_limit` must be positive integer"})
+    end;
+
+db_req(#httpd{method='GET',path_parts=[_,<<"_purged_infos_limit">>]}=Req, Db) ->
+    send_json(Req, fabric:get_purge_infos_limit(Db));
+
 % Special case to enable using an unencoded slash in the URL of design docs,
 % as slashes in document IDs must otherwise be URL encoded.
 db_req(#httpd{method='GET', mochi_req=MochiReq, path_parts=[_DbName, <<"_design/", _/binary>> | _]}=Req, _Db) ->
@@ -993,6 +1003,20 @@ update_doc_result_to_json(DocId, Error) ->
     {_Code, ErrorStr, Reason} = chttpd:error_info(Error),
     {[{id, DocId}, {error, ErrorStr}, {reason, Reason}]}.
 
+purge_results_to_json([], []) ->
+    {201, []};
+purge_results_to_json([{DocId, _Revs} | RIn], [{ok, PRevs} | ROut]) ->
+    {Code, Results} = purge_results_to_json(RIn, ROut),
+    {Code, [{DocId, couch_doc:revs_to_strs(PRevs)} | Results]};
+purge_results_to_json([{DocId, _Revs} | RIn], [{accepted, PRevs} | ROut]) ->
+    {Code, Results} = purge_results_to_json(RIn, ROut),
+    NewResults = [{DocId, couch_doc:revs_to_strs(PRevs)} | Results],
+    {erlang:max(Code, 202), NewResults};
+purge_results_to_json([{DocId, _Revs} | RIn], [Error | ROut]) ->
+    {Code, Results} = purge_results_to_json(RIn, ROut),
+    {NewCode, ErrorStr, Reason} = chttpd:error_info(Error),
+    NewResults = [{DocId, {[{error, ErrorStr}, {reason, Reason}]}} | Results],
+    {erlang:max(NewCode, Code), NewResults}.
 
 send_updated_doc(Req, Db, DocId, Json) ->
     send_updated_doc(Req, Db, DocId, Json, []).
diff --git a/src/chttpd/test/chttpd_purge_tests.erl b/src/chttpd/test/chttpd_purge_tests.erl
new file mode 100644
index 000000000..04456cb2b
--- /dev/null
+++ b/src/chttpd/test/chttpd_purge_tests.erl
@@ -0,0 +1,269 @@
+% 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(chttpd_purge_tests).
+
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+
+-define(USER, "chttpd_db_test_admin").
+-define(PASS, "pass").
+-define(AUTH, {basic_auth, {?USER, ?PASS}}).
+-define(CONTENT_JSON, {"Content-Type", "application/json"}).
+
+
+setup() ->
+    ok = config:set("admins", ?USER, ?PASS, _Persist=false),
+    TmpDb = ?tempdb(),
+    Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
+    Port = mochiweb_socket_server:get(chttpd, port),
+    Url = lists:concat(["http://", Addr, ":", Port, "/", ?b2l(TmpDb)]),
+    create_db(Url),
+    Url.
+
+
+teardown(Url) ->
+    delete_db(Url),
+    ok = config:delete("admins", ?USER, _Persist=false).
+
+
+create_db(Url) ->
+    {ok, Status, _, _} = test_request:put(Url, [?CONTENT_JSON, ?AUTH], "{}"),
+    ?assert(Status =:= 201 orelse Status =:= 202).
+
+
+create_doc(Url, Id) ->
+    test_request:put(Url ++ "/" ++ Id,
+        [?CONTENT_JSON, ?AUTH], "{\"mr\": \"rockoartischocko\"}").
+
+create_doc(Url, Id, Content) ->
+    test_request:put(Url ++ "/" ++ Id,
+        [?CONTENT_JSON, ?AUTH], "{\"mr\": \"" ++ Content ++ "\"}").
+
+
+delete_db(Url) ->
+    {ok, 200, _, _} = test_request:delete(Url, [?AUTH]).
+
+
+purge_test_() ->
+    {
+        "chttpd db tests",
+        {
+            setup,
+            fun chttpd_test_util:start_couch/0,
+            fun chttpd_test_util:stop_couch/1,
+            {
+                foreach,
+                fun setup/0,
+                fun teardown/1,
+                [
+                    fun test_empty_purge_request/1,
+                    fun test_ok_purge_request/1,
+                    fun test_partial_purge_request/1,
+                    fun test_mixed_purge_request/1,
+                    fun test_exceed_limits_on_purge_infos/1,
+                    fun should_error_set_purged_docs_limit_to0/1
+                ]
+            }
+        }
+    }.
+
+
+test_empty_purge_request(Url) ->
+    ?_test(begin
+        IdsRevs = "{}",
+        {ok, Status, _, ResultBody} = test_request:post(Url ++ "/_purge/",
+            [?CONTENT_JSON, ?AUTH], IdsRevs),
+        ResultJson = ?JSON_DECODE(ResultBody),
+        ?assert(Status =:= 201 orelse Status =:= 202),
+        ?assertEqual(
+                {[
+                    {<<"purge_seq">>, null},
+                    {<<"purged">>,{[]}}
+                ]},
+                ResultJson
+            )
+    end).
+
+
+test_ok_purge_request(Url) ->
+    ?_test(begin
+        {ok, _, _, Body} = create_doc(Url, "doc1"),
+        {Json} = ?JSON_DECODE(Body),
+        Rev1 = couch_util:get_value(<<"rev">>, Json, undefined),
+        {ok, _, _, Body2} = create_doc(Url, "doc2"),
+        {Json2} = ?JSON_DECODE(Body2),
+        Rev2 = couch_util:get_value(<<"rev">>, Json2, undefined),
+        {ok, _, _, Body3} = create_doc(Url, "doc3"),
+        {Json3} = ?JSON_DECODE(Body3),
+        Rev3 = couch_util:get_value(<<"rev">>, Json3, undefined),
+
+        IdsRevsEJson = {[
+            {<<"doc1">>, [Rev1]},
+            {<<"doc2">>, [Rev2]},
+            {<<"doc3">>, [Rev3]}
+        ]},
+        IdsRevs = binary_to_list(?JSON_ENCODE(IdsRevsEJson)),
+
+        {ok, Status, _, ResultBody} = test_request:post(Url ++ "/_purge/",
+            [?CONTENT_JSON, ?AUTH], IdsRevs),
+        ResultJson = ?JSON_DECODE(ResultBody),
+        ?assert(Status =:= 201 orelse Status =:= 202),
+        ?assertEqual(
+                {[
+                    {<<"purge_seq">>, null},
+                    {<<"purged">>, {[
+                        {<<"doc1">>, [Rev1]},
+                        {<<"doc2">>, [Rev2]},
+                        {<<"doc3">>, [Rev3]}
+                    ]}}
+                ]},
+                ResultJson
+            )
+    end).
+
+
+test_partial_purge_request(Url) ->
+    ?_test(begin
+        {ok, _, _, Body} = create_doc(Url, "doc1"),
+        {Json} = ?JSON_DECODE(Body),
+        Rev1 = couch_util:get_value(<<"rev">>, Json, undefined),
+
+        NewDoc = "{\"new_edits\": false, \"docs\": [{\"_id\": \"doc1\",
+            \"_revisions\": {\"start\": 1, \"ids\": [\"12345\", \"67890\"]},
+            \"content\": \"updated\", \"_rev\": \"" ++ ?b2l(Rev1) ++ "\"}]}",
+        {ok, _, _, _} = test_request:post(Url ++ "/_bulk_docs/",
+            [?CONTENT_JSON, ?AUTH], NewDoc),
+
+        IdsRevsEJson = {[{<<"doc1">>, [Rev1]}]},
+        IdsRevs = binary_to_list(?JSON_ENCODE(IdsRevsEJson)),
+        {ok, Status, _, ResultBody} = test_request:post(Url ++ "/_purge/",
+            [?CONTENT_JSON, ?AUTH], IdsRevs),
+        ResultJson = ?JSON_DECODE(ResultBody),
+        ?assert(Status =:= 201 orelse Status =:= 202),
+        ?assertEqual(
+            {[
+                {<<"purge_seq">>, null},
+                {<<"purged">>, {[
+                    {<<"doc1">>, [Rev1]}
+                ]}}
+            ]},
+            ResultJson
+        ),
+        {ok, Status2, _, ResultBody2} = test_request:get(Url
+            ++ "/doc1/", [?AUTH]),
+        {Json2} = ?JSON_DECODE(ResultBody2),
+        Content = couch_util:get_value(<<"content">>, Json2, undefined),
+        ?assertEqual(<<"updated">>, Content),
+        ?assert(Status2 =:= 200)
+    end).
+
+
+test_mixed_purge_request(Url) ->
+    ?_test(begin
+        {ok, _, _, Body} = create_doc(Url, "doc1"),
+        {Json} = ?JSON_DECODE(Body),
+        Rev1 = couch_util:get_value(<<"rev">>, Json, undefined),
+
+        NewDoc = "{\"new_edits\": false, \"docs\": [{\"_id\": \"doc1\",
+            \"_revisions\": {\"start\": 1, \"ids\": [\"12345\", \"67890\"]},
+            \"content\": \"updated\", \"_rev\": \"" ++ ?b2l(Rev1) ++ "\"}]}",
+        {ok, _, _, _} = test_request:post(Url ++ "/_bulk_docs/",
+            [?CONTENT_JSON, ?AUTH], NewDoc),
+
+        {ok, _, _, _Body2} = create_doc(Url, "doc2", "content2"),
+        {ok, _, _, Body3} = create_doc(Url, "doc3", "content3"),
+        {Json3} = ?JSON_DECODE(Body3),
+        Rev3 = couch_util:get_value(<<"rev">>, Json3, undefined),
+
+
+        IdsRevsEJson = {[
+            {<<"doc1">>, [Rev1]},  % partial purge
+            {<<"doc2">>, [Rev3]},  % correct format, but invalid rev
+            {<<"doc3">>, [Rev3]}   % correct format and rev
+        ]},
+        IdsRevs = binary_to_list(?JSON_ENCODE(IdsRevsEJson)),
+        {ok, Status, _, Body4} = test_request:post(Url ++ "/_purge/",
+            [?CONTENT_JSON, ?AUTH], IdsRevs),
+        ResultJson = ?JSON_DECODE(Body4),
+        ?assert(Status =:= 201 orelse Status =:= 202),
+        ?assertEqual(
+            {[
+                {<<"purge_seq">>, null},
+                {<<"purged">>, {[
+                    {<<"doc1">>, [Rev1]},
+                    {<<"doc2">>, []},
+                    {<<"doc3">>, [Rev3]}
+                ]}}
+            ]},
+            ResultJson
+        ),
+        {ok, Status2, _, Body5} = test_request:get(Url
+            ++ "/doc1/", [?AUTH]),
+        {Json5} = ?JSON_DECODE(Body5),
+        Content = couch_util:get_value(<<"content">>, Json5, undefined),
+        ?assertEqual(<<"updated">>, Content),
+        ?assert(Status2 =:= 200)
+    end).
+
+
+test_exceed_limits_on_purge_infos(Url) ->
+    ?_test(begin
+        {ok, Status1, _, _} = test_request:put(Url ++ "/_purged_infos_limit/",
+            [?CONTENT_JSON, ?AUTH], "2"),
+        ?assert(Status1 =:= 200),
+
+        {ok, _, _, Body} = create_doc(Url, "doc1"),
+        {Json} = ?JSON_DECODE(Body),
+        Rev1 = couch_util:get_value(<<"rev">>, Json, undefined),
+        {ok, _, _, Body2} = create_doc(Url, "doc2"),
+        {Json2} = ?JSON_DECODE(Body2),
+        Rev2 = couch_util:get_value(<<"rev">>, Json2, undefined),
+        {ok, _, _, Body3} = create_doc(Url, "doc3"),
+        {Json3} = ?JSON_DECODE(Body3),
+        Rev3 = couch_util:get_value(<<"rev">>, Json3, undefined),
+
+        IdsRevsEJson = {[
+            {<<"doc1">>, [Rev1]},
+            {<<"doc2">>, [Rev2]},
+            {<<"doc3">>, [Rev3]}
+        ]},
+        IdsRevs = binary_to_list(?JSON_ENCODE(IdsRevsEJson)),
+
+        {ok, Status2, _, ResultBody} = test_request:post(Url ++ "/_purge/",
+            [?CONTENT_JSON, ?AUTH], IdsRevs),
+
+        ResultJson = ?JSON_DECODE(ResultBody),
+        ?assert(Status2 =:= 201 orelse Status2 =:= 202),
+        ?assertEqual(
+            {[
+                {<<"purge_seq">>, null},
+                {<<"purged">>, {[
+                    {<<"doc1">>, [Rev1]},
+                    {<<"doc2">>, [Rev2]},
+                    {<<"doc3">>, [Rev3]}
+                ]}}
+            ]},
+            ResultJson
+        )
+
+    end).
+
+
+should_error_set_purged_docs_limit_to0(Url) ->
+    ?_test(begin
+        {ok, Status, _, _} = test_request:put(Url ++ "/_purged_infos_limit/",
+            [?CONTENT_JSON, ?AUTH], "0"),
+        ?assert(Status =:= 400)
+    end).
\ No newline at end of file
diff --git a/test/javascript/tests/erlang_views.js b/test/javascript/tests/erlang_views.js
index ec78e6506..9b15e1043 100644
--- a/test/javascript/tests/erlang_views.js
+++ b/test/javascript/tests/erlang_views.js
@@ -56,7 +56,7 @@ couchTests.erlang_views = function(debug) {
             '  {Info} = couch_util:get_value(<<"info">>, Req, {[]}), ' +
             '  Purged = couch_util:get_value(<<"purge_seq">>, Info, -1), ' +
             '  Verb = couch_util:get_value(<<"method">>, Req, <<"not_get">>), ' +
-            '  R = list_to_binary(io_lib:format("~b - ~s", [Purged, Verb])), ' +
+            '  R = list_to_binary(io_lib:format("~s - ~s", [Purged, Verb])), ' +
             '  {[{<<"code">>, 200}, {<<"headers">>, {[]}}, {<<"body">>, R}]} ' +
             'end.'
         },
@@ -85,7 +85,8 @@ couchTests.erlang_views = function(debug) {
       var url = "/" + db_name + "/_design/erlview/_show/simple/1";
       var xhr = CouchDB.request("GET", url);
       T(xhr.status == 200, "standard get should be 200");
-      T(xhr.responseText == "0 - GET");
+      T(/0-/.test(xhr.responseText));
+      T(/- GET/.test(xhr.responseText));
 
       var url = "/" + db_name + "/_design/erlview/_list/simple_list/simple_view";
       var xhr = CouchDB.request("GET", url);
diff --git a/test/javascript/tests/purge.js b/test/javascript/tests/purge.js
index 38eca8d28..0c11d9ad8 100644
--- a/test/javascript/tests/purge.js
+++ b/test/javascript/tests/purge.js
@@ -11,7 +11,6 @@
 // the License.
 
 couchTests.purge = function(debug) {
-  return console.log('TODO: this feature is not yet implemented');
   var db_name = get_random_db_name();
   var db = new CouchDB(db_name, {"X-Couch-Full-Commit":"false"});
   db.createDb();
@@ -53,21 +52,13 @@ couchTests.purge = function(debug) {
   var xhr = CouchDB.request("POST", "/" + db_name + "/_purge", {
     body: JSON.stringify({"1":[doc1._rev], "2":[doc2._rev]})
   });
-  console.log(xhr.status);
-  console.log(xhr.responseText);
-  T(xhr.status == 200);
+  T(xhr.status == 201);
 
   var result = JSON.parse(xhr.responseText);
   var newInfo = db.info();
-  
-  // purging increments the update sequence
-  T(info.update_seq+1 == newInfo.update_seq);
-  // and it increments the purge_seq
-  T(info.purge_seq+1 == newInfo.purge_seq);
-  T(result.purge_seq == newInfo.purge_seq);
 
-  T(result.purged["1"][0] == doc1._rev);
-  T(result.purged["2"][0] == doc2._rev);
+  T(result.purged["1"] == doc1._rev);
+  T(result.purged["2"] == doc2._rev);
 
   T(db.open("1") == null);
   T(db.open("2") == null);
@@ -85,7 +76,6 @@ couchTests.purge = function(debug) {
   // compaction isn't instantaneous, loop until done
   while (db.info().compact_running) {};
   var compactInfo = db.info();
-  T(compactInfo.purge_seq == newInfo.purge_seq);
 
   // purge documents twice in a row without loading views
   // (causes full view rebuilds)
@@ -97,15 +87,14 @@ couchTests.purge = function(debug) {
     body: JSON.stringify({"3":[doc3._rev]})
   });
 
-  T(xhr.status == 200);
+  T(xhr.status == 201);
 
   xhr = CouchDB.request("POST", "/" + db_name + "/_purge", {
     body: JSON.stringify({"4":[doc4._rev]})
   });
 
-  T(xhr.status == 200);
+  T(xhr.status == 201);
   result = JSON.parse(xhr.responseText);
-  T(result.purge_seq == db.info().purge_seq);
 
   var rows = db.view("test/all_docs_twice").rows;
   for (var i = 4; i < numDocs; i++) {
@@ -129,7 +118,7 @@ couchTests.purge = function(debug) {
   var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
     body: JSON.stringify({"test":[docA._rev]})
   });
-  TEquals(200, xhr.status, "single rev purge after replication succeeds");
+  TEquals(201, xhr.status, "single rev purge after replication succeeds");
 
   var xhr = CouchDB.request("GET", "/" + dbB.name + "/test?rev=" + docA._rev);
   TEquals(404, xhr.status, "single rev purge removes revision");
@@ -137,14 +126,14 @@ couchTests.purge = function(debug) {
   var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
     body: JSON.stringify({"test":[docB._rev]})
   });
-  TEquals(200, xhr.status, "single rev purge after replication succeeds");
+  TEquals(201, xhr.status, "single rev purge after replication succeeds");
   var xhr = CouchDB.request("GET", "/" + dbB.name + "/test?rev=" + docB._rev);
   TEquals(404, xhr.status, "single rev purge removes revision");
 
   var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
     body: JSON.stringify({"test":[docA._rev, docB._rev]})
   });
-  TEquals(200, xhr.status, "all rev purge after replication succeeds");
+  TEquals(201, xhr.status, "all rev purge after replication succeeds");
 
   // cleanup
   db.deleteDb();