Remove hackney, support gun for S3, clean up code

Author: speeddragon

rebar.config

@@ -1,5 +1,5 @@
 {erl_opts, [debug_info, {d, 'COWBOY_QUICER', 1}, {d, 'GUN_QUICER', 1}]}.
-{plugins, [pc, rebar3_rustler, rebar_edown_plugin, {rebar_cmd, "0.2.6"}]}.
+{plugins, [pc, rebar3_rustler, rebar_edown_plugin]}.
 
 {profiles, [
     {no_events, [{erl_opts, [{d, 'NO_EVENTS', true}]}]},
@@ -41,7 +41,7 @@
         ]}
     ]},
     {s3, [
-        {deps, [{erlcloud, "3.8.3"}]},
+        {deps, [{erlcloud, {git, "https://github.com/erlcloud/erlcloud.git", {ref, "7322624227e12120d88c47a72a2ba0a032ace9f3"}}}]},
         {erl_opts, [
             {d, 'ENABLE_S3', true}
         ]}
@@ -62,6 +62,7 @@
             {add, gun, [{erl_opts, [{d, 'GUN_QUICER', 1}]}]}
         ]}
     ]},
+    {recon, [{deps, [{recon, "2.5.6"}]}]},
     {test, [
 		{deps, [{meck, "1.1.0"}]}
     ]}
@@ -185,11 +186,6 @@
     }
 ]}.
 
-{commands, [
-    {docker_up, "docker-compose up -d"},
-    {docker_down, "docker-compose down"}
-]}.
-
 {edoc_opts, [
     {doclet, edown_doclet},
 	{dir, "docs/resources/source-code"},

rebar.lock

@@ -5,6 +5,7 @@
        {ref,"58f0502e49bf73b29d95c6d02460d1fb8d2a5273"}},
   0},
  {<<"base16">>,{pkg,<<"base16">>,<<"2.0.1">>},1},
+ {<<"certifi">>,{pkg,<<"certifi">>,<<"2.15.0">>},1},
  {<<"cowboy">>,{pkg,<<"cowboy">>,<<"2.14.0">>},0},
  {<<"cowlib">>,{pkg,<<"cowlib">>,<<"2.16.0">>},0},
  {<<"ddskerl">>,{pkg,<<"ddskerl">>,<<"0.4.2">>},1},
@@ -16,46 +17,66 @@
  {<<"erlcloud">>,{pkg,<<"erlcloud">>,<<"3.8.3">>},0},
  {<<"graphql">>,{pkg,<<"graphql_erl">>,<<"0.17.1">>},0},
  {<<"gun">>,{pkg,<<"gun">>,<<"2.2.0">>},0},
+ {<<"idna">>,{pkg,<<"idna">>,<<"6.1.1">>},1},
  {<<"jsx">>,{pkg,<<"jsx">>,<<"2.11.0">>},1},
  {<<"lhttpc">>,{pkg,<<"lhttpc">>,<<"1.7.1">>},1},
  {<<"luerl">>,{pkg,<<"luerl">>,<<"1.3.0">>},0},
+ {<<"metrics">>,{pkg,<<"metrics">>,<<"1.0.1">>},1},
+ {<<"mimerl">>,{pkg,<<"mimerl">>,<<"1.4.0">>},1},
+ {<<"parse_trans">>,{pkg,<<"parse_trans">>,<<"3.4.1">>},1},
  {<<"prometheus">>,{pkg,<<"prometheus">>,<<"6.0.3">>},0},
  {<<"prometheus_cowboy">>,{pkg,<<"prometheus_cowboy">>,<<"0.2.0">>},0},
  {<<"prometheus_httpd">>,{pkg,<<"prometheus_httpd">>,<<"2.1.15">>},0},
