Merge pull request #10930 from thalesmg/max-int-timeout-v50

check maximum timeout values in schema (5.1)

apps/emqx/src/emqx_limiter/src/emqx_limiter_schema.erl

@@ -187,7 +187,7 @@ fields(client_opts) ->
             )},
         {max_retry_time,
             ?HOCON(
-                emqx_schema:duration(),
+                emqx_schema:timeout_duration(),
                 #{
                     desc => ?DESC(max_retry_time),
                     default => <<"1h">>,

apps/emqx/src/emqx_schema.erl

@@ -30,9 +30,19 @@
 -include_lib("hocon/include/hoconsc.hrl").
 -include_lib("logger.hrl").
 
+-define(MAX_INT_TIMEOUT_MS, 4294967295).
+%% floor(?MAX_INT_TIMEOUT_MS / 1000).
+-define(MAX_INT_TIMEOUT_S, 4294967).
+
 -type duration() :: integer().
 -type duration_s() :: integer().
 -type duration_ms() :: integer().
+%% ?MAX_INT_TIMEOUT is defined loosely in some OTP modules like
+%% `erpc', `rpc' `gen' and `peer', despite affecting `receive' blocks
+%% as well.  It's `2^32 - 1'.
+-type timeout_duration() :: 0..?MAX_INT_TIMEOUT_MS.
+-type timeout_duration_s() :: 0..?MAX_INT_TIMEOUT_S.
+-type timeout_duration_ms() :: 0..?MAX_INT_TIMEOUT_MS.
 -type bytesize() :: integer().
 -type wordsize() :: bytesize().
 -type percent() :: float().
@@ -56,6 +66,9 @@
 -typerefl_from_string({duration/0, emqx_schema, to_duration}).
 -typerefl_from_string({duration_s/0, emqx_schema, to_duration_s}).
 -typerefl_from_string({duration_ms/0, emqx_schema, to_duration_ms}).
+-typerefl_from_string({timeout_duration/0, emqx_schema, to_timeout_duration}).
+-typerefl_from_string({timeout_duration_s/0, emqx_schema, to_timeout_duration_s}).
+-typerefl_from_string({timeout_duration_ms/0, emqx_schema, to_timeout_duration_ms}).
 -typerefl_from_string({bytesize/0, emqx_schema, to_bytesize}).
 -typerefl_from_string({wordsize/0, emqx_schema, to_wordsize}).
 -typerefl_from_string({percent/0, emqx_schema, to_percent}).
@@ -91,6 +104,9 @@
     to_duration/1,
     to_duration_s/1,
     to_duration_ms/1,
+    to_timeout_duration/1,
+    to_timeout_duration_s/1,
+    to_timeout_duration_ms/1,
     mk_duration/2,
     to_bytesize/1,
     to_wordsize/1,
@@ -127,6 +143,9 @@
     duration/0,
     duration_s/0,
     duration_ms/0,
+    timeout_duration/0,
+    timeout_duration_s/0,
+    timeout_duration_ms/0,
     bytesize/0,
     wordsize/0,
     percent/0,
@@ -1037,7 +1056,7 @@ fields("mqtt_quic_listener") ->
             )},
         {"idle_timeout",
             sc(
-                duration_ms(),
+                timeout_duration_ms(),
                 #{
                     default => 0,
                     desc => ?DESC(fields_mqtt_quic_listener_idle_timeout),
@@ -1054,7 +1073,7 @@ fields("mqtt_quic_listener") ->
             )},
         {"handshake_idle_timeout",
             sc(
-                duration_ms(),
+                timeout_duration_ms(),
                 #{
                     default => <<"10s">>,
                     desc => ?DESC(fields_mqtt_quic_listener_handshake_idle_timeout),
@@ -1071,7 +1090,7 @@ fields("mqtt_quic_listener") ->
             )},
         {"keep_alive_interval",
             sc(
-                duration_ms(),
+                timeout_duration_ms(),
                 #{
                     default => 0,
                     desc => ?DESC(fields_mqtt_quic_listener_keep_alive_interval),
@@ -2637,6 +2656,37 @@ to_duration_ms(Str) ->
         _ -> {error, Str}
     end.
 
+-spec to_timeout_duration(Input) -> {ok, timeout_duration()} | {error, Input} when
+    Input :: string() | binary().
+to_timeout_duration(Str) ->
+    do_to_timeout_duration(Str, fun to_duration/1, ?MAX_INT_TIMEOUT_MS, "ms").
+
+-spec to_timeout_duration_ms(Input) -> {ok, timeout_duration_ms()} | {error, Input} when
+    Input :: string() | binary().
+to_timeout_duration_ms(Str) ->
+    do_to_timeout_duration(Str, fun to_duration_ms/1, ?MAX_INT_TIMEOUT_MS, "ms").
+
+-spec to_timeout_duration_s(Input) -> {ok, timeout_duration_s()} | {error, Input} when
+    Input :: string() | binary().
+to_timeout_duration_s(Str) ->
+    do_to_timeout_duration(Str, fun to_duration_s/1, ?MAX_INT_TIMEOUT_S, "s").
+
+do_to_timeout_duration(Str, Fn, Max, Unit) ->
+    case Fn(Str) of
+        {ok, I} ->
+            case I =< Max of
+                true ->
+                    {ok, I};
+                false ->
+                    Msg = lists:flatten(
+                        io_lib:format("timeout value too large (max: ~b ~s)", [Max, Unit])
+                    ),
+                    throw(Msg)
+            end;
+        Err ->
+            Err
+    end.
+
 to_bytesize(Str) ->
     case hocon_postprocess:bytesize(Str) of
         I when is_integer(I) -> {ok, I};

apps/emqx/test/emqx_proper_types.erl

@@ -45,7 +45,9 @@
     limited_atom/0,
     limited_latin_atom/0,
     printable_utf8/0,
-    printable_codepoint/0
+    printable_codepoint/0,
+    raw_duration/0,
+    large_raw_duration/0
 ]).
 
 %% Generic Types
