chore: merge emqx-OTP-25.3.2 into emqx-OTP-26.1.2 and prepare to tag 26.1.2-1

OTP_VERSION

@@ -1 +1 @@
-26.1.2
+26.1.2-1

erts/emulator/beam/erl_bif_info.c

@@ -87,9 +87,9 @@ static char erts_system_version[] = ("Erlang/OTP " ERLANG_OTP_RELEASE
 				     " [erts-" ERLANG_VERSION "]"
 #ifndef OTP_RELEASE
 #ifdef ERLANG_GIT_VERSION
-				     " [source-" ERLANG_GIT_VERSION "]"
+				     " [emqx-" ERLANG_GIT_VERSION "]"
 #else
-				     " [source]"
+				     " [emqx]"
 #endif
 #endif	
 #if defined(ARCH_64)

lib/kernel/src/gen_tcp.erl

@@ -28,6 +28,7 @@
 -export([send/2, recv/2, recv/3, unrecv/2]).
 -export([controlling_process/2]).
 -export([fdopen/2]).
+-export([ipv6_probe/0]).
 
 -include("inet_int.hrl").
 -include("file.hrl").
@@ -120,7 +121,8 @@
         recvtclass |
         recvttl |
         pktoptions |
-	ipv6_v6only.
+        ipv6_v6only |
+        ipv6_probe.
 -type connect_option() ::
         {fd, Fd :: non_neg_integer()} |
         inet:address_family() |
@@ -131,6 +133,7 @@
         {tcp_module, module()} |
         {netns, file:filename_all()} |
         {bind_to_device, binary()} |
+        {ipv6_probe, boolean() | timeout()} |
         option().
 -type listen_option() ::
         {fd, Fd :: non_neg_integer()} |
@@ -157,6 +160,8 @@
 %% Connect a socket
 %%
 
+ipv6_probe() -> true.
+
 -spec connect(SockAddr, Opts) -> {ok, Socket} | {error, Reason} when
       SockAddr :: socket:sockaddr_in() | socket:sockaddr_in6(),
       Opts     :: [inet:inet_backend() | connect_option()],
@@ -218,21 +223,89 @@ connect(#{family := Fam} = SockAddr, Opts, Timeout)
       Reason  :: timeout | inet:posix().
 
 connect(Address, Port, Opts0, Timeout) ->
-    case inet:gen_tcp_module(Opts0) of
+    %% When neither `inet` nor `inet6` is provided in Opts0,
+    %% and if `ipv6_probe` option is given, try to connect ipv6 first.
+    {TryIpv6, Ipv6T} =
+        case proplists:get_value(ipv6_probe, Opts0) of
+                  true -> {true, 2000}; %% default 2 seconds
+                  false -> {false, 0};
+                  undefined -> {false, 0};
+                  T -> {true, T}
+              end,
+    %% delete it to avoid interference
+    Opts1 = proplists:delete(ipv6_probe, Opts0),
+    case inet:gen_tcp_module(Opts1) of
         {?MODULE, Opts} ->
-            Timer = inet:start_timer(Timeout),
-            Res = (catch connect1(Address,Port,Opts,Timer)),
-            _ = inet:stop_timer(Timer),
-            case Res of
-                {ok,S} -> {ok,S};
-                {error, einval} -> exit(badarg);
-                {'EXIT',Reason} -> exit(Reason);
-                Error -> Error
-            end;
+            connect_maybe_ipv6(Address, Port, Opts, Timeout, TryIpv6, Ipv6T);
         {GenTcpMod, Opts} ->
             GenTcpMod:connect(Address, Port, Opts, Timeout)
     end.
 
+connect_maybe_ipv6(Address, Port, Opts, Timeout, TryIpv6, Ipv6T) ->
+    case maybe_ipv6(Address, Opts, TryIpv6) of
+        {maybe, NewOpts} when TryIpv6 ->
+            try
+                {ok, _} = connect_0(Address, Port, NewOpts, Ipv6T)
+            catch
+                _ : _ ->
+                    %% fallback
+                    connect_0(Address, Port, Opts, Timeout)
+            end;
+        NewOpts ->
+            connect_0(Address, Port, NewOpts, Timeout)
+    end.
+
+connect_0(Address, Port, Opts, Timeout) ->
+    Timer = inet:start_timer(Timeout),
+    Res = (catch connect1(Address,Port,Opts,Timer)),
+    _ = inet:stop_timer(Timer),
+    case Res of
+        {ok,S} -> {ok,S};
+        {error, einval} -> exit(badarg);
+        {'EXIT',Reason} -> exit(Reason);
+        Error -> Error
+    end.
+
+maybe_ipv6({local, _}, Opts, _TryIpv6) ->
+    %% unapplicable to local sockets
+    Opts;
+maybe_ipv6(Host, Opts, TryIpv6) ->
+    case lists:member(inet, Opts) orelse lists:member(inet6, Opts) of
+        true ->
+            Opts; %% caller has made the decision
+        false when is_tuple(Host) ->
+            %% ip tuple provided
+            maybe_ipv6_1(Host, Opts);
+        false when TryIpv6 ->
+            %% string host
+            maybe_ipv6_2(Host, Opts);
+        false ->
+            Opts
+    end.
+
+maybe_ipv6_1(Ip, Opts) when tuple_size(Ip) =:= 4 -> Opts;
+maybe_ipv6_1(Ip, Opts) when tuple_size(Ip) =:= 8 -> [inet6 | Opts].
+
+maybe_ipv6_2(Host, Opts) ->
+    case inet:parse_address(Host) of
+        {ok, Ip} when is_tuple(Ip) ->
+            %% ip string provided, parsed into tuple
+            maybe_ipv6_1(Ip, Opts);
+        _ ->
+            maybe_ipv6_3(Host, Opts)
+    end.
+
+maybe_ipv6_3(Host, Opts) ->
+    case inet:getaddr(Host, inet6) of
+        {ok, _} ->
+            %% the target has a resolvable v6 IP
+            %% maybe try to connect
+            {maybe, [inet6 | Opts]};
+        _ ->
+            %% the target has no resolvable v6 IP
+            Opts
+    end.
+
 connect1(Address, Port, Opts0, Timer) ->
     {Mod, Opts} = inet:tcp_module(Opts0, Address),
     case Mod:getaddrs(Address, Timer) of

lib/mnesia/src/Makefile

@@ -55,6 +55,7 @@ MODULES= \
 	mnesia_ext_sup \
 	mnesia_frag \
 	mnesia_frag_hash \
+	mnesia_hook \
 	mnesia_index \
 	mnesia_kernel_sup \
 	mnesia_late_loader \

lib/mnesia/src/mnesia.app.src

@@ -15,6 +15,7 @@
 	     mnesia_ext_sup,
 	     mnesia_frag,
 	     mnesia_frag_hash,
+             mnesia_hook,
 	     mnesia_index,
              mnesia_kernel_sup,
 	     mnesia_late_loader,

lib/mnesia/src/mnesia_hook.erl

@@ -0,0 +1,105 @@
+%%
+%% %CopyrightBegin%
+%%
+%% Copyright Ericsson AB 1996-2021. All Rights Reserved.
+%%
+%% 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.
+%%
+%% %CopyrightEnd%
+%%
+
+-module(mnesia_hook).
+
+-include("mnesia.hrl").
+
+-export([
+         register_hook/2,
+         unregister_hook/1,
+         do_post_commit/2
+        ]).
+
+-define(hook(NAME), {mnesia_hook, NAME}).
+
+-type post_commit_hook_data() ::
+        #{ node => node()
+         , ram_copies => list()
+         , disc_copies => list()
+         , disc_only_copies => list()
+         , ext => list()
+         , schema_ops => list()
+         }.
+
+-type post_commit_hook() :: fun((_Tid, post_commit_hook_data()) -> ok).
+
+-spec register_hook(post_commit, post_commit_hook()) -> ok | {error, term()}.
+register_hook(post_commit, Hook) when is_function(Hook, 2) ->
+    persistent_term:put(?hook(post_commit), Hook);
+register_hook(_, _) ->
+    {error, bad_type}.
+
+-spec unregister_hook(post_commit) -> boolean() | {error, term()}.
+unregister_hook(post_commit) ->
+    persistent_term:erase(?hook(post_commit));
+unregister_hook(_) ->
+    {error, bad_type}.
+
+-spec do_post_commit(_Tid, #commit{}) -> ok.
+do_post_commit(Tid, Commit) ->
+    case persistent_term:get(?hook(post_commit), undefined) of
+        undefined ->
+            ok;
+        Fun ->
+            #commit{ node = Node
+                   , ram_copies = Ram
+                   , disc_copies = Disc
+                   , disc_only_copies = DiscOnly
+                   , ext = Ext
+                   , schema_ops = SchemaOps
+                   } = Commit,
+            CommitData = #{ node => Node
+                          , ram_copies => Ram
+                          , disc_copies => Disc
+                          , disc_only_copies => DiscOnly
+                          , ext => Ext
+                          , schema_ops => SchemaOps
+                          },
+            try Fun(Tid, CommitData)
+            catch EC:Err:St ->
+                    CommitTabs = commit_tabs(Ram, Disc, DiscOnly, Ext),
+                    mnesia_lib:dbg_out("Mnesia post_commit hook failed: ~p:~p~nStacktrace:~p~nCommit tables:~p~n",
+                                       [EC, Err, stack_without_args(St), CommitTabs])
+            end,
+            ok
+    end.
+
+%% May be helpful for debugging
+commit_tabs(Ram, Disc, DiscOnly, Ext) ->
+    Acc = tabs_from_ops(Ram, []),
+    Acc1 = tabs_from_ops(Disc, Acc),
+    Acc2 = tabs_from_ops(DiscOnly, Acc1),
+    lists:uniq(tabs_from_ops(Ext, Acc2)).
+
+tabs_from_ops([{{Tab, _K}, _Val, _Op} | T], Acc) ->
+    tabs_from_ops(T, [Tab | Acc]);
+tabs_from_ops([_ | T], Acc) ->
+    tabs_from_ops(T, Acc);
+tabs_from_ops([], Acc) ->
+    Acc.
+
+%% Args may contain sensitive data
+stack_without_args([{M, F, Args, Info} | T]) when is_list(Args) ->
+    [{M, F, length(Args), Info} | stack_without_args(T)];
+stack_without_args([StItem | T] ) ->
+    [StItem | stack_without_args(T)];
+stack_without_args([]) ->
+    [].