- {<<"ranch">>,{pkg,<<"ranch">>,<<"2.2.0">>},0}]}.
+ {<<"ranch">>,{pkg,<<"ranch">>,<<"2.2.0">>},0},
+ {<<"ssl_verify_fun">>,{pkg,<<"ssl_verify_fun">>,<<"1.1.7">>},1},
+ {<<"unicode_util_compat">>,{pkg,<<"unicode_util_compat">>,<<"0.7.1">>},1}]}.
 [
 {pkg_hash,[
  {<<"accept">>, <<"CD6E34A2D7E28CA38B2D3CB233734CA0C221EFBC1F171F91FEC5F162CC2D18DA">>},
  {<<"base16">>, <<"F0549F732E03BE8124ED0D19FD5EE52146CC8BE24C48CBC3F23AB44B157F11A2">>},
+ {<<"certifi">>, <<"0E6E882FCDAAA0A5A9F2B3DB55B1394DBA07E8D6D9BCAD08318FB604C6839712">>},
  {<<"cowboy">>, <<"565DCF221BA99B1255B0ADCEC24D2D8DBE79E46EC79F30F8373CCEADC6A41E2A">>},
  {<<"cowlib">>, <<"54592074EBBBB92EE4746C8A8846E5605052F29309D3A873468D76CDF932076F">>},
  {<<"ddskerl">>, <<"A51A90BE9AC9B36A94017670BED479C623B10CA9D4BDA1EDF3A0E48CAEEADA2A">>},
  {<<"eini">>, <<"FCC3CBD49BBDD9A1D9735C7365DAFFCD84481CCE81E6CB80537883AA44AC4895">>},
  {<<"erlcloud">>, <<"66DE6C7D37C6E688C7AE198D56B9CBE07B2CB80054B88A798437C533C3C7F418">>},
  {<<"graphql">>, <<"EB59FCBB39F667DC1C78C950426278015C3423F7A6ED2A121D3DB8B1D2C5F8B4">>},
  {<<"gun">>, <<"B8F6B7D417E277D4C2B0DC3C07DFDF892447B087F1CC1CAFF9C0F556B884E33D">>},
+ {<<"idna">>, <<"8A63070E9F7D0C62EB9D9FCB360A7DE382448200FBBD1B106CC96D3D8099DF8D">>},
  {<<"jsx">>, <<"08154624050333919B4AC1B789667D5F4DB166DC50E190C4D778D1587F102EE0">>},
  {<<"lhttpc">>, <<"8522AF9877765C33318A3AE486BE69BC165E835D05C3334A8166FD7B318D446B">>},
  {<<"luerl">>, <<"B56423DDB721432AB980B818FEECB84ADBAB115E2E11522CF94BCD0729CAA501">>},
+ {<<"metrics">>, <<"25F094DEA2CDA98213CECC3AEFF09E940299D950904393B2A29D191C346A8486">>},
+ {<<"mimerl">>, <<"3882A5CA67FBBE7117BA8947F27643557ADEC38FA2307490C4C4207624CB213B">>},
+ {<<"parse_trans">>, <<"6E6AA8167CB44CC8F39441D05193BE6E6F4E7C2946CB2759F015F8C56B76E5FF">>},
  {<<"prometheus">>, <<"95302236124C0F919163A7762BF7D2B171B919B6FF6148D26EB38A5D2DEF7B81">>},
  {<<"prometheus_cowboy">>, <<"526F75D9850A9125496F78BCEECCA0F237BC7B403C976D44508543AE5967DAD9">>},
  {<<"prometheus_httpd">>, <<"8F767D819A5D36275EAB9264AFF40D87279151646776069BF69FBDBBD562BD75">>},
- {<<"ranch">>, <<"25528F82BC8D7C6152C57666CA99EC716510FE0925CB188172F41CE93117B1B0">>}]},
+ {<<"ranch">>, <<"25528F82BC8D7C6152C57666CA99EC716510FE0925CB188172F41CE93117B1B0">>},
+ {<<"ssl_verify_fun">>, <<"354C321CF377240C7B8716899E182CE4890C5938111A1296ADD3EC74CF1715DF">>},
+ {<<"unicode_util_compat">>, <<"A48703A25C170EEDADCA83B11E88985AF08D35F37C6F664D6DCFB106A97782FC">>}]},
 {pkg_hash_ext,[
  {<<"accept">>, <<"CA69388943F5DAD2E7232A5478F16086E3C872F48E32B88B378E1885A59F5649">>},
  {<<"base16">>, <<"06EA2D48343282E712160BA89F692B471DB8B36ABE8394F3445FF9032251D772">>},
+ {<<"certifi">>, <<"B147ED22CE71D72EAFDAD94F055165C1C182F61A2FF49DF28BCC71D1D5B94A60">>},
  {<<"cowboy">>, <<"EA99769574550FE8A83225C752E8A62780A586770EF408816B82B6FE6D46476B">>},
  {<<"cowlib">>, <<"7F478D80D66B747344F0EA7708C187645CFCC08B11AA424632F78E25BF05DB51">>},
  {<<"ddskerl">>, <<"63F907373D7E548151D584D4DA8A38928FD26EC9477B94C0FFAAD87D7CB69FE7">>},
  {<<"eini">>, <<"DA64AE8DB7C2F502E6F20CDF44CD3D9BE364412B87FF49FEBF282540F673DFCB">>},
  {<<"erlcloud">>, <<"6ED59CBD8816045765E3E76F22D08118766DF1BB1D24F5D90794C8505BCD6D44">>},
  {<<"graphql">>, <<"4D0F08EC57EF0983E2596763900872B1AB7E94F8EE3817B9F67EEC911FF7C386">>},
  {<<"gun">>, <<"76022700C64287FEB4DF93A1795CFF6741B83FB37415C40C34C38D2A4645261A">>},
+ {<<"idna">>, <<"92376EB7894412ED19AC475E4A86F7B413C1B9FBB5BD16DCCD57934157944CEA">>},
  {<<"jsx">>, <<"EED26A0D04D217F9EECEFFFB89714452556CF90EB38F290A27A4D45B9988F8C0">>},
  {<<"lhttpc">>, <<"154EEB27692482B52BE86406DCD1D18A2405CAFCE0E8DAA4A1A7BFA7FE295896">>},
  {<<"luerl">>, <<"6B3138AA829F0FBC4CD0F083F273B4030A2B6CE99155194A6DB8C67B2C3480A4">>},
+ {<<"metrics">>, <<"69B09ADDDC4F74A40716AE54D140F93BEB0FB8978D8636EADED0C31B6F099F16">>},
+ {<<"mimerl">>, <<"13AF15F9F68C65884ECCA3A3891D50A7B57D82152792F3E19D88650AA126B144">>},
+ {<<"parse_trans">>, <<"620A406CE75DADA827B82E453C19CF06776BE266F5A67CFF34E1EF2CBB60E49A">>},
  {<<"prometheus">>, <<"53554ECADAC0354066801D514D1A244DD026175E4EE3A9A30192B71D530C8268">>},
  {<<"prometheus_cowboy">>, <<"2C7EB12F4B970D91E3B47BAAD0F138F6ADC34E53EEB0AE18068FF0AFAB441B24">>},
  {<<"prometheus_httpd">>, <<"67736D000745184D5013C58A63E947821AB90CB9320BC2E6AE5D3061C6FFE039">>},
- {<<"ranch">>, <<"FA0B99A1780C80218A4197A59EA8D3BDAE32FBFF7E88527D7D8A4787EFF4F8E7">>}]}
+ {<<"ranch">>, <<"FA0B99A1780C80218A4197A59EA8D3BDAE32FBFF7E88527D7D8A4787EFF4F8E7">>},
+ {<<"ssl_verify_fun">>, <<"FE4C190E8F37401D30167C8C405EDA19469F34577987C76DDE613E838BBC67F8">>},
+ {<<"unicode_util_compat">>, <<"B3A917854CE3AE233619744AD1E0102E05673136776FB2FA76234F3E03B23642">>}]}
 ].

