From 46c6e05436b6c0c091ce052b16cf1152ff34f04a Mon Sep 17 00:00:00 2001 From: Karl Nilsson Date: Tue, 31 Mar 2026 17:23:00 +0100 Subject: [PATCH] Add optional append_batch/2 callback for opaque batch accumulation Introduce a new optional behaviour callback `append_batch/2` that gives the implementation module full control over how operations are accumulated into a batch. When exported, each incoming op is passed through the callback instead of being prepended to a list, turning the batch into an opaque term. The callback receives `init_batch` as the accumulator for the first operation in a new batch, and the value is reset to `init_batch` after each batch completes. Returning `{finish_batch, OpaqueBatch}` from the callback completes the batch immediately, even if the batch size limit has not been reached. This allows implementations to enforce domain-specific termination conditions (e.g. total byte size, memory limit, entry count). Implementation details: - The callback fun is cached in the config record at init time to avoid module dispatch overhead on the hot path - On the normal (non-finish) path, `append_msg` returns a bare `#state{}` with zero extra allocation; the `{finish, State}` wrapper is only allocated on early batch termination - When `cast_batch/2` is used with `append_batch/2`, each operation is passed through the callback individually, allowing any operation to trigger early completion - The `reversed_batch` option has no effect when `append_batch/2` is exported, as batch ordering is entirely the caller's responsibility - The batch type is now `[op()] | init_batch | term()` to accurately reflect all three possible states --- README.md | 56 +++++++- src/gen_batch_server.erl | 233 ++++++++++++++++++++++++-------- test/gen_batch_server_SUITE.erl | 171 ++++++++++++++++++++++- 3 files changed, 401 insertions(+), 59 deletions(-) diff --git a/README.md b/README.md index 1573620..e2b49ac 100644 --- a/README.md +++ b/README.md @@ -223,11 +223,12 @@ unavailable (e.g. when a Ra member runs out of disk space). #### Module:handle_batch(Batch, State) -> Result. Types: - Batch = [Op] + Batch = [Op] | OpaqueBatch UserOp = term(), Op = {cast, UserOp} | {call, from(), UserOp} | {info, UserOp}. + OpaqueBatch = term() Result = {ok, State} | {ok, State, {continue, Continue}} | {ok, Actions, State} | @@ -242,6 +243,59 @@ unavailable (e.g. when a Ra member runs out of disk space). Called whenever a new batch is ready for processing. The implementation can optionally return a list of reply actions used to reply to `call` operations. +When the module exports `append_batch/2`, `Batch` is the opaque term built by +that callback instead of a list of `Op` tuples. + +#### Module:append_batch(Op, Batch) -> Result + + Types: + Op = {cast, UserOp} | + {call, from(), UserOp} | + {info, UserOp}. + UserOp = term() + Batch = OpaqueBatch | init_batch + OpaqueBatch = term() + Result = OpaqueBatch | {finish_batch, OpaqueBatch} + +Optional. When exported, this callback is invoked for every incoming operation +instead of the default behaviour of prepending it to a list. This turns the +batch into an opaque term fully controlled by the implementation. + +`Batch` is the atom `init_batch` for the first operation in a new batch. +Subsequent operations receive the value returned by the previous invocation. + +After each batch is processed by `handle_batch/2`, the batch state is reset to +`init_batch` for the next batch to begin. + +Returning `{finish_batch, OpaqueBatch}` completes the batch immediately (even if +the current batch size limit has not been reached) and passes `OpaqueBatch` to +`handle_batch/2`. This is useful for domain-specific termination conditions +(e.g. total byte size, entry count, or logical boundaries). + +When `cast_batch/2` is used with `append_batch/2`, each operation in the batch +is passed through the callback individually in the order received. Any operation +can trigger early batch completion by returning `{finish_batch, _}`. + +When this callback is exported the `reversed_batch` option has no effect. + +Example: Building a size-limited batch accumulator: + + append_batch(Op, init_batch) -> + #{size => op_size(Op), ops => [Op]}; + append_batch(Op, Acc = #{size := Size, ops := Ops}) -> + OpSize = op_size(Op), + NewSize = Size + OpSize, + case NewSize > 1_000_000 of + true -> + {finish_batch, Acc}; % Batch is full, finish now + false -> + Acc#{size => NewSize, ops => [Op | Ops]} + end. + + op_size({cast, Data}) -> byte_size(term_to_binary(Data)); + op_size({call, _, Data}) -> byte_size(term_to_binary(Data)); + op_size({info, Data}) -> byte_size(term_to_binary(Data)). + #### Module:handle_continue(Continue, State) -> Result Types: diff --git a/src/gen_batch_server.erl b/src/gen_batch_server.erl index ee078b6..431625c 100644 --- a/src/gen_batch_server.erl +++ b/src/gen_batch_server.erl @@ -56,11 +56,13 @@ -type response_timeout() :: timeout() | {abs, integer()}. +-type batch() :: [op()] | {batch, term()}. + -type op() :: {cast, UserOp :: term()} | {call, from(), UserOp :: term()} | {info, UserOp :: term()}. --record(config, {batch_size :: non_neg_integer(), +-record(config, { min_batch_size :: non_neg_integer(), max_batch_size :: non_neg_integer(), parent :: pid(), @@ -69,16 +71,19 @@ hibernate_after = infinity :: non_neg_integer() | infinity, reversed_batch = false :: boolean(), flush_mailbox_on_terminate = false :: false | {true, non_neg_integer()}, - batch_size_growth = exponential :: exponential | {aimd, pos_integer()}}). + batch_size_growth = exponential :: exponential | {aimd, pos_integer()}, + append_batch = undefined :: undefined | fun((op(), term()) -> term() | {finish_batch, term()})}). --record(state, {batch = [] :: [op()], +-record(state, {batch = [] :: batch(), batch_count = 0 :: non_neg_integer(), + batch_size :: non_neg_integer(), config = #config{} :: #config{}, state :: term(), needs_gc = false :: boolean(), - debug :: list()}). + debug :: list(), + leftover_ops = [] :: [op()]}). --export_type([from/0, reply_tag/0, op/0, +-export_type([from/0, reply_tag/0, op/0, batch/0, action/0, server_ref/0, request_id/0, request_id_collection/0]). @@ -97,7 +102,7 @@ {error, Reason :: term()} when State :: term(). --callback handle_batch([op()], State) -> +-callback handle_batch(batch(), State) -> {ok, State} | {ok, State, {continue, term()}} | {ok, [action()], State} | @@ -115,11 +120,15 @@ -callback format_status(State :: term()) -> term(). +-callback append_batch(op(), OpaqueBatch :: term() | init_batch) -> + OpaqueBatch :: term() | {finish_batch, OpaqueBatch :: term()}. + %% TODO: code_change -optional_callbacks([handle_continue/2, format_status/1, - terminate/2]). + terminate/2, + append_batch/2]). %%% @@ -171,26 +180,41 @@ init_it(Starter, Parent, Name0, Mod, {GBOpts, Args}, Options) -> FlushMailbox = proplists:get_value(flush_mailbox_on_terminate, GBOpts, false), BatchSizeGrowth = proplists:get_value(batch_size_growth, GBOpts, exponential), + %% call module_info/0 here to ensure that the module is loaded as else + %% it may return false when calling erlang:function_exported/3 even + %% if the callback is implemented. + _ = Mod:module_info(), + AppendBatch = case erlang:function_exported(Mod, append_batch, 2) of + true -> + fun Mod:append_batch/2; + false -> + undefined + end, + BatchInit = init_batch(AppendBatch), Conf = #config{module = Mod, parent = Parent, name = Name, - batch_size = MinBatchSize, min_batch_size = MinBatchSize, max_batch_size = MaxBatchSize, hibernate_after = HibernateAfter, reversed_batch = ReverseBatch, flush_mailbox_on_terminate = FlushMailbox, - batch_size_growth = BatchSizeGrowth}, + batch_size_growth = BatchSizeGrowth, + append_batch = AppendBatch}, case init_it(Mod, Args) of {ok, {ok, Inner0}} -> proc_lib:init_ack(Starter, {ok, self()}), State = #state{config = Conf, + batch = BatchInit, + batch_size = MinBatchSize, state = Inner0, debug = Debug}, loop_wait(State, Parent); {ok, {ok, Inner0, {continue, Continue}}} -> proc_lib:init_ack(Starter, {ok, self()}), State0 = #state{config = Conf, + batch = BatchInit, + batch_size = MinBatchSize, state = Inner0, debug = Debug}, State = handle_continue(Continue, State0), @@ -432,54 +456,142 @@ loop_wait(#state{config = #config{hibernate_after = Hib}} = State00, Parent) -> append_msg({'$gen_cast', Msg}, #state{batch = Batch, - batch_count = BatchCount} = State0) -> + batch_count = BatchCount, + config = #config{append_batch = undefined}} = State0) -> State0#state{batch = [{cast, Msg} | Batch], batch_count = BatchCount + 1}; +append_msg({'$gen_cast', Msg}, + #state{batch = Batch, + batch_count = BatchCount, + config = #config{append_batch = Fun}} = State0) -> + append_op(Fun, {cast, Msg}, Batch, BatchCount, State0); append_msg({'$gen_cast_batch', Msgs, Count}, #state{batch = Batch, - batch_count = BatchCount} = State0) -> + batch_count = BatchCount, + config = #config{append_batch = undefined}} = State0) -> State0#state{batch = Msgs ++ Batch, batch_count = BatchCount + Count}; +append_msg({'$gen_cast_batch', Msgs, _Count}, + #state{batch = Batch, + batch_count = BatchCount, + config = #config{append_batch = Fun}} = State0) -> + case fold_append(lists:reverse(Msgs), Fun, Batch, BatchCount, State0) of + {finish, Rest, State} -> + {finish, State#state{leftover_ops = Rest}}; + #state{} = State -> + State + end; append_msg({'$gen_call', From, Msg}, #state{batch = Batch, - batch_count = BatchCount} = State0) -> + batch_count = BatchCount, + config = #config{append_batch = undefined}} = State0) -> State0#state{batch = [{call, From, Msg} | Batch], batch_count = BatchCount + 1}; +append_msg({'$gen_call', From, Msg}, + #state{batch = Batch, + batch_count = BatchCount, + config = #config{append_batch = Fun}} = State0) -> + append_op(Fun, {call, From, Msg}, Batch, BatchCount, State0); append_msg(Msg, #state{batch = Batch, - batch_count = BatchCount} = State0) -> + batch_count = BatchCount, + config = #config{append_batch = undefined}} = State0) -> State0#state{batch = [{info, Msg} | Batch], - batch_count = BatchCount + 1}. + batch_count = BatchCount + 1}; +append_msg(Msg, #state{batch = Batch, + batch_count = BatchCount, + config = #config{append_batch = Fun}} = State0) -> + append_op(Fun, {info, Msg}, Batch, BatchCount, State0). + +append_op(Fun, Op, {batch, Batch}, BatchCount, State0) -> + case Fun(Op, Batch) of + {finish_batch, NewBatch} -> + {finish, State0#state{batch = {batch, NewBatch}, + batch_count = BatchCount + 1}}; + NewBatch -> + State0#state{batch = {batch, NewBatch}, + batch_count = BatchCount + 1} + end. + +fold_append([], _Fun, Batch, Count, State0) -> + State0#state{batch = Batch, batch_count = Count}; +fold_append([Op | Rest], Fun, {batch, Batch}, Count, State0) -> + case Fun(Op, Batch) of + {finish_batch, NewBatch} -> + {finish, Rest, State0#state{batch = {batch, NewBatch}, + batch_count = Count + 1}}; + NewBatch -> + fold_append(Rest, Fun, {batch, NewBatch}, Count + 1, State0) + end. enter_loop_batched(Msg, Parent, #state{debug = []} = State0) -> - loop_batched(append_msg(Msg, State0), Parent); + case append_msg(Msg, State0) of + {finish, State} -> + loop_batched_finish(State, Parent); + #state{} = State -> + loop_batched(State, Parent) + end; enter_loop_batched(Msg, Parent, State0) -> - State = handle_debug_in(State0, Msg), - %% append to batch - loop_batched(append_msg(Msg, State), Parent). + State1 = handle_debug_in(State0, Msg), + case append_msg(Msg, State1) of + {finish, State} -> + loop_batched_finish(State, Parent); + #state{} = State -> + loop_batched(State, Parent) + end. + +next_batch_size(BatchSize, #config{max_batch_size = Max, + batch_size_growth = Growth}) -> + case Growth of + exponential -> + min(Max, BatchSize * 2); + {aimd, Step} -> + min(Max, BatchSize + Step) + end. -loop_batched(#state{config = #config{batch_size = BatchSize, - max_batch_size = Max, - batch_size_growth = Growth} = Config, +loop_batched_finish(#state{config = Config, + batch_size = BatchSize, + leftover_ops = LeftoverOps} = State0, + Parent) -> + State = complete_batch(State0), + NewBatchSize = next_batch_size(BatchSize, Config), + State1 = State#state{batch_size = NewBatchSize, leftover_ops = []}, + process_leftover_ops(LeftoverOps, Parent, State1). + +process_leftover_ops([], Parent, State) -> + loop_batched(State, Parent); +process_leftover_ops(LeftoverOps, Parent, State0) -> + %% LeftoverOps are already in the correct `{cast, Msg}` format. + %% We just need to wrap them in a `$gen_cast_batch` message. + %% Note: They are in chronological order, but `$gen_cast_batch` expects + %% them to be reversed (as cast_batch_msg/1 does). + ReversedOps = lists:reverse(LeftoverOps), + Msg = {'$gen_cast_batch', ReversedOps, length(ReversedOps)}, + case append_msg(Msg, State0) of + {finish, State} -> + State1 = complete_batch(State), + NewBatchSize = next_batch_size(State1#state.batch_size, State1#state.config), + State2 = State1#state{batch_size = NewBatchSize}, + NewLeftoverOps = State2#state.leftover_ops, + process_leftover_ops(NewLeftoverOps, Parent, State2#state{leftover_ops = []}); + #state{} = State -> + loop_batched(State, Parent) + end. + +loop_batched(#state{config = Config, + batch_size = BatchSize, batch_count = BatchCount} = State0, Parent) when BatchCount >= BatchSize -> % complete batch after seeing batch_size writes State = complete_batch(State0), - % grow batch size according to the configured strategy - NewBatchSize = case Growth of - exponential -> - min(Max, BatchSize * 2); - {aimd, Step} -> - min(Max, BatchSize + Step) - end, - loop_wait(State#state{config = Config#config{batch_size = NewBatchSize}}, - Parent); -loop_batched(#state{debug = Debug} = State0, Parent) -> + NewBatchSize = next_batch_size(BatchSize, Config), + loop_wait(State#state{batch_size = NewBatchSize}, Parent); +loop_batched(State0, Parent) -> receive Msg -> case Msg of {system, From, Request} -> sys:handle_system_msg(Request, From, Parent, - ?MODULE, Debug, State0); + ?MODULE, State0#state.debug, State0); {'EXIT', Parent, Reason} -> flush_mailbox(State0#state.config), terminate(Reason, State0), @@ -491,70 +603,76 @@ loop_batched(#state{debug = Debug} = State0, Parent) -> State = complete_batch(State0), Config = State#state.config, NewBatchSize = max(Config#config.min_batch_size, - Config#config.batch_size div 2), - loop_wait(State#state{config = - Config#config{batch_size = NewBatchSize}}, - Parent) + State#state.batch_size div 2), + loop_wait(State#state{batch_size = NewBatchSize}, Parent) end. terminate(Reason, #state{config = #config{module = Mod}, state = Inner}) -> catch Mod:terminate(Reason, Inner), ok. +init_batch(undefined) -> + []; +init_batch(_) -> + {batch, init_batch}. -complete_batch(#state{batch = []} = State) -> +complete_batch(#state{batch_count = 0} = State) -> State; complete_batch(#state{batch = Batch0, config = #config{module = Mod, - reversed_batch = ReverseBatch}, + reversed_batch = ReverseBatch, + append_batch = AppendBatch}, state = Inner0, debug = Debug0} = State0) -> - Batch = case ReverseBatch of - false -> - %% reversing restores the received order + Batch = case {AppendBatch, ReverseBatch} of + {undefined, false} -> lists:reverse(Batch0); - true -> - %% accepting the batch in reverse order means we can avoid - %% reversing the list - Batch0 + {undefined, true} -> + Batch0; + _ -> + {batch, Opaque} = Batch0, + Opaque end, + BatchReset = init_batch(AppendBatch), + try Mod:handle_batch(Batch, Inner0) of - Result -> handle_batch_result(Result, State0, Debug0) + Result -> + handle_batch_result(Result, State0, Debug0, BatchReset) catch throw:Result -> - handle_batch_result(Result, State0, Debug0); + handle_batch_result(Result, State0, Debug0, BatchReset); Class:Reason:Stacktrace -> flush_mailbox(State0#state.config), terminate(Reason, State0), erlang:raise(Class, Reason, safe_stacktrace(Stacktrace)) end. -handle_batch_result({ok, Inner}, State0, _Debug0) -> - State0#state{batch = [], +handle_batch_result({ok, Inner}, State0, _Debug0, BatchReset) -> + State0#state{batch = BatchReset, state = Inner, batch_count = 0}; -handle_batch_result({ok, Inner, {continue, Continue}}, State0, _Debug0) -> +handle_batch_result({ok, Inner, {continue, Continue}}, State0, _Debug0, BatchReset) -> handle_continue(Continue, - State0#state{batch = [], + State0#state{batch = BatchReset, state = Inner, batch_count = 0}); -handle_batch_result({ok, Actions, Inner}, State0, Debug0) when is_list(Actions) -> +handle_batch_result({ok, Actions, Inner}, State0, Debug0, BatchReset) when is_list(Actions) -> {ShouldGc, Debug} = handle_actions(Actions, Debug0), - State0#state{batch = [], + State0#state{batch = BatchReset, batch_count = 0, state = Inner, needs_gc = ShouldGc, debug = Debug}; -handle_batch_result({ok, Actions, Inner, {continue, Continue}}, State0, Debug0) when is_list(Actions) -> +handle_batch_result({ok, Actions, Inner, {continue, Continue}}, State0, Debug0, BatchReset) when is_list(Actions) -> {ShouldGc, Debug} = handle_actions(Actions, Debug0), handle_continue(Continue, - State0#state{batch = [], + State0#state{batch = BatchReset, batch_count = 0, state = Inner, needs_gc = ShouldGc, debug = Debug}); -handle_batch_result({stop, Reason}, State0, _Debug0) -> +handle_batch_result({stop, Reason}, State0, _Debug0, _BatchReset) -> flush_mailbox(State0#state.config), terminate(Reason, State0), exit(Reason). @@ -574,7 +692,8 @@ handle_actions(Actions, Debug0) -> handle_continue(Continue, #state{config = #config{module = Mod}, state = Inner0} = State0) -> try Mod:handle_continue(Continue, Inner0) of - Result -> handle_continue_result(Result, State0) + Result -> + handle_continue_result(Result, State0) catch throw:Result -> handle_continue_result(Result, State0); diff --git a/test/gen_batch_server_SUITE.erl b/test/gen_batch_server_SUITE.erl index 6cd66e5..cdff37f 100644 --- a/test/gen_batch_server_SUITE.erl +++ b/test/gen_batch_server_SUITE.erl @@ -54,7 +54,12 @@ all_tests() -> send_request_receive_response, send_request_wait_response, send_request_check_response, - send_request_collection + send_request_collection, + append_batch_basic, + append_batch_finish, + append_batch_reversed_batch_ignored, + append_batch_cast_batch, + append_batch_cast_batch_with_leftover ]. groups() -> @@ -797,6 +802,136 @@ send_request_collection(Config) -> ?assert(meck:validate(Mod)), ok. +append_batch_basic(Config) -> + Mod = ?config(mod, Config), + Self = self(), + meck:new(Mod, [non_strict]), + meck:expect(Mod, init, fun(_) -> {ok, #{}} end), + meck:expect(Mod, append_batch, + fun({cast, Val}, init_batch) -> + [Val]; + ({cast, Val}, Acc) -> + [Val | Acc]; + ({info, Val}, init_batch) -> + [Val]; + ({info, Val}, Acc) -> + [Val | Acc]; + ({call, _From, Val}, init_batch) -> + [Val]; + ({call, _From, Val}, Acc) -> + [Val | Acc] + end), + meck:expect(Mod, handle_batch, + fun(Batch, State) -> + Self ! {batch, Batch}, + {ok, State} + end), + {ok, Pid} = gen_batch_server:start_link(Mod, []), + ok = gen_batch_server:cast(Pid, hello), + receive + {batch, Batch} -> + ?assertEqual([hello], Batch) + after 2000 -> exit(timeout) + end, + ?assert(meck:validate(Mod)), + ok. + +append_batch_finish(Config) -> + Mod = ?config(mod, Config), + Self = self(), + meck:new(Mod, [non_strict]), + meck:expect(Mod, init, fun(_) -> {ok, 0} end), + meck:expect(Mod, append_batch, + fun({cast, Val}, init_batch) -> + {finish_batch, [Val]}; + ({cast, Val}, Acc) -> + {finish_batch, [Val | Acc]} + end), + meck:expect(Mod, handle_batch, + fun(Batch, Count) -> + Self ! {batch, Batch}, + {ok, Count + 1} + end), + Opts = [{max_batch_size, 8192}], + {ok, Pid} = gen_batch_server:start_link(undefined, Mod, [], Opts), + ok = gen_batch_server:cast(Pid, a), + ok = gen_batch_server:cast(Pid, b), + ok = gen_batch_server:cast(Pid, c), + Batches = collect_batches(3, []), + ?assert(lists:all(fun(B) -> length(B) =:= 1 end, Batches)), + ?assert(meck:validate(Mod)), + ok. + +append_batch_reversed_batch_ignored(Config) -> + Mod = ?config(mod, Config), + Self = self(), + meck:new(Mod, [non_strict]), + meck:expect(Mod, init, fun(_) -> {ok, #{}} end), + meck:expect(Mod, append_batch, + fun({cast, Val}, init_batch) -> + [Val]; + ({cast, Val}, Acc) -> + Acc ++ [Val] + end), + meck:expect(Mod, handle_batch, + fun(Batch, State) -> + Self ! {batch, Batch}, + {ok, State} + end), + Opts = [{reversed_batch, true}], + {ok, Pid} = gen_batch_server:start_link(undefined, Mod, [], Opts), + gen_batch_server:cast(Pid, block), + timer:sleep(50), + gen_batch_server:cast(Pid, 1), + gen_batch_server:cast(Pid, 2), + gen_batch_server:cast(Pid, 3), + %% discard the first batch (the block message) + receive {batch, _} -> ok after 2000 -> exit(timeout) end, + receive + {batch, Batch} -> + ?assertEqual([1, 2, 3], Batch) + after 2000 -> exit(timeout) + end, + ?assert(meck:validate(Mod)), + ok. + +append_batch_cast_batch(Config) -> + Mod = ?config(mod, Config), + Self = self(), + meck:new(Mod, [non_strict]), + meck:expect(Mod, init, fun(_) -> {ok, #{}} end), + meck:expect(Mod, append_batch, + fun({cast, Val}, init_batch) -> + [Val]; + ({cast, Val}, Acc) -> + Acc ++ [Val] + end), + meck:expect(Mod, handle_batch, + fun(Batch, State) -> + Self ! {batch, Batch}, + {ok, State} + end), + {ok, Pid} = gen_batch_server:start_link(Mod, []), + gen_batch_server:cast_batch(Pid, [a, b, c]), + receive + {batch, Batch} -> + %% cast_batch_msg reverses the ops via foldl, then append_msg + %% reverses them back before fold_append walks them calling + %% append_batch in chronological order. + ?assertEqual([a, b, c], Batch) + after 2000 -> exit(timeout) + end, + ?assert(meck:validate(Mod)), + ok. + +collect_batches(0, Acc) -> + lists:reverse(Acc); +collect_batches(N, Acc) -> + receive + {batch, Batch} -> collect_batches(N - 1, [Batch | Acc]) + after 2000 -> exit(timeout) + end. + %% Utility wait_batch() -> wait_batch([]). @@ -810,3 +945,37 @@ wait_batch(Acc) -> after 5000 -> exit(timeout) end. +%% Test for leftover ops handling +%% This test verifies that when append_batch/2 returns {finish_batch, _} in the +%% middle of a cast_batch message, the remaining ops are preserved and processed +%% as a new batch rather than being dropped. +append_batch_cast_batch_with_leftover(Config) -> + Mod = ?config(mod, Config), + Self = self(), + meck:new(Mod, [non_strict]), + meck:expect(Mod, init, fun(_) -> {ok, #{}} end), + meck:expect(Mod, append_batch, + fun({cast, 3}, Acc) -> + {finish_batch, [3 | Acc]}; % Op 3 finishes batch, 4-5 become leftover + ({cast, Val}, init_batch) -> + [Val]; + ({cast, Val}, Acc) -> + [Val | Acc] + end), + meck:expect(Mod, handle_batch, + fun(Batch, State) -> + Self ! {batch, Batch}, + {ok, State} + end), + Opts = [{min_batch_size, 2}], + {ok, Pid} = gen_batch_server:start_link(undefined, Mod, [], Opts), + gen_batch_server:cast_batch(Pid, [1, 2, 3, 4, 5]), + %% Expect two batches: + %% 1. [3, 2, 1] - ops 1-3 (finished early by op 3) + %% 2. [5, 4] - ops 4-5 accumulate normally + Batches = collect_batches(2, []), + ?assertEqual(2, length(Batches)), + ?assertEqual([3, 2, 1], lists:nth(1, Batches)), + ?assertEqual([5, 4], lists:nth(2, Batches)), + ?assert(meck:validate(Mod)), + ok.