lib/mnesia/src/mnesia_loader.erl

@@ -210,7 +210,19 @@ do_get_network_copy(Tab, _Reason, _Ns, unknown, _Cs) ->
     verbose("Local table copy of ~tp has recently been deleted, ignored.~n", [Tab]),
     {not_loaded, storage_unknown};
 do_get_network_copy(Tab, Reason, Ns, Storage, Cs) ->
-    [Node | Tail] = Ns,
+    [Node | Tail] =
+        case ?catch_val(copy_from_node) of
+            undefined -> Ns;
+            CPNode when is_atom(CPNode) ->
+                case lists:member(CPNode, Ns) of
+                    true ->
+                        [CPNode | Ns -- [CPNode]];
+                    false ->
+                        Ns
+                end;
+            _ ->
+                Ns
+        end,
     case lists:member(Node,val({current, db_nodes})) of
 	true ->
 	    dbg_out("Getting table ~tp (~p) from node ~p: ~tp~n",
@@ -914,7 +926,16 @@ get_chunk_func(Pid, Tab, {ext, Alias, Mod}, RemoteS) ->
 get_chunk_func(Pid, Tab, Storage, RemoteS) ->
     try
         TabSize = mnesia:table_info(Tab, size),
-        KeysPerTransfer = calc_nokeys(Storage, Tab),
+        KeysPerTransfer =
+            case ?catch_val(send_table_batch_size) of
+                {'EXIT', _} ->
+                    mnesia_lib:set(send_table_batch_size, 0),
+                    calc_nokeys(Storage, Tab);
+                0 ->
+                    calc_nokeys(Storage, Tab);
+                Val when is_integer(Val) ->
+                    Val
+            end,
         ChunkData = dets:info(Tab, bchunk_format),
         UseDetsChunk =
             Storage == RemoteS andalso

lib/mnesia/src/mnesia_monitor.erl

@@ -690,8 +690,10 @@ env() ->
      pid_sort_order,
      no_table_loaders,
      dc_dump_limit,
+     copy_from_node,
      send_compressed,
      max_transfer_size,
+     send_table_batch_size,
      schema
     ].
 