src/hb_cache.erl

@@ -385,6 +385,7 @@ read(Path, Opts) ->
 %% @doc List all of the subpaths of a given path and return a map of keys and
 %% links to the subpaths, including their types.
 store_read(_Path, no_viable_store, _) ->
+    erlang:display("NO_VIABLE_STORE"),
     not_found;
 store_read(Path, Store, Opts) ->
     ResolvedFullPath = hb_store:resolve(Store, PathBin = hb_path:to_binary(Path)),
@@ -394,12 +395,16 @@ store_read(Path, Store, Opts) ->
         {store, Store}
     }),
     case hb_store:type(Store, ResolvedFullPath) of
-        not_found -> not_found;
+        not_found -> 
+            erlang:display({type_not_found, {resolved_full_path, ResolvedFullPath}}),
+            not_found;
         simple ->
             ?event({reading_data, ResolvedFullPath}),
             case hb_store:read(Store, ResolvedFullPath) of
                 {ok, Bin} -> {ok, Bin};
-                not_found -> not_found
+                not_found -> 
+                    erlang:display("NOT FOUND"),
+                    not_found
             end;
         composite ->
             ?event({reading_composite, ResolvedFullPath}),

src/hb_http_client.erl

@@ -4,7 +4,7 @@
 -behaviour(gen_server).
 -include("include/hb.hrl").
 -export([start_link/1, request/2]).
