diff --git a/.github/workflows/erlang.yml b/.github/workflows/erlang.yml index 04195c69..8a6763eb 100644 --- a/.github/workflows/erlang.yml +++ b/.github/workflows/erlang.yml @@ -13,6 +13,13 @@ jobs: matrix: otp_version: [26, 27, 28] os: [ubuntu-latest, windows-latest] + # Pin ImageOS per OS: windows-latest reports win25-vs2026, which the + # pinned setup-beam can't map. Drop once erlef/setup-beam#461 ships. + include: + - os: ubuntu-latest + image_os: ubuntu24 + - os: windows-latest + image_os: win25 steps: - name: Checkout uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1 @@ -20,6 +27,8 @@ jobs: # Pin setup-beam to a version that works with Windows (not released at # the time of commit). uses: erlef/setup-beam@566deebc640988a494af16ecdf6f820fe0d3fea4 + env: + ImageOS: ${{ matrix.image_os }} with: otp-version: ${{ matrix.otp_version }} rebar3-version: '3.25.1' diff --git a/.gitignore b/.gitignore index 49d2e3bd..0cb934ea 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ erl_crash.dump +rebar3.crashdump .sw? .*.sw? *.beam @@ -14,6 +15,7 @@ callgrind* ct.coverdata test/ct.cover.spec _build +_checkouts ra.d *.plt diff --git a/Makefile b/Makefile index 067a3487..34c9dffc 100644 --- a/Makefile +++ b/Makefile @@ -11,7 +11,7 @@ ESCRIPT_EMU_ARGS = -noinput -setcookie ra_fifo_cli dep_gen_batch_server = hex 0.10.0 dep_aten = hex 0.6.0 dep_seshat = hex 1.0.1 -dep_shu = git https://github.com/rabbitmq/shu main +dep_shu = git https://github.com/lukebakken/rmq-shu shu-fixes DEPS = aten gen_batch_server seshat shu TEST_DEPS = proper meck inet_tcp_proxy diff --git a/rebar.config b/rebar.config index 0af9d0fd..614b7783 100644 --- a/rebar.config +++ b/rebar.config @@ -2,7 +2,7 @@ {gen_batch_server, "0.10.0"}, {aten, "0.6.0"}, {seshat, "1.0.1"}, - {shu, {git, "https://github.com/rabbitmq/shu.git", {branch, "main"}}} + {shu, {git, "https://github.com/lukebakken/rmq-shu.git", {branch, "shu-fixes"}}} ]}. {profiles, diff --git a/rebar.lock b/rebar.lock index 496deadd..308401a9 100644 --- a/rebar.lock +++ b/rebar.lock @@ -3,8 +3,8 @@ {<<"gen_batch_server">>,{pkg,<<"gen_batch_server">>,<<"0.10.0">>},0}, {<<"seshat">>,{pkg,<<"seshat">>,<<"1.0.1">>},0}, {<<"shu">>, - {git,"https://github.com/rabbitmq/shu.git", - {ref,"28adff94f0cc0e309b7be049cecc9d7107ef47c8"}}, + {git,"https://github.com/lukebakken/rmq-shu.git", + {ref,"35175aaa8962cbac9e53116002b5611bde3dee01"}}, 0}]}. [ {pkg_hash,[ diff --git a/rebar3.crashdump b/rebar3.crashdump deleted file mode 100644 index 2972948e..00000000 --- a/rebar3.crashdump +++ /dev/null @@ -1,22 +0,0 @@ -Error: terminated -[{io,format, - ["===> Verifying dependencies...~n",[]], - [{file,"io.erl"}, - {line,202}, - {error_info,#{cause => {io,terminated},module => erl_stdlib_errors}}]}, - {rebar_prv_install_deps,do,1, - [{file,"/home/runner/work/rebar3/rebar3/apps/rebar/src/rebar_prv_install_deps.erl"}, - {line,73}]}, - {rebar_core,do,2, - [{file,"/home/runner/work/rebar3/rebar3/apps/rebar/src/rebar_core.erl"}, - {line,155}]}, - {rebar3,run_aux,2, - [{file,"/home/runner/work/rebar3/rebar3/apps/rebar/src/rebar3.erl"}, - {line,205}]}, - {rebar3,main,1, - [{file,"/home/runner/work/rebar3/rebar3/apps/rebar/src/rebar3.erl"}, - {line,66}]}, - {init,start_it,1,[]}, - {init,start_em,1,[]}, - {init,do_boot,3,[]}] - diff --git a/src/ra_log_meta.erl b/src/ra_log_meta.erl index 0a932ed8..cfee683f 100644 --- a/src/ra_log_meta.erl +++ b/src/ra_log_meta.erl @@ -27,16 +27,26 @@ %% centralised meta data storage server for ra servers. -type key() :: current_term | voted_for | last_applied. --type value() :: non_neg_integer() | atom() | {atom() | binary(), atom()} | {binary(), atom()}. +-type value() :: non_neg_integer() | atom() | {atom() | binary(), atom()}. -define(TIMEOUT, 30000). -% -define(SYNC_INTERVAL, 5000). - --record(?MODULE, {shu :: shu:state(), - table_name :: atom(), - data_dir :: file:filename_all(), - compact_pid :: undefined | pid(), - compact_mref :: undefined | reference()}). +-define(SYNC_INTERVAL, 5000). +%% shu's default WAL size (see shu.hrl ?DEFAULT_WAL_SIZE). Kept in sync here +%% as the fallback when no override is configured. +-define(WAL_SIZE_DEFAULT, 16777216). +%% Maximum number of distinct ra servers whose metadata this store can hold. +%% Unlike DETS this is a fixed-size store, so the ceiling is pre-allocated; +%% override with the ra_log_meta_expected_count env for very large deployments. +-define(EXPECTED_COUNT_DEFAULT, 50000). +%% Size of shu's atom table, which backs the (small, slowly-growing) set of +%% distinct voted_for node names. Raised well above shu's 256 default so node +%% churn cannot exhaust it in practice; override with ra_log_meta_atom_slots. +-define(ATOM_SLOTS_DEFAULT, 4096). + +-record(?MODULE, {shu :: shu:state(), + table_name :: atom(), + compact_pid :: undefined | pid(), + compact_mref :: undefined | reference()}). -opaque state() :: #?MODULE{}. @@ -55,137 +65,202 @@ init(#{name := System, ok = ra_lib:make_dir(Dir), MetaShu = filename:join(Dir, "meta.shu"), MetaDets = filename:join(Dir, "meta.dets"), - Schema = schema(), - {ok, ShuState0} = shu:open(MetaShu, Schema), - %% Create ETS table as today + %% Create the ETS hot-cache table as today. _ = ets:new(TblName, [named_table, public, {read_concurrency, true}]), - %% Migration from DETS if present - {RecoveredCount, ShuState1} = case filelib:is_file(MetaDets) of - true -> - migrate_from_dets(MetaDets, ShuState0, - TblName); - false -> - {0, ShuState0} - end, + %% Open the shu store, migrating from a legacy DETS file only when the shu + %% store does not yet exist. An existing meta.shu is authoritative and may + %% hold metadata that has advanced past the DETS values, so a leftover + %% meta.dets (e.g. from a rename that failed after a previous successful + %% migration) is ignored rather than re-migrated over the newer data. The + %% migration itself is crash-atomic (see migrate_from_dets/3): meta.shu only + %% appears once fully populated, so an interrupted first migration re-runs + %% from the intact meta.dets on the next start. + {RecoveredCount, ShuState1} = + case filelib:is_file(MetaShu) of + true -> + {0, open_shu(MetaShu, Schema)}; + false -> + case filelib:is_file(MetaDets) of + true -> + migrate_from_dets(MetaShu, MetaDets, Schema); + false -> + {0, open_shu(MetaShu, Schema)} + end + end, - %% Populate ETS from shu + %% Populate the ETS cache from shu. ok = populate_ets_from_shu(TblName, ShuState1), ETSCount = ets:info(TblName, size), - ?INFO("ra: meta data store initialised for system ~ts. ~b record(s) " - "converted from DETS, ~b total records", - [System, RecoveredCount, - % case RecoveredCount of - % 0 -> "shu"; - % _ -> "dets" - % end, - ETSCount]), + case RecoveredCount of + 0 -> + ?INFO("ra: meta data store initialised for system ~ts. " + "~b record(s) recovered", [System, ETSCount]); + _ -> + ?INFO("ra: meta data store initialised for system ~ts. " + "~b record(s) migrated from DETS, ~b record(s) total", + [System, RecoveredCount, ETSCount]) + end, + + ok = schedule_sync(), {ok, #?MODULE{shu = ShuState1, - table_name = TblName, - data_dir = Dir}}. + table_name = TblName}}. handle_batch(Commands, #?MODULE{table_name = TblName} = State) -> - DoInsert = - fun (Id, Key, Value, Inserts0) -> - case Inserts0 of - #{Id := Data} -> - Inserts0#{Id => update_key(Key, Value, Data)}; - _ -> - case ets:lookup(TblName, Id) of - [Data] -> - Inserts0#{Id => update_key(Key, Value, Data)}; - [] -> - Data = {Id, undefined, undefined, undefined}, - Inserts0#{Id => update_key(Key, Value, Data)} - end - end - end, - {Inserts, Replies, FinalState} = - lists:foldl( - fun ({cast, {store, Id, Key, Value}}, - {Inserts0, Replies0, State0}) -> - {DoInsert(Id, Key, Value, Inserts0), Replies0, State0}; - ({call, From, {store, Id, Key, Value}}, - {Inserts0, Replies0, State0}) -> - {DoInsert(Id, Key, Value, Inserts0), - [{reply, From, ok} | Replies0], State0}; - ({cast, {delete, Id}}, - {Inserts0, Replies0, State0}) -> - {handle_delete(TblName, Id, Inserts0), Replies0, State0}; - ({call, From, {delete, Id}}, - {Inserts0, Replies0, State0}) -> - {handle_delete(TblName, Id, Inserts0), - [{reply, From, ok} | Replies0], State0}; - ({call, From, ping}, - {Inserts0, Replies0, State0}) -> - {Inserts0, [{reply, From, ok} | Replies0], State0}; - ({info, {'DOWN', MRef, process, _Pid, {compact_result, Result}}}, - {Inserts0, Replies0, State0}) when State0#?MODULE.compact_mref == MRef -> - case shu:finish_compact(Result, State0#?MODULE.shu) of - {ok, S1} -> - {Inserts0, Replies0, State0#?MODULE{shu = S1, compact_pid = undefined, - compact_mref = undefined}}; - {error, Reason} -> - ?ERROR("ra_log_meta: compaction finish failed: ~p", [Reason]), - exit({compaction_failed, Reason}) - end; - ({info, {'DOWN', _MRef, process, Pid, Reason}}, - {_Inserts0, _Replies0, State0}) when State0#?MODULE.compact_pid == Pid -> - ?ERROR("ra_log_meta: compaction worker ~p crashed: ~p", [Pid, Reason]), - exit({compaction_worker_crashed, Reason}); - ({info, Info}, {Inserts0, Replies0, State0}) -> - ?ERROR("ra_log_meta: unexpected info message: ~p", [Info]), - {Inserts0, Replies0, State0}; - (Unhandled, Acc) -> - ?DEBUG("ra: meta data unhandled ~p", [Unhandled]), - Acc - end, {#{}, [], State}, Commands), + {Inserts, Changed, Replies, DoSync, State1} = + lists:foldl(fun handle_command/2, + {#{}, #{}, [], false, State}, Commands), + %% Apply the merged rows to the ETS cache. Objects = maps:values(Inserts), true = ets:insert(TblName, Objects), - %% Translate to shu write_batch format - WriteOps = [to_shu_write_op(Obj) || Obj <- Objects], + %% Build shu write ops containing only the fields that actually changed + %% in this batch. Fields cleared to 'undefined' are written explicitly so + %% the change is persisted - in particular the voted_for reset on a term + %% change (see rabbitmq/ra#111). Writing only changed fields also keeps a + %% pure last_applied update a high-frequency WAL write (no fsync). + WriteOps = lists:filtermap( + fun (Id) -> + case maps:get(Id, Changed) of + [] -> + false; + Fields -> + {true, + to_shu_write_op(maps:get(Id, Inserts), + Fields)} + end + end, maps:keys(Changed)), + + State2 = write_ops(State1, WriteOps), + State3 = case DoSync of + true -> sync_shu_durable(State2); + false -> State2 + end, + %% Only the periodic timer message reschedules itself; a store_sync that + %% set DoSync must not create additional timers. + case lists:member({info, sync_meta}, Commands) of + true -> ok = schedule_sync(); + false -> ok + end, + {ok, Replies, State3}. + +handle_command({cast, {store, Id, Key, Value}}, Acc) -> + apply_store(Id, Key, Value, Acc); +handle_command({call, From, {store, Id, Key, Value}}, Acc0) -> + %% store_sync must be durable on return, so force a sync after the batch + %% (a low-frequency field is synced by shu already; a high-frequency field + %% such as last_applied would otherwise only reach the WAL) + {I, C, R, _DoSync, S} = apply_store(Id, Key, Value, Acc0), + {I, C, [{reply, From, ok} | R], true, S}; +handle_command({cast, {delete, Id}}, Acc) -> + apply_delete(Id, Acc); +handle_command({call, From, {delete, Id}}, Acc0) -> + %% delete_sync must be durable on return: force a batch sync so any + %% high-frequency write co-batched with the delete is flushed too + {I, C, R, _DoSync, S} = apply_delete(Id, Acc0), + {I, C, [{reply, From, ok} | R], true, S}; +handle_command({call, From, ping}, Acc) -> + add_reply(From, Acc); +handle_command({info, sync_meta}, {I, C, R, _DoSync, S}) -> + {I, C, R, true, S}; +handle_command({info, {'DOWN', MRef, process, _Pid, Reason}}, + {I, C, R, DoSync, #?MODULE{compact_mref = MRef} = S}) -> + %% The compaction worker exited. Apply its result (finish on success, + %% abort on failure/crash) without ever crashing the meta store. + {I, C, R, DoSync, apply_compaction_result(Reason, S)}; +handle_command({info, {'EXIT', _Pid, _Reason}}, Acc) -> + %% The compaction worker is linked (see start_compact/1); with trap_exit set + %% its exit is delivered here as well as via the monitor DOWN above. The DOWN + %% clause handles the result, so ignore the EXIT. + Acc; +handle_command({info, Info}, Acc) -> + ?ERROR("ra_log_meta: unexpected info message: ~p", [Info]), + Acc; +handle_command(Unhandled, Acc) -> + ?DEBUG("ra: meta data unhandled ~p", [Unhandled]), + Acc. + +add_reply(From, {I, C, R, DoSync, S}) -> + {I, C, [{reply, From, ok} | R], DoSync, S}. + +apply_store(Id, Key, Value, + {Inserts, Changed, R, DoSync, + #?MODULE{table_name = TblName} = State}) -> + Old = case Inserts of + #{Id := D} -> + D; + _ -> + case ets:lookup(TblName, Id) of + [D] -> D; + [] -> {Id, undefined, undefined, undefined} + end + end, + New = update_key(Key, Value, Old), + Fields = add_changed_fields(Old, New, maps:get(Id, Changed, [])), + {Inserts#{Id => New}, Changed#{Id => Fields}, R, DoSync, State}. + +apply_delete(Id, {Inserts, Changed, R, DoSync, + #?MODULE{table_name = TblName, shu = S0} = State}) -> + _ = ets:delete(TblName, Id), + S1 = case shu:delete(S0, Id) of + {ok, S} -> + S; + {error, not_found} -> + %% never persisted (e.g. deleted in the same batch it was + %% created) - nothing to remove from shu + S0 + end, + {maps:remove(Id, Inserts), maps:remove(Id, Changed), R, DoSync, + State#?MODULE{shu = S1}}. + +%% Determine which shu fields changed between the old and new ETS rows and +%% add their names to the accumulator (deduplicated). +add_changed_fields(Old, New, Acc0) -> + Acc1 = case element(2, Old) =:= element(2, New) of + true -> Acc0; + false -> add_field(current_term, Acc0) + end, + Acc2 = case element(3, Old) =:= element(3, New) of + true -> Acc1; + false -> add_field(voted_for_node, + add_field(voted_for_name, Acc1)) + end, + case element(4, Old) =:= element(4, New) of + true -> Acc2; + false -> add_field(last_applied, Acc2) + end. - %% Write to shu - shu handles syncing based on schema frequency config - case shu:write_batch(FinalState#?MODULE.shu, WriteOps) of - {ok, S1} -> - {ok, Replies, FinalState#?MODULE{shu = S1}}; - {wal_full, S1} -> - %% WAL is full, kick off background compaction and retry - State1 = start_compact(FinalState#?MODULE{shu = S1}), - %% After setting compacting = true, retry the write - %% The new write will be buffered in memory until compaction completes - case shu:write_batch(State1#?MODULE.shu, WriteOps) of - {ok, S2} -> - {ok, Replies, State1#?MODULE{shu = S2}}; - {wal_full, _S2} -> - %% Still full after compacting=true - should not happen - ?ERROR("ra_log_meta: WAL still full after starting compaction for ~ts", [TblName]), - exit({wal_full_after_compaction, TblName}); - {error, Reason} = Err -> - ?ERROR("ra_log_meta: write_batch failed: ~p", [Reason]), - exit(Err) - end; - {error, Reason} = Err -> - ?ERROR("ra_log_meta: write_batch failed: ~p", [Reason]), - exit(Err) +add_field(Field, Acc) -> + case lists:member(Field, Acc) of + true -> Acc; + false -> [Field | Acc] end. -terminate(_, #?MODULE{shu = S0, compact_mref = MRef} = State) -> +terminate(_, #?MODULE{compact_mref = MRef} = State0) -> ?DEBUG("ra: meta data store is terminating", []), - %% If a compaction is in flight, wait for it to finish - S1 = case MRef of - undefined -> S0; - _ -> - await_compaction(State, 30_000) - end, - ok = shu:close(S1), - ok. + %% If a compaction is in flight, wait for it to finish so we can close + %% cleanly. The ra_log_meta child spec sets shutdown => 30_000 to give this + %% the same budget as await_compaction (see ra_log_sup). + State = case MRef of + undefined -> State0; + _ -> await_compaction(State0, 30_000) + end, + case shu:close(State#?MODULE.shu) of + ok -> + ok; + {error, Reason} -> + %% e.g. compaction_in_progress when a compaction could not be + %% finished in time. The WAL is intact on disk and will be + %% replayed on the next start, so this is recoverable. + ?WARN("ra: meta data store could not cleanly close shu: ~p", + [Reason]), + ok + end. format_status(State) -> State. @@ -209,7 +284,8 @@ delete(Name, UId) -> delete_sync(Name, UId) -> gen_batch_server:call(Name, {delete, UId}, ?TIMEOUT). -%% Wait for the metadata store to be ready (used in tests) +%% Wait for the metadata store to have processed all prior requests. Useful +%% in tests to ensure the process has finished (re)initialising. -spec await(atom()) -> ok. await(Name) -> gen_batch_server:call(Name, ping, ?TIMEOUT). @@ -240,10 +316,6 @@ maybe_fetch(MetaName, Id, Pos) -> undefined end. -handle_delete(TblName, Id, Inserts) -> - _ = ets:delete(TblName, Id), - maps:remove(Id, Inserts). - update_key(current_term, Value, Data) -> case element(2, Data) of %% current term matches the new value, nothing to do @@ -259,164 +331,465 @@ update_key(voted_for, Value, Data) -> update_key(last_applied, Value, Data) -> setelement(4, Data, Value). -%% Helper to convert ETS row {UId, CT, VF, LA} to shu write operations -to_shu_write_op({UId, CurrentTerm, VotedFor, LastApplied}) -> - FieldValues1 = case CurrentTerm of - undefined -> - []; - _ -> - [{current_term, CurrentTerm}] - end, - FieldValues2 = case VotedFor of - undefined -> - FieldValues1; - _ -> - {ServerName, Node} = decode_voted_for(VotedFor), - ServerNameBin = case ServerName of - undefined -> - undefined; - S when is_atom(S) -> - atom_to_binary(S, utf8); - B when is_binary(B) -> - B - end, - [{voted_for_name, ServerNameBin}, - {voted_for_node, Node} | FieldValues1] - end, - FieldValues3 = case LastApplied of - undefined -> - FieldValues2; - _ -> - [{last_applied, LastApplied} | FieldValues2] - end, - {UId, FieldValues3}. - -%% Decode voted_for from ETS representation to (Node, ServerName) tuple -%% If VotedFor is an atom (old format), convert to {undefined, Atom} -%% If VotedFor is a {Node, ServerName} tuple, return as-is -%% If VotedFor is undefined, return {undefined, undefined} +%% Convert an ETS row {UId, CT, VF, LA} to a shu write op containing only the +%% requested fields (using the current row values, which may be undefined). +to_shu_write_op({UId, CurrentTerm, VotedFor, LastApplied}, Fields) -> + %% Only decode voted_for when a voted_for field actually changed; the + %% dominant write is a pure last_applied update which must stay cheap. + {ServerNameBin, Node} = + case lists:member(voted_for_name, Fields) orelse + lists:member(voted_for_node, Fields) of + true -> split_voted_for(VotedFor); + false -> {undefined, undefined} + end, + FieldValues = + lists:foldl( + fun (current_term, Acc) -> + [{current_term, CurrentTerm} | Acc]; + (voted_for_name, Acc) -> + [{voted_for_name, ServerNameBin} | Acc]; + (voted_for_node, Acc) -> + [{voted_for_node, Node} | Acc]; + (last_applied, Acc) -> + [{last_applied, LastApplied} | Acc] + end, [], Fields), + {UId, FieldValues}. + +%% Split a voted_for value into its shu representation: +%% {ServerNameBinary | undefined, Node :: atom() | undefined}. +split_voted_for(VotedFor) -> + {ServerName, Node} = decode_voted_for(VotedFor), + ServerNameBin = case ServerName of + undefined -> + undefined; + S when is_atom(S) -> + atom_to_binary(S, utf8); + B when is_binary(B) -> + B + end, + {ServerNameBin, Node}. + +%% Decode a voted_for value into a {ServerName, Node} pair. +%% - a {ServerName, Node} tuple is returned as-is +%% - undefined maps to {undefined, undefined} +%% - a bare atom (legacy format) maps to {Atom, undefined} so it round-trips +%% back to the same bare atom via encode_voted_for/2 decode_voted_for({_, _} = ServerId) -> ServerId; decode_voted_for(undefined) -> {undefined, undefined}; decode_voted_for(Atom) when is_atom(Atom) -> - {undefined, Atom}. + {Atom, undefined}. + +%% Encode a {ServerName, Node} pair back into the ETS voted_for representation. +%% - both undefined -> undefined +%% - only ServerName set -> the bare atom (legacy format) +%% - both set -> a {ServerName, Node} tuple +encode_voted_for(undefined, undefined) -> undefined; +encode_voted_for(ServerName, undefined) -> ServerName; +encode_voted_for(ServerName, Node) -> {ServerName, Node}. -%% Schema definition for shu +%% Schema definition for shu. schema() -> + WalSize = application:get_env(ra, ra_log_meta_wal_size, ?WAL_SIZE_DEFAULT), + ExpectedCount = application:get_env(ra, ra_log_meta_expected_count, + ?EXPECTED_COUNT_DEFAULT), + AtomSlots = application:get_env(ra, ra_log_meta_atom_slots, + ?ATOM_SLOTS_DEFAULT), #{fields => [#{name => current_term, - type => {integer, 64}, - frequency => low}, - #{name => voted_for_name, - type => {binary, 255}, - frequency => low}, - #{name => voted_for_node, - type => {atom, 255}, - frequency => low}, - #{name => last_applied, - type => {integer, 64}, - frequency => high}], - key => {binary, 64}, - expected_count => 50000}. - -%% Populate ETS table from shu on startup + type => {integer, 64}, + frequency => low}, + #{name => voted_for_name, + type => {binary, 255}, + frequency => low}, + #{name => voted_for_node, + type => {atom, 255}, + frequency => low}, + #{name => last_applied, + type => {integer, 64}, + frequency => high}], + %% ra_uid() is an arbitrary binary; shu caps keys at 255 bytes, which + %% comfortably covers all generated and practical user-supplied uids. + key => {binary, 255}, + expected_count => ExpectedCount, + atom_table_slots => AtomSlots, + wal_size => WalSize}. + +%% Populate the ETS cache from shu on startup. populate_ets_from_shu(TblName, ShuState) -> - shu:fold( - fun(Key, Fields, _Acc) -> - CT = maps:get(current_term, Fields, undefined), - Node = maps:get(voted_for_node, Fields, undefined), - ServerNameBin = maps:get(voted_for_name, Fields, undefined), - ServerName = case ServerNameBin of - undefined -> - undefined; - B when is_binary(B) -> - binary_to_atom(B, utf8); - _ -> - ServerNameBin - end, - VF = encode_voted_for(ServerName, Node), - LA = maps:get(last_applied, Fields, undefined), - % ?DEBUG("ra_log_meta: recovered from shu - Key=~p, CT=~p, VF=~p, LA=~p", - % [Key, CT, VF, LA]), - ets:insert(TblName, {Key, CT, VF, LA}), - _Acc - end, - ok, - ShuState), + _ = shu:fold( + fun(Key, Fields, _Acc) -> + CT = maps:get(current_term, Fields, undefined), + Node = maps:get(voted_for_node, Fields, undefined), + ServerNameBin = maps:get(voted_for_name, Fields, undefined), + ServerName = case ServerNameBin of + undefined -> + undefined; + B when is_binary(B) -> + binary_to_atom(B, utf8) + end, + VF = encode_voted_for(ServerName, Node), + LA = maps:get(last_applied, Fields, undefined), + ets:insert(TblName, {Key, CT, VF, LA}), + ok + end, + ok, + ShuState), ok. -%% Encode voted_for back into ETS representation -%% If both fields are undefined, return undefined -%% If only ServerName is set, return it as an atom (legacy format) -%% If both are set, return {Node, ServerName} tuple -encode_voted_for(undefined, undefined) -> undefined; -encode_voted_for(ServerName, undefined) -> ServerName; -encode_voted_for(ServerName, Node) -> {ServerName, Node}. +%% Open the shu store, converting an open error into a clear, loud failure. +%% There is no safe automatic recovery for Raft metadata (schema_mismatch, +%% unsupported_version, corrupt header), so we crash rather than silently start +%% with empty metadata; the file is left untouched for operator intervention. +open_shu(MetaShu, Schema) -> + case shu:open(MetaShu, Schema) of + {ok, S} -> + S; + {error, Reason} -> + ?ERROR("ra_log_meta: cannot open shu store ~ts: ~p", + [MetaShu, Reason]), + error({shu_open_failed, MetaShu, Reason}) + end. -%% Migrate from DETS to shu -migrate_from_dets(MetaDets, ShuState0, _TblName) -> - {ok, DetsTable} = dets:open_file(ra_log_meta_migration, [{file, MetaDets}]), - try - Count = dets:info(DetsTable, size), - ?INFO("ra_log_meta: migrating ~b records from DETS", [Count]), - - %% Collect all DETS rows and convert to shu write operations - Ops = dets:foldl( - fun({UId, CurrentTerm, VotedFor, LastApplied}, Acc) -> - {ServerName, Node} = decode_voted_for(VotedFor), - ServerNameBin = case ServerName of - undefined -> - undefined; - S when is_atom(S) -> - atom_to_binary(S, utf8); - S -> - S - end, - WriteOp = {UId, [{current_term, CurrentTerm}, - {voted_for_name, ServerNameBin}, - {voted_for_node, Node}, - {last_applied, LastApplied}]}, - [WriteOp | Acc] - end, - [], - DetsTable), +%% Crash-atomically migrate a legacy DETS file into a new shu store. The data is +%% built in a temporary file which is only renamed to MetaShu once fully written +%% and synced, so meta.shu never exists in a partially-migrated state: an +%% interrupted migration leaves no meta.shu and the next start re-runs it from +%% the intact meta.dets. Only after meta.shu is published is meta.dets retired. +%% A corrupt or unexpectedly shaped DETS fails loudly (file preserved) rather +%% than silently discarding metadata. +migrate_from_dets(MetaShu, MetaDets, Schema) -> + TmpShu = MetaShu ++ ".migrating", + %% Discard any leftover temp from a previously interrupted attempt. A + %% failure here (other than "already absent") means we cannot guarantee a + %% clean temp, so fail loudly rather than migrating into stale/foreign + %% state; the source meta.dets is left intact for a retry. + case file:delete(TmpShu) of + ok -> ok; + {error, enoent} -> ok; + {error, DelReason} -> + ?ERROR("ra_log_meta: cannot remove stale migration temp ~ts: ~p", + [TmpShu, DelReason]), + error({migration_temp_cleanup_failed, TmpShu, DelReason}) + end, + TmpState = open_shu(TmpShu, Schema), + {Count, TmpState1} = migrate_dets_into(MetaDets, TmpState, TmpShu), + %% flush + close the temp store, then atomically publish it as meta.shu. + ok = shu:close(TmpState1), + ok = file:rename(TmpShu, MetaShu), + %% migration is now durable and published; retire the source DETS. + ok = rename_migrated(MetaDets), + {Count, open_shu(MetaShu, Schema)}. + +migrate_dets_into(MetaDets, TmpState, TmpShu) -> + case dets:open_file(ra_log_meta_migration, [{file, MetaDets}]) of + {ok, DetsTable} -> + try + Result = do_migrate_from_dets(DetsTable, TmpState), + _ = dets:close(DetsTable), + Result + catch + Class:Reason:Stack -> + ?ERROR("ra_log_meta: DETS migration failed; ~ts is " + "preserved for inspection: ~p:~p~n~p", + [MetaDets, Class, Reason, Stack]), + _ = dets:close(DetsTable), + _ = catch shu:close(TmpState), + _ = file:delete(TmpShu), + erlang:raise(Class, Reason, Stack) + end; + {error, Reason} -> + _ = catch shu:close(TmpState), + _ = file:delete(TmpShu), + ?ERROR("ra_log_meta: cannot open legacy DETS ~ts for migration; " + "it is preserved for inspection: ~p", [MetaDets, Reason]), + error({dets_open_failed, MetaDets, Reason}) + end. - ?DEBUG("ra_log_meta: migration write ops = ~p", [lists:reverse(Ops)]), +do_migrate_from_dets(DetsTable, ShuState0) -> + Count = dets:info(DetsTable, size), + ?INFO("ra_log_meta: migrating ~b record(s) from DETS", [Count]), + Ops = dets:foldl(fun dets_row_to_write_op/2, [], DetsTable), + case migrate_write(ShuState0, lists:reverse(Ops)) of + {ok, ShuState1} -> + {ok, ShuState2} = shu:sync(ShuState1), + {Count, ShuState2}; + {error, Reason} -> + error({migration_write_failed, Reason}) + end. - %% Write all to shu in a single batch and sync - {ok, ShuState1} = shu:write_batch(ShuState0, lists:reverse(Ops)), - {ok, ShuState2} = shu:sync(ShuState1), +%% Write all migration ops in a single batch (one fsync) in the common case. +%% If the configured WAL is too small to hold the batch's high-frequency +%% (last_applied) entries, fall back to incremental writes with compaction +%% between them, which cannot overflow the WAL. +migrate_write(ShuState, Ops) -> + case shu:write_batch(ShuState, Ops) of + {ok, S1} -> + {ok, S1}; + {wal_full, S1} -> + ?WARN("ra_log_meta: WAL too small for bulk migration; migrating " + "incrementally", []), + migrate_incremental(S1, Ops); + {error, _} = Err -> + Err + end. - ?INFO("ra_log_meta: migration completed, wrote to shu and synced", []), +migrate_incremental(ShuState, []) -> + {ok, ShuState}; +migrate_incremental(ShuState0, [{Key, FieldValues} | Rest]) -> + case shu:write(ShuState0, Key, FieldValues) of + {ok, S1} -> + migrate_incremental(S1, Rest); + {wal_full, S1} -> + %% flush the WAL into the record area, then retry this op into the + %% now-empty WAL + {Work, S2} = shu:prepare_compact(S1), + case shu:do_compact(Work) of + ok -> + {ok, S3} = shu:finish_compact(ok, S2), + migrate_incremental(S3, [{Key, FieldValues} | Rest]); + {error, _} = Err -> + Err + end; + {error, _} = Err -> + Err + end. - {Count, ShuState2} - after - _ = dets:close(DetsTable), - %% Rename DETS file to .migrated - _ = file:rename(MetaDets, MetaDets ++ ".migrated") +dets_row_to_write_op({UId, CurrentTerm, VotedFor, LastApplied}, Acc) -> + {ServerNameBin, Node} = split_voted_for(VotedFor), + Op = {UId, [{current_term, CurrentTerm}, + {voted_for_name, ServerNameBin}, + {voted_for_node, Node}, + {last_applied, LastApplied}]}, + [Op | Acc]. + +rename_migrated(MetaDets) -> + Migrated = MetaDets ++ ".migrated", + case file:rename(MetaDets, Migrated) of + ok -> + ?INFO("ra_log_meta: DETS migration complete, renamed ~ts to ~ts", + [MetaDets, Migrated]), + ok; + {error, Reason} -> + %% the data is already durably in shu; a failed rename just means + %% migration is retried (idempotently) on the next start + ?WARN("ra_log_meta: could not rename ~ts after migration: ~p", + [MetaDets, Reason]), + ok end. -%% Start async compaction +%% Start an asynchronous compaction. shu is flipped into 'compacting' mode +%% which buffers subsequent writes in memory until finish_compact/2 is called. +%% The worker is spawned with a link as well as a monitor: the link guarantees +%% the worker dies with this process (even on a brutal supervisor kill), so it +%% can never outlive the store and write meta.shu concurrently with the +%% restarted store. This process traps exits, so the worker's own exit arrives +%% as a harmless {'EXIT', _, _} info message (the monitor DOWN carries the +%% actual result). -dialyzer({nowarn_function, start_compact/1}). start_compact(#?MODULE{compact_pid = undefined, shu = S0} = State) -> {Work, S1} = shu:prepare_compact(S0), - {Pid, MRef} = spawn_monitor(fun () -> exit({compact_result, shu:do_compact(Work)}) end), + {Pid, MRef} = spawn_opt( + fun () -> exit({compact_result, shu:do_compact(Work)}) end, + [link, monitor]), State#?MODULE{shu = S1, compact_pid = Pid, compact_mref = MRef}; start_compact(#?MODULE{compact_pid = Pid} = State) when is_pid(Pid) -> %% already compacting State. -%% Wait for compaction to finish with timeout (used in terminate) -await_compaction(#?MODULE{compact_mref = MRef, shu = S0}, Timeout) -> +%% Apply the outcome of a compaction worker (its DOWN exit reason), returning a +%% non-compacting state. do_compact/1 never raises, so the worker normally +%% exits {compact_result, ok | {error, _}}; any other reason is an abnormal +%% crash. A failed or crashed compaction is aborted (WAL left intact, buffered +%% values retained in the WAL cache) rather than crashing the meta store. +apply_compaction_result({compact_result, ok}, #?MODULE{shu = S0} = State) -> + case shu:finish_compact(ok, S0) of + {ok, S1} -> + clear_compaction(State#?MODULE{shu = S1}); + {error, Reason} -> + ?ERROR("ra_log_meta: finish_compact failed: ~p; aborting", [Reason]), + {ok, S1} = shu:abort_compact(S0), + clear_compaction(State#?MODULE{shu = S1}) + end; +apply_compaction_result({compact_result, {error, Reason}}, + #?MODULE{shu = S0} = State) -> + ?ERROR("ra_log_meta: compaction failed: ~p; aborting", [Reason]), + {ok, S1} = shu:abort_compact(S0), + clear_compaction(State#?MODULE{shu = S1}); +apply_compaction_result(Reason, #?MODULE{shu = S0} = State) -> + ?ERROR("ra_log_meta: compaction worker crashed: ~p; aborting", [Reason]), + {ok, S1} = shu:abort_compact(S0), + clear_compaction(State#?MODULE{shu = S1}). + +clear_compaction(State) -> + State#?MODULE{compact_pid = undefined, compact_mref = undefined}. + +%% Block until the in-flight compaction worker exits, then apply its result. +%% On timeout the state is returned unchanged (still compacting); callers decide +%% how to handle that. +await_compaction(#?MODULE{compact_mref = MRef} = State, Timeout) -> receive - {'DOWN', MRef, process, _Pid, {compact_result, Result}} -> - case shu:finish_compact(Result, S0) of - {ok, S} -> S; - {error, Reason} -> - ?ERROR("ra_log_meta: compaction finish during shutdown failed: ~p", [Reason]), - S0 - end + {'DOWN', MRef, process, _Pid, Reason} -> + apply_compaction_result(Reason, State) after Timeout -> - ?ERROR("ra_log_meta: compaction worker did not finish during shutdown", []), - S0 + ?ERROR("ra_log_meta: compaction did not finish within ~bms", + [Timeout]), + State end. + +%% Write the batch to shu. A full WAL triggers a background compaction; while it +%% runs, writes are buffered in shu's bounded in-memory WAL. If that buffer also +%% fills before the compaction completes, block until it does (reclaiming the +%% WAL) and retry - degrading to synchronous only under sustained pressure. +%% Thrown shu errors are converted into a controlled exit that restarts the log +%% subtree and recovers from shu + WAL. +write_ops(State, []) -> + State; +write_ops(#?MODULE{shu = S0} = State, WriteOps) -> + try shu:write_batch(S0, WriteOps) of + {ok, S1} -> + State#?MODULE{shu = S1}; + {wal_full, S1} -> + State1 = State#?MODULE{shu = S1}, + case State1#?MODULE.compact_pid of + undefined -> + %% Start a background compaction and retry: a batch that + %% fits wal_size succeeds by buffering into the compacting + %% WAL (non-blocking). A batch that does not fit is handled + %% once a compaction is in flight (the clause below). + write_ops(start_compact(State1), WriteOps); + _ -> + %% A compaction is in flight and even its bounded in-memory + %% buffer is full. Wait for it to complete, fully reclaiming + %% the WAL, then retry once. If the batch STILL does not fit + %% it exceeds wal_size and can never be written whole, so + %% fall back to per-op writes. + State2 = ensure_compaction_finished(State1), + case shu:write_batch(State2#?MODULE.shu, WriteOps) of + {ok, S3} -> + State2#?MODULE{shu = S3}; + {wal_full, S3} -> + write_ops_incremental(State2#?MODULE{shu = S3}, + WriteOps); + {error, Reason} = Err -> + ?ERROR("ra_log_meta: write_batch failed: ~p", + [Reason]), + exit(Err) + end + end; + {error, Reason} = Err -> + ?ERROR("ra_log_meta: write_batch failed: ~p", [Reason]), + exit(Err) + catch + throw:Thrown -> + shu_write_throw(Thrown) + end. + +%% Fall back path for a batch too large to fit the WAL in one write_batch: +%% write each op individually, reclaiming the WAL with a synchronous compaction +%% whenever it fills. Guaranteed to make progress unless a single op cannot fit +%% an empty WAL, which only happens if wal_size is misconfigured below one entry. +write_ops_incremental(State, []) -> + State; +write_ops_incremental(#?MODULE{shu = S0} = State, [{Key, FieldValues} | Rest]) -> + try shu:write(S0, Key, FieldValues) of + {ok, S1} -> + write_ops_incremental(State#?MODULE{shu = S1}, Rest); + {wal_full, S1} -> + State1 = compact_now(State#?MODULE{shu = S1}), + case shu:write(State1#?MODULE.shu, Key, FieldValues) of + {ok, S2} -> + write_ops_incremental(State1#?MODULE{shu = S2}, Rest); + {wal_full, _} -> + ?ERROR("ra_log_meta: wal_size too small to hold a single " + "write op", []), + exit(wal_size_too_small); + {error, Reason} = Err -> + ?ERROR("ra_log_meta: write failed: ~p", [Reason]), + exit(Err) + end; + {error, Reason} = Err -> + ?ERROR("ra_log_meta: write failed: ~p", [Reason]), + exit(Err) + catch + throw:Thrown -> + shu_write_throw(Thrown) + end. + +%% Synchronously reclaim the WAL (used by the incremental fallback, where no +%% async compaction is in flight). +compact_now(#?MODULE{shu = S0} = State) -> + {Work, S1} = shu:prepare_compact(S0), + case shu:do_compact(Work) of + ok -> + {ok, S2} = shu:finish_compact(ok, S1), + State#?MODULE{shu = S2}; + {error, Reason} = Err -> + {ok, _} = shu:abort_compact(S1), + ?ERROR("ra_log_meta: inline compaction failed: ~p", [Reason]), + exit({compaction_failed, Err}) + end. + +-spec shu_write_throw(term()) -> no_return(). +shu_write_throw({error, atom_table_full} = ATErr) -> + ?ERROR("ra_log_meta: cannot persist metadata, shu atom table is full " + "(too many distinct node names): ~p", [ATErr]), + exit({shu_write_failed, ATErr}); +shu_write_throw({unknown_field, _} = UFErr) -> + ?ERROR("ra_log_meta: cannot persist metadata: ~p", [UFErr]), + exit({shu_write_failed, UFErr}); +shu_write_throw(Other) -> + %% unexpected thrown term; surface it as a controlled failure + ?ERROR("ra_log_meta: unexpected shu error: ~p", [Other]), + exit({shu_write_failed, Other}). + +%% Wait for the in-flight compaction to complete so the WAL is reclaimed. A +%% compaction that never completes leaves the store unable to make progress and +%% is fatal. +ensure_compaction_finished(State0) -> + State = await_compaction(State0, ?TIMEOUT), + case State#?MODULE.compact_pid of + undefined -> + State; + _ -> + ?ERROR("ra_log_meta: compaction did not complete; cannot reclaim " + "WAL", []), + exit(compaction_stuck) + end. + +%% Durable sync for store_sync/delete_sync: the caller requires the write to be +%% on disk when the call returns. If a compaction is in flight, a high-frequency +%% field written this batch is only buffered in memory and shu:sync cannot flush +%% it, so first finish the compaction (which durably replays the buffered +%% writes), then sync. When not compacting this is just a plain sync. +sync_shu_durable(#?MODULE{compact_pid = undefined} = State) -> + sync_shu(State); +sync_shu_durable(State) -> + sync_shu(ensure_compaction_finished(State)). + +%% Periodically fsync the shu store to bound loss of WAL-buffered +%% (high-frequency) fields such as last_applied, similar to the old DETS +%% auto_save interval. This is best-effort: during a compaction the buffered +%% high-frequency writes are flushed durably when the compaction completes, so a +%% compaction_in_progress result is ignored here. +sync_shu(#?MODULE{shu = S0} = State) -> + case shu:sync(S0) of + {ok, S1} -> + State#?MODULE{shu = S1}; + {error, compaction_in_progress} -> + %% pending writes are buffered and flushed when compaction ends + %% (sync_shu_durable finishes the compaction first, so this only + %% arises on the best-effort periodic path) + State; + {error, Reason} = Err -> + %% A genuine fsync failure (e.g. EIO) means the disk is failing; we + %% must not let a store_sync/delete_sync caller believe its write is + %% durable when it is not. Fail loudly (the log subtree restarts and + %% recovers from the WAL), matching the write path's behaviour. + ?ERROR("ra_log_meta: sync failed: ~p", [Reason]), + exit(Err) + end. + +schedule_sync() -> + _ = erlang:send_after(?SYNC_INTERVAL, self(), sync_meta), + ok. diff --git a/src/ra_log_sup.erl b/src/ra_log_sup.erl index 12ce9f05..2dacda7a 100644 --- a/src/ra_log_sup.erl +++ b/src/ra_log_sup.erl @@ -29,7 +29,11 @@ init([#{data_dir := DataDir, PreInit = #{id => ra_log_pre_init, start => {ra_log_pre_init, start_link, [System]}}, Meta = #{id => ra_log_meta, - start => {ra_log_meta, start_link, [Cfg]}}, + start => {ra_log_meta, start_link, [Cfg]}, + %% ra_log_meta:terminate/2 may wait up to 30s for an in-flight + %% compaction to finish before closing the shu store cleanly; give + %% it that budget instead of the 5s worker default. + shutdown => 30_000}, PoolSize = ra_log_sync:pool_size(), LogSyncWorkers = [#{id => {ra_log_sync, I}, start => {ra_log_sync, start_link, diff --git a/test/ra_log_meta_SUITE.erl b/test/ra_log_meta_SUITE.erl index 4745bb7f..dc168dec 100644 --- a/test/ra_log_meta_SUITE.erl +++ b/test/ra_log_meta_SUITE.erl @@ -25,7 +25,12 @@ all_tests() -> [ roundtrip, delete, - trigger_compaction + trigger_compaction, + migrate_from_dets, + migrate_failure_preserves_dets, + migrate_skipped_when_shu_exists, + oversized_batch_does_not_hang, + atom_table_full_is_controlled ]. groups() -> @@ -34,19 +39,60 @@ groups() -> ]. init_per_group(_, Config) -> - PrivDir = ?config(priv_dir, Config), - {ok, _} = ra:start_in(PrivDir), Config. end_per_group(_, Config) -> - application:stop(ra), Config. +%% Each test case runs against its own ra system in its own data directory so +%% that a restart (or crash) in one test cannot leave shared state that masks a +%% regression in another. init_per_testcase(TestCase, Config) -> - %% Convert test case name (atom) to binary for use as ra_uid - [{key, atom_to_binary(TestCase, utf8)} | Config]. + PrivDir = ?config(priv_dir, Config), + DataDir = filename:join(PrivDir, atom_to_list(TestCase)), + Key = atom_to_binary(TestCase, utf8), + case TestCase of + trigger_compaction -> + %% small WAL so the compaction path is reached with a modest + %% number of writes + ok = application:set_env(ra, ra_log_meta_wal_size, 64 * 1024); + oversized_batch_does_not_hang -> + %% tiny WAL so a coalesced batch of distinct uids exceeds it + ok = application:set_env(ra, ra_log_meta_wal_size, 128); + _ -> + ok = application:unset_env(ra, ra_log_meta_wal_size) + end, + case TestCase of + atom_table_full_is_controlled -> + %% shrink the atom table so a handful of distinct node names + %% exhausts it quickly + ok = application:set_env(ra, ra_log_meta_atom_slots, 8); + _ -> + ok = application:unset_env(ra, ra_log_meta_atom_slots) + end, + case TestCase of + migrate_from_dets -> + %% two voted_for forms: a legacy bare atom, and the {Name, Node} + %% tuple that production ra actually persisted + ok = write_dets(DataDir, + [{Key, 42, 'node1@host', 100}, + {<<"migrate_tuple_uid">>, 7, + {my_server, 'rabbit@host'}, 55}]), + {ok, _} = ra:start_in(DataDir); + migrate_failure_preserves_dets -> + %% a uid larger than shu's 255-byte key limit makes the migration + %% write fail; ra is started by the test itself + BigKey = binary:copy(<<"x">>, 300), + ok = write_dets(DataDir, {BigKey, 5, undefined, 10}); + _ -> + {ok, _} = ra:start_in(DataDir) + end, + [{key, Key}, {data_dir, DataDir} | Config]. end_per_testcase(_, Config) -> + catch application:stop(ra), + application:unset_env(ra, ra_log_meta_wal_size), + application:unset_env(ra, ra_log_meta_atom_slots), Config. roundtrip(Config) -> @@ -64,13 +110,18 @@ roundtrip(Config) -> {custard, cream} = ra_log_meta:fetch(ra_log_meta, Id, voted_for), %% lose and re-open 199 = ra_log_meta:fetch(ra_log_meta, Id, last_applied), - proc_lib:stop(whereis(ra_log_meta), shutdown, infinity), - timer:sleep(100), - % give it some time to restart and be ready - ok = ra_log_meta:await(ra_log_meta), + restart_meta(), 5 = ra_log_meta:fetch(ra_log_meta, Id, current_term), {custard, cream} = ra_log_meta:fetch(ra_log_meta, Id, voted_for), 199 = ra_log_meta:fetch(ra_log_meta, Id, last_applied), + %% advancing the term clears voted_for (rabbitmq/ra#111); the cleared + %% value must be persisted and survive a restart (regression guard) + ok = ra_log_meta:store_sync(ra_log_meta, Id, current_term, 6), + undefined = ra_log_meta:fetch(ra_log_meta, Id, voted_for), + restart_meta(), + 6 = ra_log_meta:fetch(ra_log_meta, Id, current_term), + undefined = ra_log_meta:fetch(ra_log_meta, Id, voted_for), + 199 = ra_log_meta:fetch(ra_log_meta, Id, last_applied), ok. delete(Config) -> @@ -79,58 +130,181 @@ delete(Config) -> Oth = <<"some_other_id">>, ok = ra_log_meta:store_sync(ra_log_meta, Oth, last_applied, 1), ok = ra_log_meta:delete(ra_log_meta, Oth), %% async - ok = ra_log_meta:delete_sync(ra_log_meta, Id), %% async - %% store some other id just to make sure the delete is processed + ok = ra_log_meta:delete_sync(ra_log_meta, Id), %% sync + undefined = ra_log_meta:fetch(ra_log_meta, Oth, last_applied), + undefined = ra_log_meta:fetch(ra_log_meta, Id, last_applied), + %% the delete must be persisted to shu: after a restart the records must + %% not resurrect from the shu store (regression guard) + restart_meta(), undefined = ra_log_meta:fetch(ra_log_meta, Oth, last_applied), undefined = ra_log_meta:fetch(ra_log_meta, Id, last_applied), ok. trigger_compaction(Config) -> Id = ?config(key, Config), - %% Write enough data to fill the WAL and trigger a background compaction. - %% Default wal_size is 16MB. To make it fast, we can overwrite a large - %% `voted_for` string (must be {Name, Node}, both binaries or atoms) repeatedly. - %% A voted_for value of {binary, atom} works well. - %% Note: the binary name max size in schema is 255. Let's make it 200 bytes. - LargeName = binary:copy(<<"A">>, 200), - LargeVotedFor = {LargeName, node()}, - - %% Since the entry size will be ~250 bytes, to fill 16MB we need ~65000 writes. - %% This might take 10-15 seconds in a test. That's fine. - [ok = ra_log_meta:store(ra_log_meta, Id, voted_for, LargeVotedFor) || _ <- lists:seq(1, 70000)], - ok = ra_log_meta:store_sync(ra_log_meta, Id, voted_for, {<<"Final">>, node()}), - - %% Verify we can read the final value and that the server is alive - {<<"Final">>, _} = ra_log_meta:fetch(ra_log_meta, Id, voted_for), + %% last_applied is the only high-frequency (WAL-backed) field. Each + %% store_sync is processed as its own batch, so a stream of increasing + %% values appends one WAL entry per call and grows the WAL until it fills + %% and triggers a background compaction. store within a single batch is + %% coalesced, so store_sync (one batch per call) is required to actually + %% fill the WAL. With the small WAL configured in init_per_testcase this + %% cycles over during the loop. + Pid = whereis(ra_log_meta), + ?assert(is_pid(Pid)), + N = 6000, + [ok = ra_log_meta:store_sync(ra_log_meta, Id, last_applied, I) + || I <- lists:seq(1, N)], + %% The process must have survived the wal_full / compaction cycles (a + %% crash would restart it under a new pid) and the final value must read + %% back correctly, including after a restart (recovered from the compacted + %% record area plus the tail of the WAL). + Pid = whereis(ra_log_meta), + N = ra_log_meta:fetch(ra_log_meta, Id, last_applied), + restart_meta(), + N = ra_log_meta:fetch(ra_log_meta, Id, last_applied), ok. migrate_from_dets(Config) -> Id = ?config(key, Config), - PrivDir = ?config(priv_dir, Config), - - %% First, stop ra so we can create a hand-made DETS file - application:stop(ra), - timer:sleep(200), - - %% Create a temporary DETS file with known data - MetaDetsPath = filename:join(PrivDir, "meta.dets"), - {ok, DetsTable} = dets:open_file(test_dets_migration, [{file, MetaDetsPath}]), - dets:insert(DetsTable, {Id, 42, 'node1@host', 100}), - dets:close(DetsTable), - - %% Restart ra - should migrate DETS to shu - {ok, _} = ra:start_in(PrivDir), - timer:sleep(500), - - %% Verify migrated data is accessible via ETS - %% Note: we only test the simple values that round-trip well + DataDir = ?config(data_dir, Config), + %% ra was started in init_per_testcase with a pre-populated meta.dets, so + %% the migration has already run. + ok = ra_log_meta:await(ra_log_meta), 42 = ra_log_meta:fetch(ra_log_meta, Id, current_term), 100 = ra_log_meta:fetch(ra_log_meta, Id, last_applied), + %% legacy bare-atom voted_for must round-trip back to the same bare atom 'node1@host' = ra_log_meta:fetch(ra_log_meta, Id, voted_for), - - %% Verify DETS file was renamed to .migrated - true = filelib:is_file(MetaDetsPath ++ ".migrated"), - false = filelib:is_file(MetaDetsPath), - + %% the {Name, Node} tuple form (what production ra persisted) must also + %% round-trip through migration + Id2 = <<"migrate_tuple_uid">>, + 7 = ra_log_meta:fetch(ra_log_meta, Id2, current_term), + 55 = ra_log_meta:fetch(ra_log_meta, Id2, last_applied), + {my_server, 'rabbit@host'} = ra_log_meta:fetch(ra_log_meta, Id2, voted_for), + %% the DETS file is renamed only after a successful migration + MetaDets = meta_dets_path(DataDir), + true = filelib:is_file(MetaDets ++ ".migrated"), + false = filelib:is_file(MetaDets), ok. +migrate_failure_preserves_dets(Config) -> + DataDir = ?config(data_dir, Config), + MetaDets = meta_dets_path(DataDir), + true = filelib:is_file(MetaDets), + %% starting ra runs the migration, which fails on the oversized key. The + %% failure must not rename (and thereby lose) the source DETS file. + _ = (catch ra:start_in(DataDir)), + timer:sleep(300), + true = filelib:is_file(MetaDets), + false = filelib:is_file(MetaDets ++ ".migrated"), + %% crash-atomicity: a failed migration must leave NO meta.shu (nor a + %% leftover .migrating temp), so the next start re-runs the migration from + %% the intact DETS rather than starting from empty/partial shu. + MetaShu = filename:join(meta_dir(DataDir), "meta.shu"), + false = filelib:is_file(MetaShu), + false = filelib:is_file(MetaShu ++ ".migrating"), + ok. + +atom_table_full_is_controlled(Config) -> + Id = ?config(key, Config), + Pid = whereis(ra_log_meta), + ?assert(is_pid(Pid)), + MRef = erlang:monitor(process, Pid), + %% Writing voted_for values with many distinct node atoms exhausts shu's + %% fixed 256-slot atom table. The resulting error must surface as a + %% controlled exit (so gen_batch_server terminates cleanly and the log + %% subtree restarts) rather than leaking as an uncaught throw. The writes + %% run in a separate process so the failing store_sync call does not take + %% down the test process. + _ = spawn(fun () -> + catch [ra_log_meta:store_sync( + ra_log_meta, Id, voted_for, + {<<"n">>, + list_to_atom("atomfull_" ++ + integer_to_list(I))}) + || I <- lists:seq(1, 400)] + end), + receive + {'DOWN', MRef, process, Pid, Reason} -> + ?assertMatch({shu_write_failed, {error, atom_table_full}}, Reason) + after 10000 -> + ct:fail(meta_process_did_not_exit) + end, + ok. + +migrate_skipped_when_shu_exists(Config) -> + Id = ?config(key, Config), + DataDir = ?config(data_dir, Config), + %% ra started in init_per_testcase with a fresh meta.shu (no DETS). Advance + %% the metadata so shu holds a value newer than any DETS would. + ok = ra_log_meta:store_sync(ra_log_meta, Id, current_term, 9), + 9 = ra_log_meta:fetch(ra_log_meta, Id, current_term), + %% stop, drop a STALE meta.dets alongside the existing meta.shu, restart. + %% Because meta.shu already exists it is authoritative; the stale DETS must + %% NOT be migrated over the newer value (regression guard). + application:stop(ra), + timer:sleep(100), + ok = write_dets(DataDir, {Id, 1, 'stale@node', 1}), + {ok, _} = ra:start_in(DataDir), + ok = ra_log_meta:await(ra_log_meta), + 9 = ra_log_meta:fetch(ra_log_meta, Id, current_term), + ok. + +oversized_batch_does_not_hang(Config) -> + %% With the tiny WAL configured in init_per_testcase, a coalesced batch of + %% many distinct uids' last_applied writes exceeds wal_size. The write path + %% must fall back to per-op writes rather than livelocking; if it hung, the + %% await/1 call below would never return and the test would time out. + N = 200, + _ = [ra_log_meta:store(ra_log_meta, + <<"batch_uid_", (integer_to_binary(I))/binary>>, + last_applied, I) || I <- lists:seq(1, N)], + ok = ra_log_meta:await(ra_log_meta), + 5 = ra_log_meta:fetch(ra_log_meta, <<"batch_uid_5">>, last_applied), + N = ra_log_meta:fetch(ra_log_meta, <<"batch_uid_200">>, last_applied), + ok. + +%%% helpers + +%% Restart the ra_log_meta process (as its supervisor would on a crash) and +%% wait for it to finish re-initialising from the shu store. Stopping it +%% triggers a one_for_all restart of the whole log subtree, which on a loaded +%% CI can take a while, so poll for the new process rather than sleeping a +%% fixed amount. +restart_meta() -> + OldPid = whereis(ra_log_meta), + proc_lib:stop(OldPid, shutdown, infinity), + await_meta_restarted(OldPid, 200). + +await_meta_restarted(_OldPid, 0) -> + ct:fail(ra_log_meta_did_not_restart); +await_meta_restarted(OldPid, N) -> + case whereis(ra_log_meta) of + Pid when is_pid(Pid), Pid =/= OldPid -> + %% the new process is registered; make sure it has finished init + try ra_log_meta:await(ra_log_meta) of + ok -> ok + catch + _:_ -> + timer:sleep(25), + await_meta_restarted(OldPid, N - 1) + end; + _ -> + timer:sleep(25), + await_meta_restarted(OldPid, N - 1) + end. + +%% ra stores metadata under //meta.{shu,dets}. +meta_dir(DataDir) -> + filename:join(DataDir, atom_to_list(node())). + +meta_dets_path(DataDir) -> + filename:join(meta_dir(DataDir), "meta.dets"). + +write_dets(DataDir, Row) -> + MetaDir = meta_dir(DataDir), + ok = filelib:ensure_dir(filename:join(MetaDir, "ignore")), + MetaDets = filename:join(MetaDir, "meta.dets"), + {ok, T} = dets:open_file(ra_log_meta_SUITE_dets, [{file, MetaDets}]), + ok = dets:insert(T, Row), + ok = dets:close(T), + ok.