Resolve comments: gzip only, simplify _bulk_docs body handling

Author: lacklacklack

rel/overlay/etc/default.ini

@@ -724,11 +724,12 @@ partitioned||* = true
 ;   *.example.com:443:[2001:db8::1]:443
 ;connect_to =
 
-; Compression settings for replication
-;compress_requests = true
+; Compress outbound replication request bodies (_bulk_docs, _revs_diff) with gzip.
+; Disabled by default. Only gzip is supported. Enable only when talking to CouchDB
+; servers that support gzip Content-Encoding on inbound requests.
+;compress_requests = false
 ;compress_min_size = 1024
 ;compression_algorithm = gzip
-;accept_encodings = gzip, deflate
 
 
 ; Some socket options that might boost performance in some scenarios:

src/couch_replicator/COMPRESSION_CONFIG.md

@@ -1,32 +1,16 @@
-# CouchDB Replicator Compression
+# CouchDB Replicator Request Compression
 
-## Overview
+The replicator can optionally gzip-compress outbound request bodies for
+`_bulk_docs` and `_revs_diff`. This reduces bandwidth during replication.
+CouchDB already supports `Content-Encoding: gzip` on inbound requests, so no
+server-side changes are needed.
 
-The replicator now supports configurable HTTP compression to reduce bandwidth during replication.
-
-## Configuration
+Compression is disabled by default. Relevant `[replicator]` config keys:
 
 ```ini
 [replicator]
-; Enable compression (default: true)
-compress_requests = true
-
-; Minimum body size to compress in bytes (default: 1024)
-compress_min_size = 1024
-
-; Algorithm: gzip (default), deflate
-compression_algorithm = gzip
-
-; Accept these encodings in responses
-accept_encodings = gzip, deflate
+compress_requests = false
+compress_min_size = 1024    ; minimum body size in bytes before compressing
 ```
 
-## Algorithms
-
-- **gzip** (default): Best compatibility, built-in to Erlang
-- **deflate**: Built-in to Erlang, slightly faster than gzip
-
-## Statistics
-
-- `couch_replicator.requests.compressed` - Total compressed requests
-- `couch_replicator.requests.compressed.{algorithm}` - Per-algorithm stats
+Metric: `couch_replicator.requests_compressed.gzip` — number of gzip-compressed requests sent.

src/couch_replicator/priv/stats_descriptions.cfg

@@ -147,15 +147,7 @@
     {desc, <<"number of times DNS overrides were applied to replication requests">>}
 ]}.
 
-{[couch_replicator, requests_compressed], [
-    {type, counter},
-    {desc, <<"number of HTTP requests compressed by the replicator">>}
-]}.
 {[couch_replicator, requests_compressed, gzip], [
     {type, counter},
     {desc, <<"number of HTTP requests compressed with gzip by the replicator">>}
 ]}.
-{[couch_replicator, requests_compressed, deflate], [
-    {type, counter},
-    {desc, <<"number of HTTP requests compressed with deflate by the replicator">>}
-]}.

src/couch_replicator/src/couch_replicator_api_wrap.erl

