Merge pull request #10947 from emqx/release-51

Release 51 to master

.github/actions/package-macos/action.yaml

@@ -43,8 +43,18 @@ runs:
         echo "OTP_SOURCE_PATH=$OTP_SOURCE_PATH" >> $GITHUB_OUTPUT
         echo "OTP_INSTALL_PATH=$OTP_INSTALL_PATH" >> $GITHUB_OUTPUT
         mkdir -p "$OTP_SOURCE_PATH" "$OTP_INSTALL_PATH"
+        # we need this to skip using cache for self-hosted runners
+        case ${{ inputs.os }} in
+          *arm64)
+            echo "SELF_HOSTED=true" >> $GITHUB_OUTPUT
+            ;;
+          *)
+            echo "SELF_HOSTED=false" >> $GITHUB_OUTPUT
+            ;;
+        esac
     - uses: actions/cache@v3
       id: cache
+      if: steps.prepare.outputs.SELF_HOSTED != 'true'
       with:
         path: ${{ steps.prepare.outputs.OTP_INSTALL_PATH }}
         key: otp-install-${{ inputs.otp }}-${{ inputs.os }}-static-ssl-disable-hipe-disable-jit
@@ -54,22 +64,36 @@ runs:
       run: |
         OTP_SOURCE_PATH="${{ steps.prepare.outputs.OTP_SOURCE_PATH }}"
         OTP_INSTALL_PATH="${{ steps.prepare.outputs.OTP_INSTALL_PATH }}"
+        SELF_HOSTED="${{ steps.prepare.outputs.SELF_HOSTED }}"
+        # when it's self-hosted, it never hits the cache,
+        # skip rebuild if it's self-hosted and the install path already has a 'bin'
+        if [ "${SELF_HOSTED:-false}" = 'true' ]; then
+          if [ -n "$OTP_INSTALL_PATH" ] && [ -d "$OTP_INSTALL_PATH/bin" ]; then
+            echo "Skip rebuilding OTP, found $OTP_INSTALL_PATH"
+            exit 0
+          fi
+        fi
+        ## when it's not self-hosted, or the install path is not found,
+        ## build otp from source code.
         if [ -d "$OTP_SOURCE_PATH" ]; then
           rm -rf "$OTP_SOURCE_PATH"
         fi
         git clone --depth 1 --branch OTP-${{ inputs.otp }} https://github.com/emqx/otp.git "$OTP_SOURCE_PATH"
         cd "$OTP_SOURCE_PATH"
         if [ "$(arch)" = arm64 ]; then
+            export CFLAGS="-O2 -g -I$(brew --prefix unixodbc)/include"
             export LDFLAGS="-L$(brew --prefix unixodbc)/lib"
-            export CC="/usr/bin/gcc -I$(brew --prefix unixodbc)/include"
+            WITH_ODBC="--with-odbc=$(brew --prefix unixodbc)"
+        else
+            WITH_ODBC=""
         fi
-        ./configure --disable-dynamic-ssl-lib --with-ssl=$(brew --prefix openssl@1.1) --disable-hipe --disable-jit --prefix="$OTP_INSTALL_PATH"
+        ./configure --disable-dynamic-ssl-lib --with-ssl=$(brew --prefix openssl@1.1) ${WITH_ODBC} --disable-hipe --disable-jit --prefix="$OTP_INSTALL_PATH"
         make -j$(nproc)
         rm -rf "$OTP_INSTALL_PATH"
         make install
         if [ "$(arch)" = arm64 ]; then
+            unset CFLAGS
             unset LDFLAGS
-            unset CC
         fi
     - name: build
       env:
@@ -87,6 +111,10 @@ runs:
       shell: bash
       run: |
         export PATH="${{ steps.prepare.outputs.OTP_INSTALL_PATH }}/bin:$PATH"
+        # inspec erl in PATH
+        which erl
+        # inspec erl command banner
+        erl -s init stop
         make ensure-rebar3
         mkdir -p $HOME/bin
         cp rebar3 $HOME/bin/rebar3

apps/emqx/include/emqx_release.hrl