--export([init/1, handle_cast/2, handle_call/3, handle_info/2, terminate/2]).
+-export([init/1, handle_cast/2, handle_call/3, handle_info/2, terminate/2, open_connection/2]).
 
 -record(state, {
 	pid_by_peer = #{},
@@ -154,7 +154,7 @@ gun_req(Args, ReestablishedConnection, Opts) ->
                     Reply ->
                         Reply
                 end;
-            {'EXIT', _} ->
+            {'EXIT', Reason} ->
                 {error, client_error};
             Error ->
                 Error

src/hb_store.erl

@@ -426,20 +426,20 @@ call_all([Store = #{<<"store-module">> := Mod} | Rest], Function, Args) ->
 %% default into all HyperBEAM distributions.
 test_stores() ->
     [
-        (hb_test_utils:test_store(hb_store_fs))#{
-            <<"benchmark-scale">> => 0.001
-        },
-        (hb_test_utils:test_store(hb_store_lmdb))#{
-            <<"benchmark-scale">> => 0.5
-        },
-        (hb_test_utils:test_store(hb_store_lru))#{
-            <<"persistent-store">> => [
-                #{
-                    <<"store-module">> => hb_store_fs,
-                    <<"name">> => <<"cache-TEST/lru">>
-                }
-            ]
-        }
+%        (hb_test_utils:test_store(hb_store_fs))#{
+%            <<"benchmark-scale">> => 0.001
+%        },
+%        (hb_test_utils:test_store(hb_store_lmdb))#{
+%            <<"benchmark-scale">> => 0.5
+%        },
+%        (hb_test_utils:test_store(hb_store_lru))#{
+%            <<"persistent-store">> => [
+%                #{
+%                    <<"store-module">> => hb_store_fs,
+%                    <<"name">> => <<"cache-TEST/lru">>
+%                }
+%            ]
+%        }
     ] ++ rocks_stores() ++ s3_stores().
 
 -ifdef(ENABLE_ROCKSDB).
@@ -465,6 +465,7 @@ generate_test_suite(Suite) ->
     generate_test_suite(Suite, test_stores()).
 generate_test_suite(Suite, Stores) ->
     hb:init(),