@@ -629,6 +631,20 @@ printable_codepoint() ->
         {1, range(16#E000, 16#FFFD)}
     ]).
 
+raw_duration() ->
+    ?LET(
+        {Value, Unit},
+        {pos_integer(), oneof([<<"d">>, <<"h">>, <<"m">>, <<"s">>, <<"ms">>])},
+        <<(integer_to_binary(Value))/binary, Unit/binary>>
+    ).
+
+large_raw_duration() ->
+    ?LET(
+        {Value, Unit},
+        {range(1_000_000, inf), oneof([<<"d">>, <<"h">>, <<"m">>])},
+        <<(integer_to_binary(Value))/binary, Unit/binary>>
+    ).
+
 %%--------------------------------------------------------------------
 %% Iterators
 %%--------------------------------------------------------------------

apps/emqx/test/emqx_schema_tests.erl

@@ -809,3 +809,31 @@ set_envs([{_Name, _Value} | _] = Envs) ->
 
 unset_envs([{_Name, _Value} | _] = Envs) ->
     lists:map(fun({Name, _}) -> os:unsetenv(Name) end, Envs).
+
+timeout_types_test_() ->
+    [
+        ?_assertEqual(
+            {ok, 4294967295},
+            typerefl:from_string(emqx_schema:timeout_duration(), <<"4294967295ms">>)
+        ),
+        ?_assertEqual(
+            {ok, 4294967295},
+            typerefl:from_string(emqx_schema:timeout_duration_ms(), <<"4294967295ms">>)
+        ),
+        ?_assertEqual(
+            {ok, 4294967},
+            typerefl:from_string(emqx_schema:timeout_duration_s(), <<"4294967000ms">>)
+        ),
+        ?_assertThrow(
+            "timeout value too large (max: 4294967295 ms)",
+            typerefl:from_string(emqx_schema:timeout_duration(), <<"4294967296ms">>)
+        ),
+        ?_assertThrow(
+            "timeout value too large (max: 4294967295 ms)",
+            typerefl:from_string(emqx_schema:timeout_duration_ms(), <<"4294967296ms">>)
+        ),
+        ?_assertThrow(
+            "timeout value too large (max: 4294967 s)",
+            typerefl:from_string(emqx_schema:timeout_duration_s(), <<"4294967001ms">>)
+        )
+    ].

apps/emqx/test/props/prop_emqx_schema.erl

