Add response compression to replicator

rel/overlay/etc/default.ini

@@ -744,6 +744,9 @@ partitioned||* = true
 ; Enable gzip only when the target supports Content-Encoding: gzip on inbound requests.
 ;request_compression = none
 ;compress_min_size = 1024
+; The replicator always sends Accept-Encoding: gzip on outbound requests so the
+; remote server can compress its responses. Decompression is automatic.
+; Monitor with: _stats/couch_replicator/responses_decompressed
 
 
 ; Some socket options that might boost performance in some scenarios:

src/couch_replicator/priv/stats_descriptions.cfg

@@ -151,3 +151,8 @@
     {type, counter},
     {desc, <<"number of HTTP requests compressed with gzip by the replicator">>}
 ]}.
+
+{[couch_replicator, responses_decompressed, gzip], [
+    {type, counter},
+    {desc, <<"number of HTTP responses decompressed with gzip by the replicator">>}
+]}.

src/couch_replicator/src/couch_replicator_api_wrap.erl

@@ -179,7 +179,11 @@ get_missing_revs(#httpdb{} = Db, IdRevs) ->
             {method, post},
             {path, "_revs_diff"},
             {body, Body},
-            {headers, [{"Content-Type", "application/json"} | ExtraHeaders]}
+            {headers, [
+                {"Content-Type", "application/json"},
+                {"Accept-Encoding", "gzip"}
+                | ExtraHeaders
+            ]}
         ],
         fun
             (200, _, {Props}) ->
@@ -222,7 +226,8 @@ bulk_get(#httpdb{} = Db, #{} = IdRevs, Options) ->
         {body, ReqBody},
         {headers, [
             {"Content-Type", "application/json"},
-            {"Accept", "application/json"}
+            {"Accept", "application/json"},
+            {"Accept-Encoding", "gzip"}
             | ExtraHeaders
         ]}
     ],
@@ -507,7 +512,8 @@ update_docs(#httpdb{} = HttpDb, DocList, Options, UpdateType) ->
     end,
     Headers0 = [
         {"Content-Type", "application/json"},
-        {"X-Couch-Full-Commit", FullCommit}
+        {"X-Couch-Full-Commit", FullCommit},
+        {"Accept-Encoding", "gzip"}
     ],
     {Body, Headers} =
         case should_compress_request(HttpDb, Len) of

src/couch_replicator/src/couch_replicator_httpc.erl

@@ -194,8 +194,9 @@ process_response({ok, Code, Headers, Body}, Worker, HttpDb, Params, Callback) ->
         Ok when Ok >= 200, Ok < 500 ->
             backoff_success(HttpDb, Params),
             couch_stats:increment_counter([couch_replicator, responses, success]),
+            Body1 = maybe_decompress_response(Headers, Body),
             EJson =
-                case Body of
+                case Body1 of
                     <<>> ->
                         null;
                     Json ->
@@ -265,6 +266,17 @@ process_stream_response(ReqId, Worker, HttpDb, Params, Callback) ->
         maybe_retry(timeout, Worker, HttpDb, Params)
     end.
 
+maybe_decompress_response(_Headers, <<>>) ->
+    <<>>;
+maybe_decompress_response(Headers, Body) ->
+    case lists:keyfind("content-encoding", 1, [{string:to_lower(K), V} || {K, V} <- Headers]) of
+        {_, "gzip"} ->
+            couch_stats:increment_counter([couch_replicator, responses_decompressed, gzip]),
+            zlib:gunzip(Body);
+        _ ->
+            Body
+    end.
+
 process_auth_response(HttpDb, Code, Headers, Params) ->
     case couch_replicator_auth:handle_response(HttpDb, Code, Headers) of
         {continue, HttpDb1} ->

src/couch_replicator/test/eunit/couch_replicator_compression_tests.erl

@@ -31,21 +31,22 @@ compression_test_() ->
                 ?TDEF_FE(should_compress_when_enabled, ?TIMEOUT_EUNIT),
                 ?TDEF_FE(should_compress_large_batch, ?TIMEOUT_EUNIT),
                 ?TDEF_FE(should_compress_per_job, ?TIMEOUT_EUNIT),
-                ?TDEF_FE(job_compression_overrides_global_disabled, ?TIMEOUT_EUNIT)
+                ?TDEF_FE(job_compression_overrides_global_disabled, ?TIMEOUT_EUNIT),
+                ?TDEF_FE(decompress_counter_increments_on_replication, ?TIMEOUT_EUNIT)
             ]
         }
     }.
 
 setup() ->
-    Ctx = couch_replicator_test_helper:test_setup(),
+    {Ctx, {Source, Target}} = couch_replicator_test_helper:test_setup(),
     config:set("replicator", "request_compression", "none", false),
     config:set("replicator", "compress_min_size", "1024", false),
-    Ctx.
+    {Ctx, {Source, Target}}.
 
-teardown(Ctx) ->
+teardown({Ctx, {Source, Target}}) ->
     config:delete("replicator", "request_compression", false),
     config:delete("replicator", "compress_min_size", false),
-    couch_replicator_test_helper:test_teardown(Ctx).
+    couch_replicator_test_helper:test_teardown({Ctx, {Source, Target}}).
 
 should_not_compress_by_default({_Ctx, {Source, Target}}) ->
     Before = couch_stats:sample([couch_replicator, requests_compressed, gzip]),
@@ -111,6 +112,14 @@ populate_db(DbName, Count) ->
     {ok, _} = fabric:update_docs(DbName, Docs, [?ADMIN_CTX]),
     ok.
 
+decompress_counter_increments_on_replication({_Ctx, {Source, Target}}) ->
+    populate_db(Source, ?DOCS_COUNT),
+    Before = couch_stats:sample([couch_replicator, responses_decompressed, gzip]),
+    replicate(Source, Target),
+    couch_replicator_test_helper:cluster_compare_dbs(Source, Target),
+    After = couch_stats:sample([couch_replicator, responses_decompressed, gzip]),
+    ?assert(After > Before).
+
 replicate(Source, Target) ->
     replicate_with_options(Source, Target, []).
 

src/docs/src/config/replicator.rst

@@ -139,6 +139,9 @@ Replicator Database Configuration
         ``"request_compression": "gzip"`` in the replication document or
         ``_replicate`` request body, which overrides the global setting.
 
+        Monitor the number of compressed requests via the
+        ``couch_replicator.requests_compressed.gzip`` stat.
+
     .. config:option:: compress_min_size :: Minimum body size for compression
 
         .. versionadded:: 3.6
@@ -151,6 +154,19 @@ Replicator Database Configuration
             [replicator]
             compress_min_size = 1024
 
+    .. config:option:: response_compression :: Decompress inbound response bodies
+
+        .. versionadded:: 3.6
+
+        Decompress gzip-encoded inbound response bodies (``_bulk_docs``,
+        ``_revs_diff``, ``_bulk_get``) received from the replication source.
+        The replicator always sends ``Accept-Encoding: gzip`` and automatically
+        decompresses gzip responses received from the remote server. This
+        behaviour is always enabled.
+
+        Monitor the number of decompressed responses via the
+        ``couch_replicator.responses_decompressed.gzip`` stat.
+
     .. config:option:: retries_per_request :: Number of retries per request
 
         .. versionchanged:: 2.1.1