diff --git a/apps/emqx/src/emqx_schema.erl b/apps/emqx/src/emqx_schema.erl
index ab3d7bf71..cf2317e03 100644
--- a/apps/emqx/src/emqx_schema.erl
+++ b/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,
@@ -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};
diff --git a/apps/emqx/test/emqx_proper_types.erl b/apps/emqx/test/emqx_proper_types.erl
index e1d95227b..c1b41c292 100644
--- a/apps/emqx/test/emqx_proper_types.erl
+++ b/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
 %%--------------------------------------------------------------------
diff --git a/apps/emqx/test/emqx_schema_tests.erl b/apps/emqx/test/emqx_schema_tests.erl
index 3dcfa331e..ad2341460 100644
--- a/apps/emqx/test/emqx_schema_tests.erl
+++ b/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">>)
+        )
+    ].
diff --git a/apps/emqx/test/props/prop_emqx_schema.erl b/apps/emqx/test/props/prop_emqx_schema.erl
new file mode 100644
index 000000000..5d5e8f017
--- /dev/null
+++ b/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
+        )
+    ).