@@ -0,0 +1,99 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. 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.
+%%--------------------------------------------------------------------
+
+-module(prop_emqx_schema).
+
+-include_lib("proper/include/proper.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+-define(MAX_INT_TIMEOUT_MS, 4294967295).
+
+%%--------------------------------------------------------------------
+%% Helper fns
+%%--------------------------------------------------------------------
+
+parse(Value, Type) ->
+    typerefl:from_string(Type, Value).
+
+timeout_within_bounds(RawDuration) ->
+    case emqx_schema:to_duration_ms(RawDuration) of
+        {ok, I} when I =< ?MAX_INT_TIMEOUT_MS ->
+            true;
+        _ ->
+            false
+    end.
+
+parses_the_same(Value, Type1, Type2) ->
+    parse(Value, Type1) =:= parse(Value, Type2).
+
+%%--------------------------------------------------------------------
+%% Properties
+%%--------------------------------------------------------------------
+
+prop_timeout_duration_refines_duration() ->
+    ?FORALL(
+        RawDuration,
+        emqx_proper_types:raw_duration(),
+        ?IMPLIES(
+            timeout_within_bounds(RawDuration),
+            parses_the_same(RawDuration, emqx_schema:duration(), emqx_schema:timeout_duration())
+        )
+    ).
+
+prop_timeout_duration_ms_refines_duration_ms() ->
+    ?FORALL(
+        RawDuration,
+        emqx_proper_types:raw_duration(),
+        ?IMPLIES(
+            timeout_within_bounds(RawDuration),
+            parses_the_same(
+                RawDuration, emqx_schema:duration_ms(), emqx_schema:timeout_duration_ms()
+            )
+        )
+    ).
+
+prop_timeout_duration_s_refines_duration_s() ->
+    ?FORALL(
+        RawDuration,
+        emqx_proper_types:raw_duration(),
+        ?IMPLIES(
+            timeout_within_bounds(RawDuration),
+            parses_the_same(RawDuration, emqx_schema:duration_s(), emqx_schema:timeout_duration_s())
+        )
+    ).
+
+prop_timeout_duration_is_valid_for_receive_after() ->
+    ?FORALL(
+        RawDuration,
+        emqx_proper_types:large_raw_duration(),
+        ?IMPLIES(
+            not timeout_within_bounds(RawDuration),
+            begin
+                %% we have to use the the non-strict version, because it's invalid
+                {ok, Timeout} = parse(RawDuration, emqx_schema:duration()),
+                Ref = make_ref(),
+                timer:send_after(20, {Ref, ok}),
+                ?assertError(
+                    timeout_value,
+                    receive
+                        {Ref, ok} -> error(should_be_invalid)
+                    after Timeout -> error(should_be_invalid)
+                    end
+                ),
+                true
+            end
+        )
+    ).

apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse_connector.erl

@@ -93,7 +93,7 @@ fields(config) ->
             )},
         {connect_timeout,
             hoconsc:mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"15s">>,
                     desc => ?DESC("connect_timeout")

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub.erl

@@ -47,7 +47,7 @@ fields(bridge_config) ->
     [
         {connect_timeout,
             sc(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"15s">>,
                     desc => ?DESC("connect_timeout")
@@ -80,7 +80,7 @@ fields(bridge_config) ->
             )},
         {request_timeout,
             sc(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     required => false,
                     deprecated => {since, "e5.0.1"},

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.erl

@@ -144,7 +144,7 @@ request_config() ->
             )},
         {request_timeout,
             mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"15s">>,
                     desc => ?DESC("config_request_timeout")

apps/emqx_bridge_kafka/src/emqx_bridge_kafka.erl

@@ -165,20 +165,20 @@ fields("config") ->
                 }
             )},
         {connect_timeout,
-            mk(emqx_schema:duration_ms(), #{
+            mk(emqx_schema:timeout_duration_ms(), #{
                 default => <<"5s">>,
                 desc => ?DESC(connect_timeout)
             })},
         {min_metadata_refresh_interval,
             mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"3s">>,
                     desc => ?DESC(min_metadata_refresh_interval)
                 }
             )},
         {metadata_request_timeout,
-            mk(emqx_schema:duration_ms(), #{
+            mk(emqx_schema:timeout_duration_ms(), #{
                 default => <<"5s">>,
                 desc => ?DESC(metadata_request_timeout)
             })},

apps/emqx_bridge_pulsar/src/emqx_bridge_pulsar.erl