+    application:ensure_all_started(hb),
     lists:map(
         fun(Store = #{<<"store-module">> := Mod}) ->
             {foreach,
@@ -474,7 +475,8 @@ generate_test_suite(Suite, Stores) ->
                     hb_store:reset(Store)
                 end,
                 fun(_) ->
-                    hb_store:reset(Store)
+                    %hb_store:reset(Store)
+                    ok
                     % hb_store:stop(Store)
                 end,
                 [
@@ -531,8 +533,8 @@ store_suite_test_() ->
 
 benchmark_suite_test_() ->
     generate_test_suite([
-        {"benchmark key read write", fun benchmark_key_read_write/1},
-        {"benchmark list", fun benchmark_list/1},
+        %{"benchmark key read write", fun benchmark_key_read_write/1},
+        %{"benchmark list", fun benchmark_list/1},
         {"benchmark message read write", fun benchmark_message_read_write/1}
     ]).
 
@@ -850,6 +852,7 @@ benchmark_message_read_write(Store, WriteOps, ReadOps) ->
                     fun({MsgID, Msg}, Count) -> 
                         case hb_cache:read(MsgID, Opts) of
                             {ok, Base} ->
+                                erlang:display(Base),        
                                 case hb_cache:ensure_all_loaded(Base, Opts) of
                                     Msg -> Count;
                                     _ -> Count + 1

src/hb_store_s3.erl

@@ -12,7 +12,7 @@
 -export([read/2, write/3, list/2, type/2]).
 -export([make_group/2, make_link/3, resolve/2]).
 -export([path/2, add_path/3]).
--export([default_test_opts/0]).
+-export([default_test_opts/0, get_config/1]).
 
 %% Helper functions
 -export([match/2]).
@@ -36,6 +36,10 @@
 -define(LINKS_NS, <<"links/">>).
 -define(GROUPS_NS, <<"groups/">>).
 
+-define(DEFAULT_RETRY_DELAY, 1000).
+-define(DEFAULT_RETRY_MODE, exponential_backoff).
+-define(DEFAULT_RETRIES, 5).
+-define(MAX_RETRY_DELAY, 300000).
 %%%-----------------------------------------------------------------------------
 %%% Configuration and Initialization (Phase 2)
 %%%-----------------------------------------------------------------------------
@@ -68,7 +72,8 @@ start(Opts) ->
             s3_bucket_after_host = false,
             s3_bucket_access_method = ForcePathStyle,
             aws_region = Region,
-            http_client = httpc
+            % Use `gun_pool` to define a connection pool.
+            http_client = fun gun_request/6
         },
         ok ?= test_bucket_access(Bucket, Config),
         StoreRef = get_store_ref(Opts),
@@ -89,6 +94,43 @@ start(Opts) ->
             {error, Error}
     end.
 
+
+%% Interface erlcloud_s3 with HB HTTP Client
+gun_request(URL, Method, Headers, Body, Timeout, _Config) when is_atom(Method) ->
+    case uri_string:parse(URL) of 
+        #{port := Port, scheme := Scheme, host := Host} = ParsedURL  ->
+            Peer = uri_string:normalize(#{port => Port, scheme => Scheme, host => Host}),
+            HeadersMap = maps:from_list(Headers),
+            MethodBinary = string:uppercase(atom_to_binary(Method)),
+            Args = #{ 
+                     peer => Peer, 
+                     path => uri_string:normalize(maps:with([path, fragment, query], ParsedURL)), 
+                     method => MethodBinary, 
+                     headers => HeadersMap, 
+                     body => Body
+                    },
+            Opts = #{connect_timeout => Timeout},
+            Response = hb_http_client:request(Args, Opts),
+            handle_gun_response(Response);
+        Reason -> 
+            ?event(error, {parsing_url, {url, URL},{reason, Reason}}),
+            {error, Reason}
+    end.
+
+handle_gun_response({ok, Status, ResponseHeaders, Body}) -> 
+        {ok, {{Status, undefined}, header_str(ResponseHeaders), Body}};
+
+handle_gun_response({error, _} = Error) -> 
+    Error.
+
+header_str(Hdrs) ->
+    [{string:to_lower(to_list_string(K)), to_list_string(V)} || {K, V} <- Hdrs].
+
+to_list_string(Val) when erlang:is_binary(Val) ->
+  erlang:binary_to_list(Val);
+to_list_string(Val) when erlang:is_list(Val) ->
+  Val.
+
 %% @doc Validate that all required configuration keys are present.
 %% Required keys: bucket, priv_access_key_id, priv_secret_access_key
 validate_config(Opts) ->
@@ -107,7 +149,7 @@ test_bucket_access(Bucket, Config) ->
     try erlcloud_s3:list_objects(BucketStr, [{max_keys, 1}], Config) of
         L when is_list(L) -> ok
     catch 
-        _Class:Reason -> 
+        _Class:Reason:Stacktrace -> 
             case Reason of 
                 {aws_error, {http_error, 404, _, _}} ->
                     error({bucket_not_found, Bucket});
@@ -116,6 +158,7 @@ test_bucket_access(Bucket, Config) ->
                      {inet,[inet],econnrefused}]}}} -> 
                     error("Check if MinIO is running locally. Eg: `rebar3 cmd docker_up`");
                 _ -> 