@@ -171,13 +171,14 @@ ensure_full_commit(#httpdb{} = Db) ->
 
 get_missing_revs(#httpdb{} = Db, IdRevs) ->
     JsonBody = {[{Id, couch_doc:revs_to_strs(Revs)} || {Id, Revs} <- IdRevs]},
+    {Body, ExtraHeaders} = maybe_compress(?JSON_ENCODE(JsonBody)),
     send_req(
         Db,
         [
             {method, post},
             {path, "_revs_diff"},
-            {body, ?JSON_ENCODE(JsonBody)},
-            {headers, [{"Content-Type", "application/json"}]}
+            {body, Body},
+            {headers, [{"Content-Type", "application/json"} | ExtraHeaders]}
         ],
         fun
             (200, _, {Props}) ->
@@ -477,40 +478,35 @@ update_docs(#httpdb{} = HttpDb, DocList, Options, UpdateType) ->
     % Note: nginx and other servers don't like PUT/POST requests without
     % a Content-Length header, so we can't do a chunked transfer encoding
     % and JSON encode each doc only before sending it through the socket.
-    {Docs, Len} = lists:mapfoldl(
+    {Docs, _} = lists:mapfoldl(
         fun
             (#doc{} = Doc, Acc) ->
                 Json = ?JSON_ENCODE(couch_doc:to_json_obj(Doc, [revs, attachments])),
                 {Json, Acc + iolist_size(Json)};
             (Doc, Acc) ->
                 {Doc, Acc + iolist_size(Doc)}
         end,
-        byte_size(Prefix) + byte_size(Suffix) + length(DocList) - 1,
+        0,
         DocList
     ),
-    BodyFun = fun
-        (eof) ->
-            eof;
-        ([]) ->
-            {ok, Suffix, eof};
-        ([prefix | Rest]) ->
-            {ok, Prefix, Rest};
-        ([Doc]) ->
-            {ok, Doc, []};
-        ([Doc | RestDocs]) ->
-            {ok, [Doc, ","], RestDocs}
-    end,
+    % Collect the full body into a binary so we can optionally gzip it.
+    % Content-Length is required (nginx etc. reject chunked PUT/POST).
+    RawBody = iolist_to_binary(
+        [Prefix, lists:join(",", Docs), Suffix]
+    ),
+    {Body, ExtraHeaders} = maybe_compress(RawBody),
     Headers = [
-        {"Content-Length", Len},
+        {"Content-Length", byte_size(Body)},
         {"Content-Type", "application/json"},
         {"X-Couch-Full-Commit", FullCommit}
+        | ExtraHeaders
     ],
     send_req(
         HttpDb,
         [
             {method, post},
             {path, "_bulk_docs"},
-            {body, {BodyFun, [prefix | Docs]}},
+            {body, Body},
             {headers, Headers}
         ],
         fun
@@ -1052,6 +1048,33 @@ header_value(Key, Headers, Default) ->
         _ ->
             Default
     end.
+    
+%% Compress Body with gzip if enabled and body is large enough.
+%% Returns {Body, ExtraHeaders} where ExtraHeaders may contain Content-Encoding.
+maybe_compress(Body) when is_binary(Body) ->
+    case config:get_boolean("replicator", "compress_requests", false) of
+        true ->
+            Algorithm = config:get("replicator", "compression_algorithm", "gzip"),
+            MinSize = config:get_integer("replicator", "compress_min_size", 1024),
+            case byte_size(Body) >= MinSize of
+                true -> compress_with(Algorithm, Body);
+                false -> {Body, []}
+            end;
+        false ->
+            {Body, []}
+    end.
+
+compress_with("gzip", Body) ->
+    Compressed = zlib:gzip(Body),
+    couch_stats:increment_counter([couch_replicator, requests_compressed]),
+    couch_stats:increment_counter([couch_replicator, requests_compressed, gzip]),
+    {Compressed, [{"Content-Encoding", "gzip"}]};
+compress_with(Other, Body) ->
+    couch_log:warning(
+        "~p: unsupported compression_algorithm ~p, skipping compression",
+        [?MODULE, Other]
+    ),
+    {Body, []}.
 
 % Normalize an #httpdb{} or #db{} record such that it can be used for
 % comparisons. This means remove things like pids and also sort options / props.

src/couch_replicator/src/couch_replicator_httpc.erl

@@ -41,80 +41,6 @@
 % where we may end up processing an unbounded number of messages.
 -define(MAX_DISCARDED_MESSAGES, 100).
 
-should_compress_request(Body) when is_binary(Body) ->
-    MinSize = config:get_integer("replicator", "compress_min_size", 1024),
-    byte_size(Body) >= MinSize;
-should_compress_request(Body) when is_list(Body) ->
-    should_compress_request(iolist_to_binary(Body));
-should_compress_request(_) ->
-    false.
-
-get_compression_algorithm() ->
-    % Supported: gzip (default), deflate
-    Algorithm = config:get("replicator", "compression_algorithm", "gzip"),
-    case Algorithm of
-        "gzip" -> gzip;
-        "deflate" -> deflate;
-        _ ->
-            couch_log:warning(
-                "couch_replicator_httpc: Unknown compression algorithm ~p, using gzip",
-                [Algorithm]
-            ),
-            gzip
-    end.
-
-compress_body(Body) when is_binary(Body) ->
-    compress_body_with_algorithm(Body, get_compression_algorithm());
-compress_body(Body) when is_list(Body) ->
-    compress_body(iolist_to_binary(Body));
-compress_body(Body) ->
-    Body.
-
-compress_body_with_algorithm(Body, gzip) ->
-    zlib:gzip(Body);
-compress_body_with_algorithm(Body, deflate) ->
-    zlib:compress(Body).
-
-get_content_encoding(gzip) -> "gzip";
-get_content_encoding(deflate) -> "deflate".
-
-decompress_body(Headers, Body) ->
-    case lists:keyfind("Content-Encoding", 1, Headers) of
-        {"Content-Encoding", Encoding} ->
-            decompress_body_with_encoding(Encoding, Body);
-        _ ->
-            Body
-    end.
-
-decompress_body_with_encoding("gzip", Body) ->
-    try
-        zlib:gunzip(Body)
-    catch
-        error:data_error ->
-            couch_log:warning(
-                "couch_replicator_httpc: Failed to decompress gzip response, using original",
-                []
-            ),
-            Body
-    end;
-decompress_body_with_encoding("deflate", Body) ->
-    try
-        zlib:uncompress(Body)
-    catch
-        error:data_error ->
-            couch_log:warning(
-                "couch_replicator_httpc: Failed to decompress deflate response, using original",
-                []
-            ),
-            Body
-    end;
-decompress_body_with_encoding(Other, Body) ->
-    couch_log:warning(
-        "couch_replicator_httpc: Unknown content encoding ~p, using original body",
-        [Other]
-    ),
-    Body.
-
 setup(Db) ->
     #httpdb{
         httpc_pool = nil,
@@ -183,36 +109,11 @@ stop_http_worker() ->
 
 send_ibrowse_req(#httpdb{headers = BaseHeaders} = HttpDb0, Params) ->
     Method = get_value(method, Params, get),
-    UserHeaders0 = get_value(headers, Params, []),
-    % Accept multiple compression algorithms
-    AcceptEncodings = config:get("replicator", "accept_encodings", "gzip, deflate, zstd"),
-    UserHeaders1 = case lists:keyfind("Accept-Encoding", 1, UserHeaders0) of
-        false -> [{"Accept-Encoding", AcceptEncodings} | UserHeaders0];
-        _ -> UserHeaders0
-    end,
-    Body0 = get_value(body, Params, []),
-    CompressEnabled = config:get_boolean("replicator", "compress_requests", true),
-    ShouldCompress = should_compress_request(Body0),
-    {Body, UserHeaders2} = case CompressEnabled andalso ShouldCompress of
-        true ->
-            Algorithm = get_compression_algorithm(),
-            CompressedBody = compress_body(Body0),
-            ContentEncoding = get_content_encoding(Algorithm),
-            UpdatedHeaders = case lists:keyfind("Content-Encoding", 1, UserHeaders1) of
-                false -> [{"Content-Encoding", ContentEncoding} | UserHeaders1];
-                _ -> UserHeaders1
-            end,
-            % Track compression algorithm usage
-            couch_stats:increment_counter([couch_replicator, requests_compressed]),
-            couch_stats:increment_counter([couch_replicator, requests_compressed, Algorithm]),
-            {CompressedBody, UpdatedHeaders};
-        false ->
-            {Body0, UserHeaders1}
-    end,
-    
-    Headers1 = merge_headers(BaseHeaders, UserHeaders2),
+    UserHeaders = get_value(headers, Params, []),
+    Headers1 = merge_headers(BaseHeaders, UserHeaders),
     {Headers2, HttpDb} = couch_replicator_auth:update_headers(HttpDb0, Headers1),
     Url0 = full_url(HttpDb, Params),
+    Body = get_value(body, Params, []),
     case get_value(path, Params) == "_changes" of
         true ->
             Timeout = infinity;
@@ -282,9 +183,6 @@ process_response({error, connection_closing}, Worker, HttpDb, Params, _Cb) ->
 process_response({ibrowse_req_id, ReqId}, Worker, HttpDb, Params, Callback) ->
     process_stream_response(ReqId, Worker, HttpDb, Params, Callback);
 process_response({ok, Code, Headers, Body}, Worker, HttpDb, Params, Callback) ->
-    % Decompress body if it's gzip compressed
-    DecompressedBody = decompress_body(Headers, Body),
-    
     case list_to_integer(Code) of
         R when R =:= 301; R =:= 302; R =:= 303 ->
             backoff_success(HttpDb, Params),
@@ -298,7 +196,7 @@ process_response({ok, Code, Headers, Body}, Worker, HttpDb, Params, Callback) ->
             backoff_success(HttpDb, Params),
             couch_stats:increment_counter([couch_replicator, responses, success]),
             EJson =
-                case DecompressedBody of
+                case Body of
                     <<>> ->
                         null;
                     Json ->

src/couch_replicator/test/eunit/couch_replicator_compression_tests.erl

@@ -26,32 +26,32 @@ compression_test_() ->
             fun couch_replicator_test_helper:test_setup/0,
             fun couch_replicator_test_helper:test_teardown/1,
             [
-                ?TDEF_FE(should_compress_http_requests, ?TIMEOUT_EUNIT)
+                ?TDEF_FE(should_not_compress_by_default, ?TIMEOUT_EUNIT),
+                ?TDEF_FE(should_compress_when_enabled, ?TIMEOUT_EUNIT)
             ]
         }
     }.
 
-should_compress_http_requests({_Ctx, {Source, Target}}) ->
-    config:set("replicator", "compress_min_size", "10", false),
+should_not_compress_by_default({_Ctx, {Source, Target}}) ->
+    Before = couch_stats:sample([couch_replicator, requests_compressed, gzip]),
+    populate_db(Source, ?DOCS_COUNT),
+    replicate(Source, Target),
+    compare_dbs(Source, Target),
+    After = couch_stats:sample([couch_replicator, requests_compressed, gzip]),
+    ?assertEqual(Before, After).
+
+should_compress_when_enabled({_Ctx, {Source, Target}}) ->
     config:set("replicator", "compress_requests", "true", false),
+    config:set("replicator", "compress_min_size", "10", false),
     config:set("replicator", "compression_algorithm", "gzip", false),
-    InitialCompressed = couch_stats:sample([couch_replicator, requests_compressed]),
-    InitialGzip = couch_stats:sample([couch_replicator, requests_compressed, gzip]),
-    ?assertEqual(0, InitialCompressed),
-    ?assertEqual(0, InitialGzip),
+    Before = couch_stats:sample([couch_replicator, requests_compressed, gzip]),
     populate_db(Source, ?DOCS_COUNT),
     replicate(Source, Target),
-    compare_dbs(Source, Target),   
-    FinalCompressed = couch_stats:sample([couch_replicator, requests_compressed]),
-    FinalGzip = couch_stats:sample([couch_replicator, requests_compressed, gzip]),
-    ?assertEqual(?DOCS_COUNT, FinalCompressed,
-                 io_lib:format("Expected ~p compressed requests, got: ~p", [?DOCS_COUNT, FinalCompressed])),
-    ?assertEqual(?DOCS_COUNT, FinalGzip,
-                 io_lib:format("Expected ~p gzip requests, got: ~p", [?DOCS_COUNT, FinalGzip])),
-    ?assertEqual(FinalCompressed, FinalGzip,
-                 "All compressed requests should use gzip algorithm"),
-    config:delete("replicator", "compress_min_size", false),
+    compare_dbs(Source, Target),
+    After = couch_stats:sample([couch_replicator, requests_compressed, gzip]),
+    ?assert(After > Before),
     config:delete("replicator", "compress_requests", false),
+    config:delete("replicator", "compress_min_size", false),
     config:delete("replicator", "compression_algorithm", false).
 
 populate_db(DbName, Count) ->