@@ -62,7 +62,7 @@ fields(config) ->
             )},
         {connect_timeout,
             mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"5s">>,
                     desc => ?DESC("connect_timeout")
@@ -86,11 +86,12 @@ fields(producer_opts) ->
                 default => <<"1MB">>, desc => ?DESC("producer_send_buffer")
             })},
         {sync_timeout,
-            mk(emqx_schema:duration_ms(), #{
+            mk(emqx_schema:timeout_duration_ms(), #{
                 default => <<"3s">>, desc => ?DESC("producer_sync_timeout")
             })},
         {retention_period,
             mk(
+                %% not used in a `receive ... after' block, just timestamp comparison
                 hoconsc:union([infinity, emqx_schema:duration_ms()]),
                 #{default => infinity, desc => ?DESC("producer_retention_period")}
             )},

apps/emqx_bridge_rabbitmq/src/emqx_bridge_rabbitmq_connector.erl

@@ -84,7 +84,7 @@ fields(config) ->
             )},
         {timeout,
             hoconsc:mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"5s">>,
                     desc => ?DESC("timeout")
@@ -100,7 +100,7 @@ fields(config) ->
             )},
         {publish_confirmation_timeout,
             hoconsc:mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"30s">>,
                     desc => ?DESC("timeout")
@@ -117,7 +117,7 @@ fields(config) ->
             )},
         {heartbeat,
             hoconsc:mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"30s">>,
                     desc => ?DESC("heartbeat")

apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq_connector.erl

@@ -58,12 +58,12 @@ fields(config) ->
             mk(binary(), #{default => <<>>, desc => ?DESC(security_token), sensitive => true})},
         {sync_timeout,
             mk(
-                emqx_schema:duration(),
+                emqx_schema:timeout_duration(),
                 #{default => <<"3s">>, desc => ?DESC(sync_timeout)}
             )},
         {refresh_interval,
             mk(
-                emqx_schema:duration(),
+                emqx_schema:timeout_duration(),
                 #{default => <<"3s">>, desc => ?DESC(refresh_interval)}
             )},
         {send_buffer,

apps/emqx_conf/src/emqx_conf_schema.erl

@@ -528,7 +528,7 @@ fields("node") ->
             )},
         {"crash_dump_seconds",
             sc(
-                emqx_schema:duration_s(),
+                emqx_schema:timeout_duration_s(),
                 #{
                     mapping => "vm_args.-env ERL_CRASH_DUMP_SECONDS",
                     default => <<"30s">>,
@@ -550,7 +550,7 @@ fields("node") ->
             )},
         {"dist_net_ticktime",
             sc(
-                emqx_schema:duration_s(),
+                emqx_schema:timeout_duration_s(),
                 #{
                     mapping => "vm_args.-kernel net_ticktime",
                     default => <<"2m">>,
@@ -821,7 +821,7 @@ fields("rpc") ->
             )},
         {"socket_keepalive_idle",
             sc(
-                emqx_schema:duration_s(),
+                emqx_schema:timeout_duration_s(),
                 #{
                     mapping => "gen_rpc.socket_keepalive_idle",
                     default => <<"15m">>,
@@ -830,7 +830,7 @@ fields("rpc") ->
             )},
         {"socket_keepalive_interval",
             sc(
-                emqx_schema:duration_s(),
+                emqx_schema:timeout_duration_s(),
                 #{
                     mapping => "gen_rpc.socket_keepalive_interval",
                     default => <<"75s">>,
@@ -972,7 +972,7 @@ fields("log_overload_kill") ->
             )},
         {"restart_after",
             sc(
-                hoconsc:union([emqx_schema:duration_ms(), infinity]),
+                hoconsc:union([emqx_schema:timeout_duration_ms(), infinity]),
                 #{
                     default => <<"5s">>,
                     desc => ?DESC("log_overload_kill_restart_after")

apps/emqx_connector/src/emqx_connector_http.erl

@@ -67,7 +67,7 @@ fields(config) ->
     [
         {connect_timeout,
             sc(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"15s">>,
                     desc => ?DESC("connect_timeout")
@@ -80,7 +80,7 @@ fields(config) ->
             )},
         {retry_interval,
             sc(
-                emqx_schema:duration(),
+                emqx_schema:timeout_duration(),
                 #{deprecated => {since, "5.0.4"}}
             )},
         {pool_type,
@@ -138,7 +138,7 @@ fields("request") ->
             )},
         {request_timeout,
             sc(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     required => false,
                     desc => ?DESC("request_timeout")

apps/emqx_connector/src/emqx_connector_ldap.erl

@@ -201,5 +201,5 @@ port(type) -> integer();
 port(default) -> 389;
 port(_) -> undefined.
 
-duration(type) -> emqx_schema:duration_ms();
+duration(type) -> emqx_schema:timeout_duration_ms();
 duration(_) -> undefined.

apps/emqx_connector/src/emqx_connector_mongo.erl

@@ -108,7 +108,7 @@ fields(topology) ->
         {wait_queue_timeout_ms, duration("wait_queue_timeout")},
         {heartbeat_frequency_ms,
             hoconsc:mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"200s">>,
                     desc => ?DESC("heartbeat_period")
@@ -422,7 +422,7 @@ r_mode(_) -> undefined.
 
 duration(Desc) ->
     #{
-        type => emqx_schema:duration_ms(),
+        type => emqx_schema:timeout_duration_ms(),
         required => false,
         desc => ?DESC(Desc)
     }.

apps/emqx_dashboard/src/emqx_dashboard_schema.erl

@@ -38,7 +38,7 @@ fields("dashboard") ->
         {default_password, fun default_password/1},
         {sample_interval,
             ?HOCON(
-                emqx_schema:duration_s(),
+                emqx_schema:timeout_duration_s(),
                 #{
                     default => <<"10s">>,
                     desc => ?DESC(sample_interval),

apps/emqx_dashboard/src/emqx_dashboard_swagger.erl

@@ -819,6 +819,12 @@ typename_to_spec("duration_s()", _Mod) ->
     #{type => string, example => <<"1h">>};
 typename_to_spec("duration_ms()", _Mod) ->
     #{type => string, example => <<"32s">>};
+typename_to_spec("timeout_duration()", _Mod) ->
+    #{type => string, example => <<"12m">>};
+typename_to_spec("timeout_duration_s()", _Mod) ->
+    #{type => string, example => <<"1h">>};
+typename_to_spec("timeout_duration_ms()", _Mod) ->
+    #{type => string, example => <<"32s">>};
 typename_to_spec("percent()", _Mod) ->
     #{type => number, example => <<"12%">>};
 typename_to_spec("file()", _Mod) ->

apps/emqx_dashboard/test/emqx_swagger_remote_schema.erl

@@ -32,7 +32,7 @@ fields("root") ->
             )},
         {default_username, fun default_username/1},
         {default_password, fun default_password/1},
-        {sample_interval, mk(emqx_schema:duration_s(), #{default => <<"10s">>})},
+        {sample_interval, mk(emqx_schema:timeout_duration_s(), #{default => <<"10s">>})},
         {token_expired_time, mk(emqx_schema:duration(), #{default => <<"30m">>})}
     ];
 fields("ref1") ->

apps/emqx_dashboard/test/emqx_swagger_requestBody_SUITE.erl

@@ -751,7 +751,7 @@ schema("/object") ->
         {per_page, mk(range(1, 100), #{required => true, desc => <<"good per page desc">>})},
         {timeout,
             mk(
-                hoconsc:union([infinity, emqx_schema:duration_s()]),
+                hoconsc:union([infinity, emqx_schema:timeout_duration_s()]),
                 #{default => 5, required => true}
             )},
         {inner_ref, mk(hoconsc:ref(?MODULE, good_ref), #{})}
@@ -761,7 +761,7 @@ schema("/nest/object") ->
         {per_page, mk(range(1, 100), #{desc => <<"good per page desc">>})},
         {timeout,
             mk(
-                hoconsc:union([infinity, emqx_schema:duration_s()]),
+                hoconsc:union([infinity, emqx_schema:timeout_duration_s()]),
                 #{default => 5, required => true}
             )},
         {nest_object, [
@@ -785,7 +785,7 @@ schema("/ref/array/with/key") ->
         {per_page, mk(range(1, 100), #{desc => <<"good per page desc">>})},
         {timeout,
             mk(
-                hoconsc:union([infinity, emqx_schema:duration_s()]),
+                hoconsc:union([infinity, emqx_schema:timeout_duration_s()]),
                 #{default => 5, required => true}
             )},
         {array_refs, mk(hoconsc:array(hoconsc:ref(?MODULE, good_ref)), #{})}

apps/emqx_dashboard/test/emqx_swagger_response_SUITE.erl

@@ -573,7 +573,7 @@ schema("/object") ->
         {per_page, mk(range(1, 100), #{required => true, desc => <<"good per page desc">>})},
         {timeout,
             mk(
-                hoconsc:union([infinity, emqx_schema:duration_s()]),
+                hoconsc:union([infinity, emqx_schema:timeout_duration_s()]),
                 #{default => 5, required => true}
             )},
         {inner_ref, mk(hoconsc:ref(?MODULE, good_ref), #{})}
@@ -584,7 +584,7 @@ schema("/nest/object") ->
         {per_page, mk(range(1, 100), #{desc => <<"good per page desc">>})},
         {timeout,
             mk(
-                hoconsc:union([infinity, emqx_schema:duration_s()]),
+                hoconsc:union([infinity, emqx_schema:timeout_duration_s()]),
                 #{default => 5, required => true}
             )},
         {nest_object, [
@@ -613,13 +613,14 @@ schema("/ref/array/with/key") ->
         {per_page, mk(range(1, 100), #{desc => <<"good per page desc">>})},
         {timeout,
             mk(
-                hoconsc:union([infinity, emqx_schema:duration_s()]),
+                hoconsc:union([infinity, emqx_schema:timeout_duration_s()]),
                 #{default => 5, required => true}
             )},
         {assert, mk(float(), #{desc => <<"money">>})},
         {number_ex, mk(number(), #{desc => <<"number example">>})},
         {percent_ex, mk(emqx_schema:percent(), #{desc => <<"percent example">>})},
-        {duration_ms_ex, mk(emqx_schema:duration_ms(), #{desc => <<"duration ms example">>})},
+        {duration_ms_ex,
+            mk(emqx_schema:timeout_duration_ms(), #{desc => <<"duration ms example">>})},
         {atom_ex, mk(atom(), #{desc => <<"atom ex">>})},
         {array_refs, mk(hoconsc:array(hoconsc:ref(?MODULE, good_ref)), #{})}
     ]);

apps/emqx_exhook/src/emqx_exhook.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_exhook, [
     {description, "EMQX Extension for Hook"},
-    {vsn, "5.0.12"},
+    {vsn, "5.0.13"},
     {modules, []},
     {registered, []},
     {mod, {emqx_exhook_app, []}},

apps/emqx_exhook/src/emqx_exhook_schema.erl

@@ -63,7 +63,7 @@ fields(server) ->
                 example => <<"http://127.0.0.1:9000">>
             })},
         {request_timeout,
-            ?HOCON(emqx_schema:duration(), #{
+            ?HOCON(emqx_schema:timeout_duration(), #{
                 default => <<"5s">>,
                 desc => ?DESC(request_timeout)
             })},
@@ -74,7 +74,7 @@ fields(server) ->
                 default => #{<<"keepalive">> => true, <<"nodelay">> => true}
             })},
         {auto_reconnect,
-            ?HOCON(hoconsc:union([false, emqx_schema:duration()]), #{
+            ?HOCON(hoconsc:union([false, emqx_schema:timeout_duration()]), #{
                 default => <<"60s">>,
                 desc => ?DESC(auto_reconnect)
             })},

apps/emqx_ft/src/emqx_ft_schema.erl

@@ -66,7 +66,7 @@ fields(file_transfer) ->
             )},
         {init_timeout,
             mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     desc => ?DESC("init_timeout"),
                     required => false,
@@ -75,7 +75,7 @@ fields(file_transfer) ->
             )},
         {store_segment_timeout,
             mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     desc => ?DESC("store_segment_timeout"),
                     required => false,
@@ -84,7 +84,7 @@ fields(file_transfer) ->
             )},
         {assemble_timeout,
             mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     desc => ?DESC("assemble_timeout"),
                     required => false,
@@ -195,7 +195,7 @@ fields(local_storage_segments_gc) ->
     [
         {interval,
             mk(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     desc => ?DESC("storage_gc_interval"),
                     required => false,
@@ -204,6 +204,7 @@ fields(local_storage_segments_gc) ->
             )},
         {maximum_segments_ttl,
             mk(
+                %% not used in a `receive ... after' block, just timestamp comparison
                 emqx_schema:duration_s(),
                 #{
                     desc => ?DESC("storage_gc_max_segments_ttl"),
@@ -213,6 +214,7 @@ fields(local_storage_segments_gc) ->
             )},
         {minimum_segments_ttl,
             mk(
+                %% not used in a `receive ... after' block, just timestamp comparison
                 emqx_schema:duration_s(),
                 #{
                     desc => ?DESC("storage_gc_min_segments_ttl"),

apps/emqx_gateway_coap/src/emqx_coap_api.erl

@@ -105,7 +105,7 @@ request_body() ->
     [
         {token, mk(binary(), #{desc => ?DESC(token)})},
         {method, mk(enum([get, put, post, delete]), #{desc => ?DESC(method)})},
-        {timeout, mk(emqx_schema:duration_ms(), #{desc => ?DESC(timeout)})},
+        {timeout, mk(emqx_schema:timeout_duration_ms(), #{desc => ?DESC(timeout)})},
         {content_type,
             mk(
                 enum(['text/plain', 'application/json', 'application/octet-stream']),

apps/emqx_node_rebalance/src/emqx_node_rebalance.app.src

@@ -1,6 +1,6 @@
 {application, emqx_node_rebalance, [
     {description, "EMQX Node Rebalance"},
-    {vsn, "5.0.1"},
+    {vsn, "5.0.2"},
     {registered, [
         emqx_node_rebalance_sup,
         emqx_node_rebalance,

apps/emqx_node_rebalance/src/emqx_node_rebalance_api.erl

@@ -356,7 +356,7 @@ fields(rebalance_start) ->
     [
         {"wait_health_check",
             mk(
-                emqx_schema:duration_s(),
+                emqx_schema:timeout_duration_s(),
                 #{
                     desc => ?DESC(wait_health_check),
                     required => false
@@ -414,7 +414,7 @@ fields(rebalance_start) ->
             )},
         {"wait_takeover",
             mk(
-                emqx_schema:duration_s(),
+                emqx_schema:timeout_duration_s(),
                 #{
                     desc => ?DESC(wait_takeover),
                     required => false

apps/emqx_prometheus/src/emqx_prometheus_schema.erl

@@ -48,7 +48,7 @@ fields("prometheus") ->
             )},
         {interval,
             ?HOCON(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"15s">>,
                     required => true,

apps/emqx_resource/src/schema/emqx_resource_schema.erl

@@ -76,18 +76,18 @@ worker_pool_size(default) -> ?WORKER_POOL_SIZE;
 worker_pool_size(required) -> false;
 worker_pool_size(_) -> undefined.
 
-resume_interval(type) -> emqx_schema:duration_ms();
+resume_interval(type) -> emqx_schema:timeout_duration_ms();
 resume_interval(importance) -> ?IMPORTANCE_HIDDEN;
 resume_interval(desc) -> ?DESC("resume_interval");
 resume_interval(required) -> false;
 resume_interval(_) -> undefined.
 
-metrics_flush_interval(type) -> emqx_schema:duration_ms();
+metrics_flush_interval(type) -> emqx_schema:timeout_duration_ms();
 metrics_flush_interval(importance) -> ?IMPORTANCE_HIDDEN;
 metrics_flush_interval(required) -> false;
 metrics_flush_interval(_) -> undefined.
 
-health_check_interval(type) -> emqx_schema:duration_ms();
+health_check_interval(type) -> emqx_schema:timeout_duration_ms();
 health_check_interval(desc) -> ?DESC("health_check_interval");
 health_check_interval(default) -> ?HEALTHCHECK_INTERVAL_RAW;
 health_check_interval(required) -> false;
@@ -115,7 +115,7 @@ start_after_created(default) -> ?START_AFTER_CREATED_RAW;
 start_after_created(required) -> false;
 start_after_created(_) -> undefined.
 
-start_timeout(type) -> emqx_schema:duration_ms();
+start_timeout(type) -> emqx_schema:timeout_duration_ms();
 start_timeout(desc) -> ?DESC("start_timeout");
 start_timeout(default) -> ?START_TIMEOUT_RAW;
 start_timeout(required) -> false;
@@ -133,7 +133,7 @@ query_mode(default) -> async;
 query_mode(required) -> false;
 query_mode(_) -> undefined.
 
-request_ttl(type) -> hoconsc:union([emqx_schema:duration_ms(), infinity]);
+request_ttl(type) -> hoconsc:union([emqx_schema:timeout_duration_ms(), infinity]);
 request_ttl(aliases) -> [request_timeout];
 request_ttl(desc) -> ?DESC("request_ttl");
 request_ttl(default) -> ?DEFAULT_REQUEST_TTL_RAW;
@@ -166,7 +166,7 @@ batch_size(default) -> ?DEFAULT_BATCH_SIZE;
 batch_size(required) -> false;
 batch_size(_) -> undefined.
 
-batch_time(type) -> emqx_schema:duration_ms();
+batch_time(type) -> emqx_schema:timeout_duration_ms();
 batch_time(desc) -> ?DESC("batch_time");
 batch_time(default) -> ?DEFAULT_BATCH_TIME_RAW;
 batch_time(required) -> false;
@@ -196,6 +196,10 @@ desc("creation_opts") -> ?DESC("creation_opts").
 
 get_value_with_unit(Value) when is_integer(Value) ->
     <<(erlang:integer_to_binary(Value))/binary, "ms">>;
+get_value_with_unit(Value) when is_list(Value) ->
+    %% Must ensure it's a binary, otherwise formatting the error
+    %% message will fail.
+    list_to_binary(Value);
 get_value_with_unit(Value) ->
     Value.
 

apps/emqx_resource/test/emqx_resource_schema_tests.erl

@@ -0,0 +1,108 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. 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.
+%%--------------------------------------------------------------------
+
+-module(emqx_resource_schema_tests).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+
+%%===========================================================================
+%% Test cases
+%%===========================================================================
+
+health_check_interval_validator_test_() ->
+    [
+        ?_assertMatch(
+            #{<<"resource_opts">> := #{<<"health_check_interval">> := 150_000}},
+            parse_and_check_webhook_bridge(webhook_bridge_health_check_hocon(<<"150s">>))
+        ),
+        ?_assertMatch(
+            #{<<"resource_opts">> := #{<<"health_check_interval">> := 3_600_000}},
+            parse_and_check_webhook_bridge(webhook_bridge_health_check_hocon(<<"3600000ms">>))
+        ),
+        ?_assertThrow(
+            {_, [
+                #{
+                    kind := validation_error,
+                    reason := <<"Health Check Interval (3600001ms) is out of range", _/binary>>,
+                    value := 3600001
+                }
+            ]},
+            parse_and_check_webhook_bridge(webhook_bridge_health_check_hocon(<<"3600001ms">>))
+        ),
+        {"bad parse: negative number",
+            ?_assertThrow(
+                {_, [
+                    #{
+                        kind := validation_error,
+                        reason := <<"Health Check Interval (-10ms) is out of range", _/binary>>,
+                        value := "-10ms"
+                    }
+                ]},
+                parse_and_check_webhook_bridge(webhook_bridge_health_check_hocon(<<"-10ms">>))
+            )},
+        {"bad parse: underscores",
+            ?_assertThrow(
+                {_, [
+                    #{
+                        kind := validation_error,
+                        reason :=
+                            <<"Health Check Interval (3_600_000ms) is out of range", _/binary>>,
+                        value := "3_600_000ms"
+                    }
+                ]},
+                parse_and_check_webhook_bridge(webhook_bridge_health_check_hocon(<<"3_600_000ms">>))
+            )},
+        ?_assertThrow(
+            #{exception := "timeout value too large" ++ _},
+            parse_and_check_webhook_bridge(
+                webhook_bridge_health_check_hocon(<<"150000000000000s">>)
+            )
+        )
+    ].
+
+%%===========================================================================
+%% Helper functions
+%%===========================================================================
+
+parse_and_check_webhook_bridge(Hocon) ->
+    #{<<"bridges">> := #{<<"webhook">> := #{<<"simple">> := Conf}}} = check(parse(Hocon)),
+    Conf.
+
+parse(Hocon) ->
+    {ok, Conf} = hocon:binary(Hocon),
+    Conf.
+
+check(Conf) when is_map(Conf) ->
+    hocon_tconf:check_plain(emqx_bridge_schema, Conf).
+
+%%===========================================================================
+%% Data section
+%%===========================================================================
+
+%% erlfmt-ignore
+webhook_bridge_health_check_hocon(HealthCheckInterval) ->
+io_lib:format(
+"""
+bridges.webhook.simple {
+  url = \"http://localhost:4000\"
+  body = \"body\"
+  resource_opts {
+    health_check_interval = \"~s\"
+  }
+}
+""",
+[HealthCheckInterval]).

apps/emqx_retainer/src/emqx_retainer_schema.erl

@@ -45,13 +45,14 @@ fields("retainer") ->
         {enable, sc(boolean(), enable, true)},
         {msg_expiry_interval,
             sc(
+                %% not used in a `receive ... after' block, just timestamp comparison
                 emqx_schema:duration_ms(),
                 msg_expiry_interval,
                 <<"0s">>
             )},
         {msg_clear_interval,
             sc(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 msg_clear_interval,
                 <<"0s">>
             )},