+                    ?event(error, {error, {reason, Reason}, {stacktrace, Stacktrace}}),
                     error({bucket_access_failed, Reason})
             end
     end.
@@ -141,10 +184,15 @@ get_config(Opts) ->
 %%%-----------------------------------------------------------------------------
 
 %% @doc Write a value to a key in S3.
--spec write(opts(), key(), value()) -> ok | retry.
+-spec write(opts(), key(), value()) -> ok | not_found.
 write(Opts, Key, Value) when is_list(Key) ->
     write(Opts, hb_store:join(Key), Value);
 write(Opts, Key, Value) when is_binary(Key) ->
+    RetryAttempts = maps:get(<<"retry-attempts">>, Opts, ?DEFAULT_RETRIES),
+    write(Opts, Key, Value, RetryAttempts).
+write(_Opts, _Key, _Value, 0) -> 
+    ok;
+write(Opts, Key, Value, AttemptsRemaining) ->
     maybe
         #{bucket := Bucket, prefix := Prefix, config := Config} = get_config(Opts),
 
@@ -169,7 +217,17 @@ write(Opts, Key, Value) when is_binary(Key) ->
     else
         Error ->
             ?event(error, {s3_write_error, {key, Key}, {reason, Error}}),
-            retry
+            MaxRetries = maps:get(<<"retry-attempts">>, Opts, ?DEFAULT_RETRIES),
+            MinRetryDelay = maps:get(<<"min-retry-delay">>, Opts, ?DEFAULT_RETRY_DELAY),
+            MaxRetryDelay = maps:get(<<"max-retry-delay">>, Opts, ?MAX_RETRY_DELAY),
+            RetryTime = case maps:get(<<"retry-mode">>, Opts, ?DEFAULT_RETRY_MODE) of
+                            exponential_backoff -> 
+                                min(MinRetryDelay * math:pow(2, MaxRetries - AttemptsRemaining), MaxRetryDelay);
+                            _ -> MinRetryDelay 
+                        end,
+            ?event(store_s3, {retry_in, RetryTime}),
+            timer:sleep(RetryTime),
+            write(Opts, Key, Value, AttemptsRemaining - 1)
     end.
 
 %% @doc Build full S3 key with optional prefix.
@@ -376,6 +434,7 @@ list(Opts, Path) ->
         BucketStr = hb_util:list(Bucket),
         PrefixStr = hb_util:list(SearchPrefix),
         ListOpts = [{prefix, PrefixStr}, {delimiter, "/"}],
+        ?event(store_s3, {list_opts, ListOpts}),
         try erlcloud_s3:list_objects(BucketStr, ListOpts, Config) of
             L when is_list(L) ->
                 Children = extract_children(SearchPrefix, L),
@@ -658,6 +717,9 @@ default_test_opts() ->
         <<"force_path_style">> => true
      }.
 
+default_test_opts(Opts) -> 
+    maps:merge(default_test_opts(), Opts).
+
 -ifdef(ENABLE_S3).
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").