@@ -740,10 +742,14 @@ default_env(no_table_loaders) ->
     2;
 default_env(dc_dump_limit) ->
     4;
+default_env(copy_from_node) ->
+    undefined;
 default_env(send_compressed) ->
     0;
 default_env(max_transfer_size) ->
     64000;
+default_env(send_table_batch_size) ->
+    0;
 default_env(schema) ->
     [].
 
@@ -792,8 +798,10 @@ do_check_type(pid_sort_order, "standard") -> standard;
 do_check_type(pid_sort_order, _) -> false;
 do_check_type(no_table_loaders, N) when is_integer(N), N > 0 -> N;
 do_check_type(dc_dump_limit,N) when is_number(N), N > 0 -> N;
+do_check_type(copy_from_node, L) when is_atom(L) -> L;
 do_check_type(send_compressed, L) when is_integer(L), L >= 0, L =< 9 -> L;
 do_check_type(max_transfer_size, N) when is_integer(N), N > 0 -> N;
+do_check_type(send_table_batch_size, L) when is_integer(L), L >= 0 -> L;
 do_check_type(schema, L) when is_list(L) -> L.
 
 bool(true) -> true;

lib/mnesia/src/mnesia_tm.erl

@@ -1832,6 +1832,7 @@ do_commit(Tid, C, DumperMode) ->
     R4 = do_update(Tid, disc_only_copies, C#commit.disc_only_copies, R3),
     R5 = do_update_ext(Tid, C#commit.ext, R4),
     mnesia_subscr:report_activity(Tid),
+    mnesia_hook:do_post_commit(Tid, C),
     R5.
 
 %% This could/should be optimized

lib/ssl/src/ssl.erl

@@ -1589,6 +1589,7 @@ ssl_options() ->
      middlebox_comp_mode,
      max_fragment_length,
      next_protocol_selector,  next_protocols_advertised,
+     certificate_status,
      ocsp_stapling, ocsp_responder_certs, ocsp_nonce,
      padding_check,
      partial_chain,
@@ -1622,10 +1623,11 @@ update_options(Opts, Role, InheritedSslOpts) when is_map(InheritedSslOpts) ->
     {UserSslOpts, _} = split_options(Opts, ssl_options()),
     process_options(UserSslOpts, InheritedSslOpts, #{role => Role}).
 
-process_options(UserSslOpts, SslOpts0, Env) ->
+process_options(UserSslOpts, SslOpts00, Env) ->
     %% Reverse option list so we get the last set option if set twice,
     %% users depend on it.
     UserSslOptsMap = proplists:to_map(lists:reverse(UserSslOpts)),
+    SslOpts0  = opt_certificate_status(UserSslOptsMap, SslOpts00, Env),
     SslOpts1  = opt_protocol_versions(UserSslOptsMap, SslOpts0, Env),
     SslOpts2  = opt_verification(UserSslOptsMap, SslOpts1, Env),
     SslOpts3  = opt_certs(UserSslOptsMap, SslOpts2, Env),
@@ -1982,6 +1984,15 @@ opt_tickets(UserOpts, #{versions := Versions} = Opts, #{role := server}) ->
     Opts#{session_tickets => SessionTickets, early_data => EarlyData,
           anti_replay => AntiReplay, stateless_tickets_seed => STS}.
 
+opt_certificate_status(UserOpts, Opts, #{role := _Role}) ->
+    {_, CertificateStatus} = get_opt(certificate_status, undefined, UserOpts, Opts),
+    case CertificateStatus of
+        undefined -> ok;
+        #certificate_status{} -> ok;
+        _Value -> option_error(certificate_status, CertificateStatus)
+    end,
+    Opts#{certificate_status => CertificateStatus}.
+
 opt_ocsp(UserOpts, #{versions := _Versions} = Opts, #{role := Role}) ->
     {Stapling, SMap} =
         case get_opt(ocsp_stapling, ?DEFAULT_OCSP_STAPLING, UserOpts, Opts) of