apps/emqx_rule_engine/src/emqx_rule_engine_schema.erl

@@ -216,7 +216,7 @@ rule_engine_settings() ->
             ?HOCON(boolean(), #{default => true, desc => ?DESC("rule_engine_ignore_sys_message")})},
         {jq_function_default_timeout,
             ?HOCON(
-                emqx_schema:duration_ms(),
+                emqx_schema:timeout_duration_ms(),
                 #{
                     default => <<"10s">>,
                     desc => ?DESC("rule_engine_jq_function_default_timeout")

apps/emqx_s3/src/emqx_s3.app.src

@@ -1,6 +1,6 @@
 {application, emqx_s3, [
     {description, "EMQX S3"},
-    {vsn, "5.0.7"},
+    {vsn, "5.0.8"},
     {modules, []},
     {registered, [emqx_s3_sup]},
     {applications, [

apps/emqx_s3/src/emqx_s3_schema.erl

@@ -68,6 +68,7 @@ fields(s3) ->
             )},
         {url_expire_time,
             mk(
+                %% not used in a `receive ... after' block, just timestamp comparison
                 emqx_schema:duration_s(),
                 #{
                     default => "1h",

apps/emqx_slow_subs/src/emqx_slow_subs.app.src

@@ -1,7 +1,7 @@
 {application, emqx_slow_subs, [
     {description, "EMQX Slow Subscribers Statistics"},
     % strict semver, bump manually!
-    {vsn, "1.0.5"},
+    {vsn, "1.0.6"},
     {modules, []},
     {registered, [emqx_slow_subs_sup]},
     {applications, [kernel, stdlib, emqx]},

apps/emqx_slow_subs/src/emqx_slow_subs_schema.erl

@@ -30,12 +30,14 @@ fields("slow_subs") ->
         {enable, sc(boolean(), false, enable)},
         {threshold,
             sc(
+                %% not used in a `receive ... after' block, just timestamp comparison
                 emqx_schema:duration_ms(),
                 <<"500ms">>,
                 threshold
             )},
         {expire_interval,
             sc(
+                %% not used in a `receive ... after' block, just timestamp comparison
                 emqx_schema:duration_ms(),
                 <<"300s">>,
                 expire_interval

changes/ce/fix-10930.en.md

@@ -0,0 +1,3 @@
+Added a schema validation for values that might be used in timeouts to avoid invalid values.
+
+Before this fix, it was possible to use absurd values in the schema that would exceed the system limit, causing a crash.