fix outbox

apps/efka/src/iot/efka_iot_client.erl

@@ -29,6 +29,9 @@
 %% 激活状态下
 -define(STATE_ACTIVATED, activated).
 -define(SSL_PING_INTERVAL, 30000).
+-define(OUTBOX_SEGMENT_RECORD_LIMIT, 2000).
+-define(OUTBOX_MAX_SEGMENTS, 5).
+-define(OUTBOX_MAX_RECORD_BYTES, 16 * 1024 * 1024).
 
 -record(state, {
     socket :: undefined | ssl:sslsocket(),
@@ -78,7 +81,7 @@ start_link() ->
 
 -spec init(list()) -> {ok, atom(), #state{}}.
 init([]) ->
-    case efka_iot_outbox:open() of
+    case efka_iot_outbox:open(outbox_options()) of
         {ok, Outbox} ->
             erlang:start_timer(0, self(), create_transport),
             {ok, ?STATE_DISCONNECTED, #state{socket = undefined, outbox = Outbox}};
@@ -90,6 +93,16 @@ init([]) ->
 callback_mode() ->
     handle_event_function.
 
+-spec outbox_options() -> map().
+outbox_options() ->
+    {ok, DetsDir} = application:get_env(efka, dets_dir),
+    #{
+        dir => filename:join(DetsDir, "iot_outbox"),
+        segment_record_limit => ?OUTBOX_SEGMENT_RECORD_LIMIT,
+        max_segments => ?OUTBOX_MAX_SEGMENTS,
+        max_record_bytes => ?OUTBOX_MAX_RECORD_BYTES
+    }.
+
 %% 异步发送数据，连接存在时直接发送；否则写入持久化 outbox。
 -spec handle_event(term(), term(), atom(), #state{}) -> term().
 handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->

apps/efka/src/iot/efka_iot_outbox.erl

@@ -12,6 +12,13 @@
 %%%   writer_segment => N}。进程启动时仍然会扫描 segment 文件，并以 segment
 %%%   文件里的真实数据作为准确信息；因此即使 metadata 落后，也不会导致跳过或
 %%%   删除未消费的数据。
+%%% - outbox 目录通过 open/1 的 dir 参数传入，本模块不直接读取应用环境变量。
+%%%
+%%% open/1 参数：
+%%% - dir：outbox 数据目录。
+%%% - segment_record_limit：每个 segment 最多保存的 record 数。
+%%% - max_segments：最多保留的 live segment 数。
+%%% - max_record_bytes：允许读取的单条 record 最大 payload 字节数。
 %%%
 %%% record 格式：
 %%% - 4 字节 unsigned big-endian payload size。
@@ -20,9 +27,11 @@
 %%% 写入逻辑：
 %%% - `append/2' 将一条 record 写入当前 writer segment，随后 fsync 当前
 %%%   segment，推进 write_seq，并持久化 metadata。
-%%% - 每个 segment 最多保存 200000 条数据。
+%%% - 每个 segment 最多保存多少条数据由 open/1 的 segment_record_limit
-%%% - outbox 最多保留 5 个 live segment。当当前 segment 已满，且已经存在
+%%%   参数决定。
-%%%   5 个 live segment 时，新 packet 会被丢弃，已有磁盘数据保持不变。
+%%% - outbox 最多保留多少个 live segment 由 open/1 的 max_segments 参数
+%%%   决定。当当前 segment 已满，且 live segment 数已经达到上限时，新
+%%%   packet 会被丢弃，已有磁盘数据保持不变。
 %%% - 当当前 segment 已满，但 live segment 数量少于 5 个时，writer 会关闭
 %%%   旧文件，并开始写入 `segment-(N + 1).log'。
 %%%
@@ -44,7 +53,7 @@
 %%%-------------------------------------------------------------------
 -module(efka_iot_outbox).
 
--export([open/0, close/1, append/2, next/1, ack/2]).
+-export([open/1, close/1, append/2, next/1, ack/2]).
 -export_type([outbox/0]).
 
 -record(segment, {
@@ -60,6 +69,9 @@
     metadata_path :: file:filename_all(),
     writer_segment = 1 :: pos_integer(),
     fd :: file:fd(),
+    segment_record_limit :: pos_integer(),
+    max_segments :: pos_integer(),
+    max_record_bytes :: pos_integer(),
     segments = [] :: [#segment{}],
     next_seq = 1 :: pos_integer(),
     write_seq = 0 :: non_neg_integer(),
@@ -71,19 +83,25 @@
 -define(METADATA_FILE, "metadata.term").
 -define(SEGMENT_PREFIX, "segment-").
 -define(SEGMENT_EXT, ".log").
--define(SEGMENT_RECORD_LIMIT, 200000).
--define(MAX_SEGMENTS, 5).
--define(MAX_RECORD_BYTES, 16 * 1024 * 1024).
 
--spec open() -> {ok, outbox()} | {error, term()}.
+-type open_options() :: #{
-open() ->
+    dir := file:filename_all(),
-    {ok, DetsDir} = application:get_env(efka, dets_dir),
+    segment_record_limit := pos_integer(),
-    Dir = filename:join(DetsDir, "iot_outbox"),
+    max_segments := pos_integer(),
+    max_record_bytes := pos_integer()
+}.
+
+-spec open(open_options()) -> {ok, outbox()} | {error, term()}.
+open(Options) when is_map(Options) ->
+    Dir = maps:get(dir, Options),
+    SegmentRecordLimit = positive_option(segment_record_limit, Options),
+    MaxSegments = positive_option(max_segments, Options),
+    MaxRecordBytes = positive_option(max_record_bytes, Options),
     MetadataPath = filename:join(Dir, ?METADATA_FILE),
     maybe
         ok ?= ensure_dir(Dir),
         {_MetaWriteSeq, MetaAckedSeq, MetaWriterSegment} = read_metadata(MetadataPath),
-        {ok, Segments} ?= load_segments(Dir),
+        {ok, Segments} ?= load_segments(Dir, MaxRecordBytes),
         ScannedWriteSeq = max_segment_end(Segments),
         WriteSeq = ScannedWriteSeq,
         AckedSeq = min(MetaAckedSeq, WriteSeq),
@@ -94,6 +112,9 @@ open() ->
             metadata_path = MetadataPath,
             writer_segment = WriterSegment,
             fd = Fd,
+            segment_record_limit = SegmentRecordLimit,
+            max_segments = MaxSegments,
+            max_record_bytes = MaxRecordBytes,
             segments = Segments,
             next_seq = WriteSeq + 1,
             write_seq = WriteSeq,
@@ -135,14 +156,14 @@ append(Packet, Outbox0) when is_binary(Packet) ->
     end.
 
 -spec next(outbox()) -> eof | {ok, pos_integer(), binary()} | {error, term()}.
-next(#outbox{segments = Segments, acked_seq = AckedSeq}) ->
+next(#outbox{segments = Segments, acked_seq = AckedSeq, max_record_bytes = MaxRecordBytes}) ->
     case next_segment(Segments, AckedSeq) of
         undefined ->
             eof;
         #segment{path = Path} ->
             case file:open(Path, [read, binary]) of
                 {ok, Fd} ->
-                    Result = read_next_unacked(Fd, AckedSeq),
+                    Result = read_next_unacked(Fd, AckedSeq, MaxRecordBytes),
                     _ = file:close(Fd),
                     Result;
                 {error, enoent} ->
@@ -173,6 +194,15 @@ ack(Seq, #outbox{acked_seq = AckedSeq}) when is_integer(Seq) ->
 ensure_dir(Dir) ->
     filelib:ensure_dir(filename:join(Dir, "dummy")).
 
+-spec positive_option(atom(), map()) -> pos_integer().
+positive_option(Key, Options) ->
+    case maps:get(Key, Options) of
+        Value when is_integer(Value), Value > 0 ->
+            Value;
+        Value ->
+            error({invalid_option, Key, Value})
+    end.
+
 -spec read_metadata(file:filename_all()) ->
     {non_neg_integer(), non_neg_integer(), pos_integer()}.
 read_metadata(MetadataPath) ->
@@ -215,29 +245,29 @@ persist_metadata(#outbox{
             {error, Reason}
     end.
 
--spec load_segments(file:filename_all()) -> {ok, [#segment{}]} | {error, term()}.
+-spec load_segments(file:filename_all(), pos_integer()) -> {ok, [#segment{}]} | {error, term()}.
-load_segments(Dir) ->
+load_segments(Dir, MaxRecordBytes) ->
     case file:list_dir(Dir) of
         {ok, Names} ->
             Ids = lists:sort([Id || Name <- Names, {ok, Id} <- [parse_segment_id(Name)]]),
-            load_segments(Dir, Ids, []);
+            load_segments(Dir, Ids, MaxRecordBytes, []);
         {error, enoent} ->
             {ok, []};
         {error, Reason} ->
             {error, Reason}
     end.
 
--spec load_segments(file:filename_all(), [pos_integer()], [#segment{}]) ->
+-spec load_segments(file:filename_all(), [pos_integer()], pos_integer(), [#segment{}]) ->
     {ok, [#segment{}]} | {error, term()}.
-load_segments(_Dir, [], Acc) ->
+load_segments(_Dir, [], _MaxRecordBytes, Acc) ->
     {ok, lists:reverse(Acc)};
-load_segments(Dir, [Id | Rest], Acc) ->
+load_segments(Dir, [Id | Rest], MaxRecordBytes, Acc) ->
     Path = segment_path(Dir, Id),
-    case scan_segment(Path, Id) of
+    case scan_segment(Path, Id, MaxRecordBytes) of
         {ok, undefined} ->
-            load_segments(Dir, Rest, Acc);
+            load_segments(Dir, Rest, MaxRecordBytes, Acc);
         {ok, Segment} ->
-            load_segments(Dir, Rest, [Segment | Acc]);
+            load_segments(Dir, Rest, MaxRecordBytes, [Segment | Acc]);
         {error, Reason} ->
             {error, Reason}
     end.
@@ -290,9 +320,9 @@ ensure_writable_segment(Outbox = #outbox{segments = Segments, writer_segment = W
     case find_segment(WriterSegment, Segments) of
         undefined ->
             {ok, Outbox};
-        #segment{records = Records} when Records < ?SEGMENT_RECORD_LIMIT ->
+        #segment{records = Records} when Records < Outbox#outbox.segment_record_limit ->
             {ok, Outbox};
-        #segment{} when length(Segments) >= ?MAX_SEGMENTS ->
+        #segment{} when length(Segments) >= Outbox#outbox.max_segments ->
             {dropped, capacity_reached, Outbox};
         #segment{} ->
             rotate_writer(Outbox, WriterSegment + 1)
@@ -420,12 +450,12 @@ delete_files([Path | Rest]) ->
             {error, {delete_failed, Path, Reason}}
     end.
 
--spec scan_segment(file:filename_all(), pos_integer()) ->
+-spec scan_segment(file:filename_all(), pos_integer(), pos_integer()) ->
     {ok, #segment{} | undefined} | {error, term()}.
-scan_segment(Path, Id) ->
+scan_segment(Path, Id, MaxRecordBytes) ->
     case file:open(Path, [read, binary]) of
         {ok, Fd} ->
-            Result = scan_segment(Fd, Id, Path, 0, 0, 0),
+            Result = scan_segment(Fd, Id, Path, MaxRecordBytes, 0, 0, 0),
             _ = file:close(Fd),
             Result;
         {error, enoent} ->
@@ -434,11 +464,11 @@ scan_segment(Path, Id) ->
             {error, Reason}
     end.
 
--spec scan_segment(file:fd(), pos_integer(), file:filename_all(),
+-spec scan_segment(file:fd(), pos_integer(), file:filename_all(), pos_integer(),
     non_neg_integer(), non_neg_integer(), non_neg_integer()) ->
     {ok, #segment{} | undefined} | {error, term()}.
-scan_segment(Fd, Id, Path, StartSeq, EndSeq, Records) ->
+scan_segment(Fd, Id, Path, MaxRecordBytes, StartSeq, EndSeq, Records) ->
-    case read_record(Fd) of
+    case read_record(Fd, MaxRecordBytes) of
         eof when Records =:= 0 ->
             {ok, undefined};
         eof ->
@@ -454,7 +484,7 @@ scan_segment(Fd, Id, Path, StartSeq, EndSeq, Records) ->
                 0 -> Seq;
                 _ -> StartSeq
             end,
-            scan_segment(Fd, Id, Path, NStartSeq, Seq, Records + 1);
+            scan_segment(Fd, Id, Path, MaxRecordBytes, NStartSeq, Seq, Records + 1);
         {error, Reason} ->
             {error, Reason}
     end.
@@ -465,26 +495,26 @@ encode_record(Seq, Packet) ->
     Size = byte_size(Payload),
     <<Size:32/unsigned-big, Payload/binary>>.
 
--spec read_next_unacked(file:fd(), non_neg_integer()) ->
+-spec read_next_unacked(file:fd(), non_neg_integer(), pos_integer()) ->
     eof | {ok, pos_integer(), binary()} | {error, term()}.
-read_next_unacked(Fd, AckedSeq) ->
+read_next_unacked(Fd, AckedSeq, MaxRecordBytes) ->
-    case read_record(Fd) of
+    case read_record(Fd, MaxRecordBytes) of
         eof ->
             eof;
         {ok, Seq, Packet} when Seq > AckedSeq ->
             {ok, Seq, Packet};
         {ok, _Seq, _Packet} ->
-            read_next_unacked(Fd, AckedSeq);
+            read_next_unacked(Fd, AckedSeq, MaxRecordBytes);
         {error, Reason} ->
             {error, Reason}
     end.
 
--spec read_record(file:fd()) -> eof | {ok, pos_integer(), binary()} | {error, term()}.
+-spec read_record(file:fd(), pos_integer()) -> eof | {ok, pos_integer(), binary()} | {error, term()}.
-read_record(Fd) ->
+read_record(Fd, MaxRecordBytes) ->
     case file:read(Fd, 4) of
         eof ->
             eof;
-        {ok, <<Size:32/unsigned-big>>} when Size > 0, Size =< ?MAX_RECORD_BYTES ->
+        {ok, <<Size:32/unsigned-big>>} when Size > 0, Size =< MaxRecordBytes ->
             read_record_payload(Fd, Size);
         {ok, <<Size:32/unsigned-big>>} ->
             {error, {invalid_record_size, Size}};

apps/efka/src/tests/efka_iot_outbox_test.erl

@@ -0,0 +1,315 @@
+%%%-------------------------------------------------------------------
+%%% @doc
+%%% efka_iot_outbox 的手动模拟测试模块。
+%%%
+%%% 该模块基于 gen_server 实现，用两个定时 tick 模拟一边写入、一边读取：
+%%% - write_tick 每次向 efka_iot_outbox 写入一条 packet。
+%%% - read_tick 每次从 efka_iot_outbox 读取并 ack 一条 packet。
+%%% - 默认 read_interval = write_interval * 2，因此读取速度是写入速度的
+%%%   50%。
+%%% - 读取到的数据会通过 logger:debug/2 输出。
+%%%
+%%% 默认使用独立临时 dir，避免污染真实运行目录。也可以通过 start_link/1
+%%% 传入 {dir, Dir} 指定 outbox 目录。
+%%%
+%%% 示例：
+%%% <pre>
+%%% efka_iot_outbox_test:start_link([
+%%%     {dir, "/private/tmp/efka_iot_outbox_test"},
+%%%     {write_interval, 100},
+%%%     {read_interval, 200},
+%%%     {max_writes, 1000}
+%%% ]).
+%%% efka_iot_outbox_test:stats().
+%%% efka_iot_outbox_test:stop().
+%%% </pre>
+%%% @end
+%%%-------------------------------------------------------------------
+-module(efka_iot_outbox_test).
+
+-behaviour(gen_server).
+
+-export([start/0, start/1, start_link/0, start_link/1, stop/0, stats/0]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+-export([test/0]).
+
+-define(SERVER, ?MODULE).
+-define(DEFAULT_WRITE_INTERVAL, 100).
+-define(DEFAULT_SEGMENT_RECORD_LIMIT, 2000).
+-define(DEFAULT_MAX_SEGMENTS, 5).
+-define(DEFAULT_MAX_RECORD_BYTES, 16 * 1024 * 1024).
+
+-record(state, {
+    outbox :: efka_iot_outbox:outbox(),
+    dir :: file:filename_all(),
+    write_interval = ?DEFAULT_WRITE_INTERVAL :: pos_integer(),
+    read_interval = ?DEFAULT_WRITE_INTERVAL * 2 :: pos_integer(),
+    max_writes = infinity :: pos_integer() | infinity,
+    write_timer = undefined :: undefined | reference(),
+    read_timer = undefined :: undefined | reference(),
+    write_count = 0 :: non_neg_integer(),
+    appended_count = 0 :: non_neg_integer(),
+    read_count = 0 :: non_neg_integer(),
+    dropped_count = 0 :: non_neg_integer(),
+    error_count = 0 :: non_neg_integer()
+}).
+
+-type option() ::
+    {dir, file:filename_all()} |
+    {write_interval, pos_integer()} |
+    {read_interval, pos_integer()} |
+    {max_writes, pos_integer() | infinity} |
+    {segment_record_limit, pos_integer()} |
+    {max_segments, pos_integer()} |
+    {max_record_bytes, pos_integer()}.
+
+
+test() ->
+    efka_iot_outbox_test:start_link([
+        {write_interval, 10},
+        {read_interval, 20},
+        {max_writes, 100_0000}
+    ]).
+
+-spec start() -> {ok, pid()} | {error, term()}.
+start() ->
+    start([]).
+
+-spec start([option()] | map()) -> {ok, pid()} | {error, term()}.
+start(Options) ->
+    gen_server:start({local, ?SERVER}, ?MODULE, Options, []).
+
+-spec start_link() -> {ok, pid()} | {error, term()}.
+start_link() ->
+    start_link([]).
+
+-spec start_link([option()] | map()) -> {ok, pid()} | {error, term()}.
+start_link(Options) ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, Options, []).
+
+-spec stop() -> ok.
+stop() ->
+    gen_server:stop(?SERVER).
+
+-spec stats() -> map().
+stats() ->
+    gen_server:call(?SERVER, stats).
+
+-spec init([option()] | map()) -> {ok, #state{}} | {stop, term()}.
+init(Options0) ->
+    process_flag(trap_exit, true),
+    Options = normalize_options(Options0),
+    WriteInterval = positive_option(write_interval, Options, ?DEFAULT_WRITE_INTERVAL),
+    ReadInterval = positive_option(read_interval, Options, WriteInterval * 2),
+    SegmentRecordLimit = positive_option(segment_record_limit, Options, ?DEFAULT_SEGMENT_RECORD_LIMIT),
+    MaxSegments = positive_option(max_segments, Options, ?DEFAULT_MAX_SEGMENTS),
+    MaxRecordBytes = positive_option(max_record_bytes, Options, ?DEFAULT_MAX_RECORD_BYTES),
+    MaxWrites = max_writes_option(Options),
+
+    Dir = "/private/tmp/efka_outbox/",
+    ok = filelib:ensure_dir(filename:join(Dir, "dummy")),
+    OutboxOptions = #{
+        dir => Dir,
+        segment_record_limit => SegmentRecordLimit,
+        max_segments => MaxSegments,
+        max_record_bytes => MaxRecordBytes
+    },
+    case efka_iot_outbox:open(OutboxOptions) of
+        {ok, Outbox} ->
+            State0 = #state{
+                outbox = Outbox,
+                dir = Dir,
+                write_interval = WriteInterval,
+                read_interval = ReadInterval,
+                max_writes = MaxWrites
+            },
+            State1 = schedule_write(State0, 0),
+            State2 = schedule_read(State1, ReadInterval),
+            {ok, State2};
+        {error, Reason} ->
+            {stop, Reason}
+    end.
+
+-spec handle_call(term(), gen_server:from(), #state{}) ->
+    {reply, term(), #state{}}.
+handle_call(stats, _From, State) ->
+    {reply, state_stats(State), State};
+handle_call(_Request, _From, State) ->
+    {reply, {error, unknown_call}, State}.
+
+-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
+handle_info(write_tick, State0 = #state{write_timer = WriteTimer}) ->
+    State1 = State0#state{write_timer = undefined},
+    State2 = case can_write(State1) of
+        true ->
+            append_one(State1);
+        false ->
+            State1
+    end,
+    State3 = case can_write(State2) of
+        true ->
+            schedule_write(State2, State2#state.write_interval);
+        false ->
+            State2
+    end,
+    _ = WriteTimer,
+    {noreply, State3};
+handle_info(read_tick, State0 = #state{read_timer = ReadTimer}) ->
+    State1 = State0#state{read_timer = undefined},
+    State2 = read_one(State1),
+    State3 = schedule_read(State2, State2#state.read_interval),
+    _ = ReadTimer,
+    {noreply, State3};
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+-spec terminate(term(), #state{}) -> ok.
+terminate(_Reason, #state{outbox = Outbox, write_timer = WriteTimer, read_timer = ReadTimer}) ->
+    cancel_timer(WriteTimer),
+    cancel_timer(ReadTimer),
+    efka_iot_outbox:close(Outbox),
+    ok.
+
+-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+-spec append_one(#state{}) -> #state{}.
+append_one(State = #state{outbox = Outbox, write_count = WriteCount}) ->
+    PacketId = WriteCount + 1,
+    Packet = term_to_binary(#{
+        source => efka_iot_outbox_test,
+        packet_id => PacketId,
+        monotonic_time => erlang:monotonic_time(millisecond)
+    }),
+    case efka_iot_outbox:append(Packet, Outbox) of
+        {ok, NOutbox} ->
+            State#state{
+                outbox = NOutbox,
+                write_count = PacketId,
+                appended_count = State#state.appended_count + 1
+            };
+        {dropped, capacity_reached, NOutbox} ->
+            logger:warning("[efka_iot_outbox_test] outbox capacity reached, packet_id: ~p", [PacketId]),
+            State#state{
+                outbox = NOutbox,
+                write_count = PacketId,
+                dropped_count = State#state.dropped_count + 1
+            };
+        {error, Reason} ->
+            logger:warning("[efka_iot_outbox_test] append failed, packet_id: ~p, reason: ~p", [PacketId, Reason]),
+            State#state{
+                write_count = PacketId,
+                error_count = State#state.error_count + 1
+            }
+    end.
+
+-spec read_one(#state{}) -> #state{}.
+read_one(State = #state{outbox = Outbox}) ->
+    case efka_iot_outbox:next(Outbox) of
+        {ok, Seq, Packet} ->
+            logger:debug("[efka_iot_outbox_test] read seq: ~p, packet: ~p", [Seq, decode_packet(Packet)]),
+            case efka_iot_outbox:ack(Seq, Outbox) of
+                {ok, NOutbox} ->
+                    State#state{
+                        outbox = NOutbox,
+                        read_count = State#state.read_count + 1
+                    };
+                {error, Reason} ->
+                    logger:warning("[efka_iot_outbox_test] ack failed, seq: ~p, reason: ~p", [Seq, Reason]),
+                    State#state{error_count = State#state.error_count + 1}
+            end;
+        eof ->
+            State;
+        {error, Reason} ->
+            logger:warning("[efka_iot_outbox_test] read failed, reason: ~p", [Reason]),
+            State#state{error_count = State#state.error_count + 1}
+    end.
+
+-spec schedule_write(#state{}, non_neg_integer()) -> #state{}.
+schedule_write(State, Delay) ->
+    Ref = erlang:send_after(Delay, self(), write_tick),
+    State#state{write_timer = Ref}.
+
+-spec schedule_read(#state{}, non_neg_integer()) -> #state{}.
+schedule_read(State, Delay) ->
+    Ref = erlang:send_after(Delay, self(), read_tick),
+    State#state{read_timer = Ref}.
+
+-spec cancel_timer(undefined | reference()) -> ok.
+cancel_timer(undefined) ->
+    ok;
+cancel_timer(Ref) ->
+    _ = erlang:cancel_timer(Ref),
+    ok.
+
+-spec can_write(#state{}) -> boolean().
+can_write(#state{max_writes = infinity}) ->
+    true;
+can_write(#state{write_count = WriteCount, max_writes = MaxWrites}) ->
+    WriteCount < MaxWrites.
+
+-spec state_stats(#state{}) -> map().
+state_stats(#state{
+    dir = Dir,
+    write_interval = WriteInterval,
+    read_interval = ReadInterval,
+    max_writes = MaxWrites,
+    write_count = WriteCount,
+    appended_count = AppendedCount,
+    read_count = ReadCount,
+    dropped_count = DroppedCount,
+    error_count = ErrorCount
+}) ->
+    #{
+        dir => Dir,
+        write_interval => WriteInterval,
+        read_interval => ReadInterval,
+        max_writes => MaxWrites,
+        write_count => WriteCount,
+        appended_count => AppendedCount,
+        read_count => ReadCount,
+        dropped_count => DroppedCount,
+        error_count => ErrorCount,
+        pending_count => AppendedCount - ReadCount
+    }.
+
+-spec normalize_options([option()] | map()) -> map().
+normalize_options(Options) when is_map(Options) ->
+    Options;
+normalize_options(Options) when is_list(Options) ->
+    maps:from_list(Options).
+
+-spec positive_option(atom(), map(), pos_integer()) -> pos_integer().
+positive_option(Key, Options, Default) ->
+    case maps:get(Key, Options, Default) of
+        Value when is_integer(Value), Value > 0 ->
+            Value;
+        Value ->
+            error({invalid_option, Key, Value})
+    end.
+
+-spec max_writes_option(map()) -> pos_integer() | infinity.
+max_writes_option(Options) ->
+    case maps:get(max_writes, Options, infinity) of
+        infinity ->
+            infinity;
+        Value when is_integer(Value), Value > 0 ->
+            Value;
+        Value ->
+            error({invalid_option, max_writes, Value})
+    end.
+
+-spec decode_packet(binary()) -> term().
+decode_packet(Packet) ->
+    try binary_to_term(Packet, [safe]) of
+        Term ->
+            Term
+    catch
+        error:_ ->
+            Packet
+    end.