diff --git a/apps/emqx_durable_storage/src/emqx_ds_message_storage_bitmask.erl b/apps/emqx_durable_storage/src/emqx_ds_message_storage_bitmask.erl
index 57608e5cb..74a50c302 100644
--- a/apps/emqx_durable_storage/src/emqx_ds_message_storage_bitmask.erl
+++ b/apps/emqx_durable_storage/src/emqx_ds_message_storage_bitmask.erl
@@ -277,13 +277,13 @@ store(DB = #db{handle = DBHandle, cf = CFHandle}, MessageID, PublishedAt, Topic,
     Value = make_message_value(Topic, MessagePayload),
     rocksdb:put(DBHandle, CFHandle, Key, Value, DB#db.write_options).
 
--spec make_iterator(db(), emqx_ds:replay()) ->
+-spec make_iterator(db(), emqx_ds_replay:replay()) ->
     {ok, iterator()} | {error, _TODO}.
 make_iterator(DB, Replay) ->
     Options = emqx_ds_conf:shard_iteration_options(DB#db.shard),
     make_iterator(DB, Replay, Options).
 
--spec make_iterator(db(), emqx_ds:replay(), iteration_options()) ->
+-spec make_iterator(db(), emqx_ds_replay:replay(), iteration_options()) ->
     % {error, invalid_start_time}? might just start from the beginning of time
     % and call it a day: client violated the contract anyway.
     {ok, iterator()} | {error, _TODO}.
@@ -337,7 +337,7 @@ preserve_iterator(#it{cursor = Cursor}) ->
     },
     term_to_binary(State).
 
--spec restore_iterator(db(), emqx_ds:replay(), binary()) ->
+-spec restore_iterator(db(), emqx_ds_replay:replay(), binary()) ->
     {ok, iterator()} | {error, _TODO}.
 restore_iterator(DB, Replay, Serial) when is_binary(Serial) ->
     State = binary_to_term(Serial),
@@ -419,7 +419,7 @@ hash(Input, Bits) ->
     % at most 32 bits
     erlang:phash2(Input, 1 bsl Bits).
 
--spec make_keyspace_filter(emqx_ds:replay(), keymapper()) -> keyspace_filter().
+-spec make_keyspace_filter(emqx_ds_replay:replay(), keymapper()) -> keyspace_filter().
 make_keyspace_filter({TopicFilter, StartTime}, Keymapper) ->
     Bitstring = compute_bitstring(TopicFilter, StartTime, Keymapper),
     HashBitmask = compute_topic_bitmask(TopicFilter, Keymapper),
diff --git a/apps/emqx_durable_storage/src/emqx_ds_storage_layer.erl b/apps/emqx_durable_storage/src/emqx_ds_storage_layer.erl
index 017423b02..7c73dafaf 100644
--- a/apps/emqx_durable_storage/src/emqx_ds_storage_layer.erl
+++ b/apps/emqx_durable_storage/src/emqx_ds_storage_layer.erl
@@ -63,7 +63,7 @@
 -record(it, {
     shard :: emqx_ds:shard(),
     gen :: gen_id(),
-    replay :: emqx_ds:replay(),
+    replay :: emqx_ds_replay:replay(),
     module :: module(),
     data :: term()
 }).
@@ -104,10 +104,10 @@
 -callback store(_Schema, binary(), emqx_ds:time(), emqx_ds:topic(), binary()) ->
     ok | {error, _}.
 
--callback make_iterator(_Schema, emqx_ds:replay()) ->
+-callback make_iterator(_Schema, emqx_ds_replay:replay()) ->
     {ok, _It} | {error, _}.
 
--callback restore_iterator(_Schema, emqx_ds:replay(), binary()) -> {ok, _It} | {error, _}.
+-callback restore_iterator(_Schema, emqx_ds_replay:replay(), binary()) -> {ok, _It} | {error, _}.
 
 -callback preserve_iterator(_Schema, _It) -> term().
 
@@ -132,7 +132,7 @@ store(Shard, GUID, Time, Topic, Msg) ->
     {_GenId, #{module := Mod, data := Data}} = meta_lookup_gen(Shard, Time),
     Mod:store(Data, GUID, Time, Topic, Msg).
 
--spec make_iterator(emqx_ds:shard(), emqx_ds:replay()) ->
+-spec make_iterator(emqx_ds:shard(), emqx_ds_replay:replay()) ->
     {ok, iterator()} | {error, _TODO}.
 make_iterator(Shard, Replay = {_, StartTime}) ->
     {GenId, Gen} = meta_lookup_gen(Shard, StartTime),
@@ -160,12 +160,12 @@ next(It = #it{module = Mod, data = ItData}) ->
             end
     end.
 
--spec preserve_iterator(iterator(), emqx_ds:replay_id()) ->
+-spec preserve_iterator(iterator(), emqx_ds_replay:replay_id()) ->
     ok | {error, _TODO}.
 preserve_iterator(It = #it{}, ReplayID) ->
     iterator_put_state(ReplayID, It).
 
--spec restore_iterator(emqx_ds:shard(), emqx_ds:replay_id()) ->
+-spec restore_iterator(emqx_ds:shard(), emqx_ds_replay:replay_id()) ->
     {ok, iterator()} | {error, _TODO}.
 restore_iterator(Shard, ReplayID) ->
     case iterator_get_state(Shard, ReplayID) of
diff --git a/apps/emqx_durable_storage/test/props/emqx_ds_message_storage_bitmask_shim.erl b/apps/emqx_durable_storage/test/props/emqx_ds_message_storage_bitmask_shim.erl
index 59668ca01..bbe16f518 100644
--- a/apps/emqx_durable_storage/test/props/emqx_ds_message_storage_bitmask_shim.erl
+++ b/apps/emqx_durable_storage/test/props/emqx_ds_message_storage_bitmask_shim.erl
@@ -29,7 +29,7 @@ store(Tab, MessageID, PublishedAt, Topic, Payload) ->
     true = ets:insert(Tab, {{PublishedAt, MessageID}, Topic, Payload}),
     ok.
 
--spec iterate(t(), emqx_ds:replay()) ->
+-spec iterate(t(), emqx_ds_replay:replay()) ->
     [binary()].
 iterate(Tab, {TopicFilter, StartTime}) ->
     ets:foldr(