@@ -32,10 +32,10 @@
 %% `apps/emqx/src/bpapi/README.md'
 
 %% Community edition
--define(EMQX_RELEASE_CE, "5.1.0-alpha.2").
+-define(EMQX_RELEASE_CE, "5.1.0-alpha.3").
 
 %% Enterprise edition
--define(EMQX_RELEASE_EE, "5.1.0-alpha.2").
+-define(EMQX_RELEASE_EE, "5.1.0-alpha.3").
 
 %% the HTTP API version
 -define(EMQX_API_VERSION, "5.0").

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_connector.erl

@@ -99,7 +99,7 @@ choose_ingress_pool_size(
         {_Filter, #{share := _Name}} ->
             % NOTE: this is shared subscription, many workers may subscribe
             PoolSize;
-        {_Filter, #{}} ->
+        {_Filter, #{}} when PoolSize > 1 ->
             % NOTE: this is regular subscription, only one worker should subscribe
             ?SLOG(warning, #{
                 msg => "mqtt_bridge_ingress_pool_size_ignored",
@@ -110,6 +110,8 @@ choose_ingress_pool_size(
                 config_pool_size => PoolSize,
                 pool_size => 1
             }),
+            1;
+        {_Filter, #{}} when PoolSize == 1 ->
             1
     end.
 

apps/emqx_ft/src/emqx_ft_fs_util.erl

@@ -29,6 +29,8 @@
 
 -export([fold/4]).
 
+-export([mk_temp_filename/1]).
+
 -type foldfun(Acc) ::
     fun(
         (
@@ -178,3 +180,24 @@ fold(FoldFun, Acc, It) ->
         none ->
             Acc
     end.
+
+-spec mk_temp_filename(file:filename()) ->
+    file:filename().
+mk_temp_filename(Filename) ->
+    % NOTE
+    % Using only the first 200 characters of the filename to avoid making filenames
+    % exceeding 255 bytes in UTF-8. It's actually too conservative, `Suffix` can be
+    % at most 16 bytes.
+    Unique = erlang:unique_integer([positive]),
+    Suffix = binary:encode_hex(<<Unique:64>>),
+    mk_filename([string:slice(Filename, 0, 200), ".", Suffix]).
+
+mk_filename(Comps) ->
+    lists:append(lists:map(fun mk_filename_component/1, Comps)).
+
+mk_filename_component(A) when is_atom(A) ->
+    atom_to_list(A);
+mk_filename_component(B) when is_binary(B) ->
+    unicode:characters_to_list(B);
+mk_filename_component(S) when is_list(S) ->
+    S.

apps/emqx_ft/src/emqx_ft_schema.erl

@@ -42,7 +42,9 @@
 %% on most filesystems. Even though, say, S3 does not have such limitations, it's
 %% still useful to have a limit on the filename length, to avoid having to deal with
 %% limits in the storage backends.
--define(MAX_FILENAME_BYTELEN, 255).
+%% Usual realistic limit is 255 bytes actually, but we leave some room for backends
+%% to spare.
+-define(MAX_FILENAME_BYTELEN, 240).
 
 -import(hoconsc, [ref/2, mk/2]).
 
@@ -145,7 +147,7 @@ fields(local_storage_segments) ->
     [
         {root,
             mk(
-                binary(),
+                string(),
                 #{
                     desc => ?DESC("local_storage_segments_root"),
                     required => false
@@ -182,7 +184,7 @@ fields(local_storage_exporter) ->
     [
         {root,
             mk(
-                binary(),
+                string(),
                 #{
                     desc => ?DESC("local_storage_exporter_root"),
                     required => false

apps/emqx_ft/src/emqx_ft_storage_exporter_fs.erl

@@ -128,7 +128,17 @@ complete(
     Filemeta = FilemetaIn#{checksum => Checksum},
     ok = file:close(Handle),
     _ = filelib:ensure_dir(ResultFilepath),
-    _ = file:write_file(mk_manifest_filename(ResultFilepath), encode_filemeta(Filemeta)),
+    ManifestFilepath = mk_manifest_filename(ResultFilepath),
+    case file:write_file(ManifestFilepath, encode_filemeta(Filemeta)) of
+        ok ->
+            ok;
+        {error, Reason} ->
+            ?SLOG(warning, "filemeta_write_failed", #{
+                path => ManifestFilepath,
+                meta => Filemeta,
+                reason => Reason
+            })
+    end,
     file:rename(Filepath, ResultFilepath).
 
 -spec discard(export_st()) ->
@@ -452,8 +462,7 @@ mk_manifest_filename(Filename) when is_binary(Filename) ->
     <<Filename/binary, ?MANIFEST>>.
 
 mk_temp_absfilepath(Options, Transfer, Filename) ->
-    Unique = erlang:unique_integer([positive]),
+    TempFilename = emqx_ft_fs_util:mk_temp_filename(Filename),
-    TempFilename = integer_to_list(Unique) ++ "." ++ Filename,
     filename:join(mk_absdir(Options, Transfer, temporary), TempFilename).
 
 mk_absdir(Options, _Transfer, temporary) ->

apps/emqx_ft/src/emqx_ft_storage_fs.erl

@@ -445,16 +445,8 @@ write_file_atomic(Storage, Transfer, Filepath, Content) when is_binary(Content)
     end.
 
 mk_temp_filepath(Storage, Transfer, Filename) ->
-    Unique = erlang:unique_integer([positive]),
+    TempFilename = emqx_ft_fs_util:mk_temp_filename(Filename),
-    filename:join(get_subdir(Storage, Transfer, temporary), mk_filename([Unique, ".", Filename])).
+    filename:join(get_subdir(Storage, Transfer, temporary), TempFilename).
-
-mk_filename(Comps) ->
-    lists:append(lists:map(fun mk_filename_component/1, Comps)).
-
-mk_filename_component(I) when is_integer(I) -> integer_to_list(I);
-mk_filename_component(A) when is_atom(A) -> atom_to_list(A);
-mk_filename_component(B) when is_binary(B) -> unicode:characters_to_list(B);
-mk_filename_component(S) when is_list(S) -> S.
 
 write_contents(Filepath, Content) ->
     file:write_file(Filepath, Content).

apps/emqx_ft/test/emqx_ft_SUITE.erl

@@ -261,6 +261,7 @@ t_nasty_clientids_fileids(_Config) ->
         fun({ClientId, FileId}) ->
             ok = emqx_ft_test_helpers:upload_file(ClientId, FileId, "justfile", ClientId),
             [Export] = list_files(ClientId),
+            ?assertMatch(#{meta := #{name := "justfile"}}, Export),
             ?assertEqual({ok, ClientId}, read_export(Export))
         end,
         Transfers
@@ -270,13 +271,15 @@ t_nasty_filenames(_Config) ->
     Filenames = [
         {<<"nasty1">>, "146%"},
         {<<"nasty2">>, "🌚"},
-        {<<"nasty3">>, "中文.txt"}
+        {<<"nasty3">>, "中文.txt"},
+        {<<"nasty4">>, _239Bytes = string:join(lists:duplicate(240 div 5, "LONG"), ".")}
     ],
     ok = lists:foreach(
         fun({ClientId, Filename}) ->
             FileId = unicode:characters_to_binary(Filename),
             ok = emqx_ft_test_helpers:upload_file(ClientId, FileId, Filename, FileId),
             [Export] = list_files(ClientId),
+            ?assertMatch(#{meta := #{name := Filename}}, Export),
             ?assertEqual({ok, FileId}, read_export(Export))
         end,
         Filenames

apps/emqx_ft/test/emqx_ft_conf_SUITE.erl

@@ -87,7 +87,7 @@ t_update_config(_Config) ->
         )
     ),
     ?assertEqual(
-        <<"/tmp/path">>,
+        "/tmp/path",
         emqx_config:get([file_transfer, storage, local, segments, root])
     ),
     ?assertEqual(
@@ -150,7 +150,7 @@ t_disable_restore_config(Config) ->
     ),
     ok = emqtt:stop(Client),
     % Restore local storage backend
-    Root = iolist_to_binary(emqx_ft_test_helpers:root(Config, node(), [segments])),
+    Root = emqx_ft_test_helpers:root(Config, node(), [segments]),
     ?assertMatch(
         {ok, _},
         emqx_conf:update(
@@ -177,7 +177,7 @@ t_disable_restore_config(Config) ->
                 [
                     #{
                         ?snk_kind := garbage_collection,
-                        storage := #{segments := #{root := Root}}
+                        storage := #{segments := #{gc := #{interval := 1000}}}
                     }
                 ],
                 ?of_kind(garbage_collection, Trace)

apps/emqx_ft/test/emqx_ft_fs_util_tests.erl

@@ -63,3 +63,24 @@ unescape_filename_test_() ->
         ?_assertEqual(Input, emqx_ft_fs_util:unescape_filename(Filename))
      || {Filename, Input} <- ?NAMES
     ].
+
+mk_temp_filename_test_() ->
+    [
+        ?_assertMatch(
+            "." ++ Suffix when length(Suffix) == 16,
+            emqx_ft_fs_util:mk_temp_filename(<<>>)
+        ),
+        ?_assertMatch(
+            "file.name." ++ Suffix when length(Suffix) == 16,
+            emqx_ft_fs_util:mk_temp_filename("file.name")
+        ),
+        ?_assertMatch(
+            "safe.🦺." ++ Suffix when length(Suffix) == 16,
+            emqx_ft_fs_util:mk_temp_filename(<<"safe.🦺"/utf8>>)
+        ),
+        ?_assertEqual(
+            % FilenameSlice + Dot + Suffix
+            200 + 1 + 16,
+            length(emqx_ft_fs_util:mk_temp_filename(lists:duplicate(63, "LONG")))
+        )
+    ].