From 1a99fec600740ac2b1c2b6f7f8d69fc8b2a37c31 Mon Sep 17 00:00:00 2001 From: Ulf Wiger Date: Tue, 4 Aug 2026 15:30:31 +0200 Subject: [PATCH] WIP migrate table encoding --- doc/online-encoding-migration.md | 440 +++++++++++++++++++++ src/mnesia_rocksdb_admin.erl | 495 ++++++++++++++++++++++-- src/mrdb.erl | 35 +- test/mnesia_rocksdb_migration_SUITE.erl | 115 +++++- 4 files changed, 1056 insertions(+), 29 deletions(-) create mode 100644 doc/online-encoding-migration.md diff --git a/doc/online-encoding-migration.md b/doc/online-encoding-migration.md new file mode 100644 index 0000000..3bdb4fa --- /dev/null +++ b/doc/online-encoding-migration.md @@ -0,0 +1,440 @@ +# Online encoding migration (design note) + +**Status:** design +**Branch:** `uw-migration` +**Scope (MWP):** change **key/value encoding** of an existing **set** table by +copying into a new RocksDB column family while the table remains online. + +Changing Mnesia `type` (`set` ↔ `ordered_set`) is **out of scope** for the first +cut. For `mrdb` users, what matters for range seeks and key order is the +**rocksdb key encoding**, not the Mnesia metadata type. Aligning Mnesia type +via a custom schema transaction is possible later and needs separate design. + +--- + +## 1. Motivation + +Some tables were created with encodings that are awkward for production use. +Example: integer primary keys with `{term, {object, term}}` encoding. The +Erlang external term format preserves sort order for positive integers only +up to 32-bit max; beyond that keys use `SMALL_BIG_EXT` and **byte order ≠ +numeric order**. RocksDB iterators and seeks then mis-order heights. + +Historically apps worked around this by creating a second table +(`gmmp_tallies` → `gmmp_tallies2`) and migrating in application code. That is +painful and easy to get wrong. + +This feature moves **encoding migration into the backend**: dual-write to a +new column family (CF), copy old → new, then flip the live `db_ref` and drop +the old CF. + +--- + +## 2. Goals and non-goals + +### Goals (MWP) + +- Online migration of **encoding** for **set** tables (semantics `set`). +- Table remains readable and writable during migration. +- Safe under concurrent inserts/updates/deletes. +- Resume after process/node restart (without a durable RocksDB snapshot). +- Indexes (ordered) continue to work: updated on live writes as today; + index CFs are **not** dual-copied for encoding-only main-table migrate. +- Explicit start/complete as admin/schema-level operations (not silent). + +### Non-goals (MWP) + +- **Bag** tables (supported for completeness; poor fit for rocksdb dual-write). +- Changing Mnesia **type** (`set` / `ordered_set` / `bag`) in this pass. +- Rebuilding or re-encoding **index** column families. +- Multi-node coordinated migration (single backend instance first). +- Retaining a RocksDB snapshot across process restart. +- Record shape / arity / attribute renames (use `mnesia:transform_table` + or app logic). + +--- + +## 3. Existing building blocks + +| Piece | Role today | +|---|---| +| `persistent_term` meta map | `Name → db_ref()` via `put_pt` / `get_ref` / `ensure_ref` | +| CF create/drop | `create_column_family` / `drop_column_family` | +| Offline standalone→CF migrate | `migrate_standalone/2`: chunk `select`, insert new, delete old, `put_pt` | +| Snapshots | `mrdb:snapshot/1`, `release_snapshot/1` (DB-instance scoped) | +| Hot write path | `insert_` / `delete_` encode with ref’s `encoding` | + +The new feature is an **online, same-DB CF** migrate with dual-write, not a +reimplementation of standalone migration. + +--- + +## 4. Design overview + +### 4.0 Column-family naming (versions) + +RocksDB CF names must be **unique** within a DB. The existing map is: + +| Logical resource | Physical CF name | +|---|---| +| data table `Tab` | `"{d, Tab}"` (gen 0) | +| index | `"{i, Tab, I}"` | +| retainer | `"{r, Tab, R}"` | +| admin | `"{a, Alias}"` | + +That mapping has **no version**. Encoding migration needs the old and new data +CFs to coexist until cutover, so data CFs are versioned: + +| Generation | Physical CF name | +|---|---| +| 0 (legacy / first create) | `"{d, Tab}"` | +| N ≥ 1 | `"{d, Tab, N}"` | + +- `tab_to_cf_name(Tab)` still produces gen 0 for brand-new tables. +- Migration target = `data_cf_name(Tab, LiveGen + 1)`. +- Live gen is stored on the `db_ref` as `cf_gen` / `cf_name` and durably as + admin info `cf_gen` for the table. +- `cf_name_to_tab/2` maps **any** data gen back to logical `Tab`. +- On open, if several data gens exist for `Tab`, prefer the highest gen for + which we have a handle; refine with durable `cf_gen` when present (interrupted + migration may leave an incomplete higher gen — recovery should prefer the + recorded live gen once that path is fully wired). + +**Cutover does not recreate a “canonical” CF name.** The versioned target CF +*is* the new live CF; the previous generation is dropped. No second full copy. + +``` + ┌─────────────────────────────────────┐ + │ Old CF gen G (reads; encoding E_old)│ + │ name: {d,Tab} or {d,Tab,G} │ + └──────────────┬──────────────────────┘ + │ dual-write + │ (live updates) + ▼ + ┌─────────────────────────────────────┐ + │ New CF gen G+1 (encoding E_new) │ + │ name: {d,Tab,G+1} │ + │ + live dual-writes │ + │ + migrator copy (skip if exists) │ + └─────────────────────────────────────┘ + │ + ┌──────────────┴──────────────────────┐ + │ Admin info │ + │ encoding_migration meta, cf_gen │ + └─────────────────────────────────────┘ +``` + +### 4.1 Phases + +| Phase | Reads | Writes | Notes | +|---|---|---|---| +| `idle` | current CF | current CF | Normal operation | +| `dual_write` / `copying` | **old** CF only | old **and** new (re-encode) | Migrator walks old → new | +| `copy_done` | old | dual-write still on | Optional verify | +| `cutover` | **new** | **new** only | PT flip; then drop old + scratch | + +### 4.2 Ref shape during migration + +Live ref in persistent_term (conceptual): + +```erlang +OldRef#{ + migration => NewRef, %% target CF ref (encoding E_new) + migration_epoch => pos_integer(), + migration_phase => copying | copy_done +} +``` + +- **Reads / select / iterators:** use the outer (old) ref only until cutover. +- **Inserts / deletes / merges (set):** apply to old, then if `migration` + is set, re-encode and apply to `NewRef`. +- **Indexes:** `update_index` runs **once**, as today, using the live main ref + (old CF) for `read(R, Key)`. Index CF layout is independent of main + encoding (index keys are `{IxVal, Key}` under the **index** ref’s encoding). + +### 4.3 Migrator install rule + +For each key `K` observed on the migration walk: + +``` +if not present in live old(K) → skip %% deleted after snapshot / walk +else if present in new(K) → skip %% dual-write already installed latest +else put new(live_value(K)) %% re-encode with NewRef +``` + +Rationale: + +- **Skip if new has key** — dual-write wins; migrator must not clobber newer data. +- **Live re-check of old** — avoids **delete resurrection** when iterating a + snapshot that still contains `K` after a dual-delete removed it from both CFs. +- Prefer **live** value over snapshot-only value when installing. + +### 4.4 Snapshot (best-effort) + +While the migrator process lives: + +1. `Snapshot = rocksdb:snapshot(DbRef)` (same DB as old/new CFs). +2. Iterate **old** CF with `{snapshot, Snapshot}` read options from `cursor`. +3. Apply the install rule above (always re-check live old + new). + +On migrator crash/restart: snapshot is gone. Resume from scratch **cursor** +on **live** old CF with the same install rule. Correctness does not depend on +the snapshot; the snapshot only reduces iterator churn. + +### 4.5 Scratch metadata + +Either a dedicated scratch CF or admin-info keys (admin CF is enough for MWP). + +Minimum fields: + +```erlang +#{ table => tabname() + , phase => dual_write | copying | copy_done | cutover + , target => #{encoding := encoding()} %% validated + , cursor => '$first' | Key :: term() %% last successfully processed logical key + , epoch => pos_integer() + , started_at => millisecond() + } +``` + +- Cursor is the **logical** key (Erlang term), not the encoded rocksdb key. +- Resume seek on old CF uses **old** encoding: `encode_key(Cursor, OldRef)`. + +### 4.6 Cutover + +1. Ensure `phase = copy_done` (and optional verification). +2. **Barrier:** no in-flight activities that still hold a pre-migration ref + map (see §5.2), or epoch re-resolution (see below). +3. Schema/admin transaction: + - `put_pt(Tab, NewRef)` without `migration` (NewRef is gen G+1 CF) + - persist live `cf_gen = G+1` in admin info + - drop **old** generation CF only + - clear migration meta +4. Subsequent `get_ref(Tab)` returns new encoding and `cf_gen = G+1`. + +No rename and no second bulk copy: the migration target CF is permanent. + +### 4.7 Mnesia type deferred + +For `mrdb` callers, ordered seeks and range logic depend on **key encoding** +(e.g. `sext` vs `term`), not on `mnesia:table_info(Tab, type)`. + +Changing Mnesia type later may use a custom schema transaction so metadata +matches reality; that must not be conflated with CF encoding migrate and needs +extra care around `table_info`, load hooks, and any code that branches on type. + +--- + +## 5. Concurrency and correctness + +### 5.1 Dual-write before copy + +Order of start: + +1. Create empty new CF (target encoding). +2. Persist migration meta + install `migration => NewRef` on live ref (**dual-write ON**). +3. Only then start the copy walk. + +If copy ran before dual-write, concurrent writes could land only on old and be +missed or overwritten incorrectly. + +### 5.2 Stale in-process `db_ref` maps + +`ensure_ref/1` short-circuits when given a map (activities/batches hold refs). +After cutover, a process still holding the **old** map would write a CF that +is about to be dropped. + +MWP options (pick at least one): + +1. **Quiesce:** refuse cutover while table activity is non-zero / use a + short exclusive period. +2. **Epoch:** bump `migration_epoch`; `ensure_ref(Map)` re-fetches from PT if + map epoch ≠ current. +3. **Delay drop:** keep old CF read-only until epoch drained (refcounting). + +Recommended MWP: **epoch on ref + re-resolve when mismatch**, plus **delay +drop of old CF** until a quiet period or explicit admin “finalize”. + +### 5.3 Transactions and batches + +Prefer one RocksDB `WriteBatch` (same `db_ref`, two `cf_handle`s) so dual-write +insert/delete is atomic across old and new under `as_batch`. + +Index updates stay in the same batch as today (`batch_if_index`). + +### 5.4 Indexes and encoding independence + +Index CFs store `{IxVal, MainKey}` under the **index** encoding. Main encoding +change does not require rewriting index keys. Live writes continue to maintain +indexes against the logical object; after cutover, `index_read` still yields +`MainKey` terms and reads the new main CF with the new encoding. + +**Do not** dual-write index CFs in MWP encoding migration. + +### 5.5 Crash recovery + +On admin/backend restart: + +1. Reload CF handles and PT from durable admin state. +2. If migration meta says `copying` / `dual_write` / `copy_done`: + - Re-bind live ref with `migration => NewRef`. + - Resume dual-write. + - If not `copy_done`, restart migrator from `cursor` (live iterate). +3. Never drop old CF unless cutover fully committed. + +--- + +## 6. API sketch (MWP) + +```erlang +%% Start (admin / schema-level). Validates encoding; rejects bag; rejects +%% unsupported type changes. +-spec migrate_encoding(alias(), tabname(), encoding()) -> ok | {error, term()}. +migrate_encoding(Alias, Tab, NewEncoding) -> ... + +%% Optional: progress / status +-spec migration_status(tabname()) -> + idle | #{phase := atom(), cursor := term(), target := map()}. + +%% Complete cutover (if not automatic after copy_done + barrier) +-spec finalize_migration(alias(), tabname()) -> ok | {error, term()}. +``` + +Reporting can mirror `migrate_standalone/3` (`Rpt` pid + progress messages). + +--- + +## 7. Risks and open points + +| Risk | Mitigation | +|---|---| +| Delete resurrection under snapshot | Live re-check before install (§4.3) | +| Stale activity refs after cutover | Epoch re-resolve + delayed CF drop (§5.2) | +| Partial dual-write in batch | Multi-CF WriteBatch (§5.3) | +| PT lost while admin meta says migrating | Recovery rebuilds PT from admin (§5.5) | +| Huge tables | Chunked walk + cursor; progress reports | +| bag / type / index encoding | Explicitly out of MWP | + +Open (later): + +- Automatic vs manual finalize after `copy_done`. +- Optional full verification (count / sample) before cutover. +- Mnesia type alignment schema transaction. +- Bag and index-encoding migration. + +--- + +## 8. Relation to application workarounds + +Application dual tables (`*2`) remain valid for record-shape changes. Backend +encoding migration is aimed at cases like “same record, wrong key encoding” +without app-visible table rename. + +Example target: migrate `gmmp_generations` from `{term,{object,term}}` to a +sext (or other order-preserving) key encoding so height seeks stay valid past +2³² − 1 without a `generations2` table. + +--- + +# Implementation checklist + +Phased so each phase is testable. Bags and Mnesia type change stay excluded. + +## Phase 0 — Spec freeze and hooks + +- [ ] Freeze MWP scope: **set + encoding only**; document rejections for bag / type. +- [ ] Define durable migration meta schema (admin info keys vs scratch CF). +- [ ] Define public API: `migrate_encoding/3`, `migration_status/1`, + `finalize_migration/2` (names flexible). +- [ ] Define progress report messages (align with `migrate_standalone` rpt). +- [ ] List all write paths that must dual-write: insert, delete, delete_object, + update_counter/merge (if encoding-compatible), clear_table policy + (reject migrate while clearing / dual-clear both). + +## Phase 1 — Ref + dual-write plumbing + +- [ ] Extend live `db_ref` with optional `migration`, `migration_epoch`, + `migration_phase` (or keep phase only in admin meta). +- [ ] `ensure_ref/1`: when map has stale epoch, re-fetch from PT (or document + quiesce-only MWP if epoch deferred). +- [ ] Implement `dual_put` / `dual_delete` helpers: old encoding + new encoding; + same WriteBatch when `as_batch` active. +- [ ] Wire dual-write into `insert_` / `delete_` / related set paths only. +- [ ] Confirm `update_index` still runs once and reads from **old** main ref + during migration; no index dual-write. +- [ ] Unit tests: concurrent insert/delete while `migration` set updates both + CFs; index remains consistent with logical data. + +## Phase 2 — Create target CF + start migration + +- [ ] Validate `NewEncoding` via `mnesia_rocksdb_lib:check_encoding/2`. +- [ ] Reject bag semantics; reject no-op (same encoding); reject if migration + already active. +- [ ] Create empty new CF with target encoding (and consistent vsn/access_type). +- [ ] Persist migration meta; install dual-write on live ref (schema/admin txn). +- [ ] Test: after start, reads still see old data; writes appear on both CFs + (inspect via raw/ref). + +## Phase 3 — Copy walk + scratch cursor + +- [ ] Migrator process (supervised or admin-linked) with optional RocksDB snapshot. +- [ ] Iterate old CF from cursor; for each object apply install rule (§4.3): + - [ ] live old missing → skip + - [ ] new present → skip + - [ ] else put re-encoded live value to new +- [ ] Update scratch cursor after each chunk (batch cursor updates). +- [ ] Progress reporting every N keys. +- [ ] On completion set `phase = copy_done`. +- [ ] Tests: + - [ ] empty table + - [ ] populated table, no concurrent writes + - [ ] concurrent overwrites (new has key → migrator skips; final value correct) + - [ ] concurrent deletes (no resurrection) + - [ ] restart migrator mid-way (cursor resume, no snapshot) + +## Phase 4 — Recovery + +- [ ] On backend/admin start: detect incomplete migration from durable meta. +- [ ] Re-install dual-write ref; restart migrator from cursor if needed. +- [ ] Test kill -9 / admin restart mid-copy; complete and verify data. + +## Phase 5 — Cutover + drop + +- [ ] Barrier policy implemented (epoch and/or quiesce and/or delayed drop). +- [ ] `finalize_migration`: PT → NewRef only; clear migration fields. +- [ ] Drop old CF; clear scratch/meta. +- [ ] Test: post-cutover read/write/select/index_read; encoding is E_new. +- [ ] Test: in-flight activity behaviour under chosen barrier policy. + +## Phase 6 — Optional verify + polish + +- [ ] Optional key-count or sample verification under dual-write before cutover. +- [ ] Metrics: keys copied, skipped (exists), skipped (deleted), duration. +- [ ] Docs: user guide section + this design note linked from README. +- [ ] Dialyzer / property tests for encode round-trip old→new. + +## Phase 7 — Explicitly later (not MWP) + +- [ ] Mnesia `type` change via custom schema transaction. +- [ ] Bag table migration. +- [ ] Index CF encoding rebuild. +- [ ] Multi-node / multi-replica orchestration. + +--- + +## 9. Suggested first spike (days, not weeks) + +1. Dual-write only (no migrator): force `migration => NewRef` in test; prove + insert/delete hit both CFs; indexes still correct. +2. Migrator on quiescent table: full copy + cutover + drop. +3. Add concurrent writes/deletes + install rule. +4. Add cursor + restart. +5. Snapshot as optimization only. + +--- + +## 10. Document history + +| Date | Note | +|---|---| +| 2026-08-03 | Initial design from gm_mining_pool / generations encoding discussion | diff --git a/src/mnesia_rocksdb_admin.erl b/src/mnesia_rocksdb_admin.erl index dda1e16..32ea43f 100644 --- a/src/mnesia_rocksdb_admin.erl +++ b/src/mnesia_rocksdb_admin.erl @@ -19,7 +19,13 @@ ]). -export([ migrate_standalone/2 - , migrate_standalone/3 ]). + , migrate_standalone/3 + %% Online encoding migration (set tables): dual-write + CF copy + , migrate_encoding/3 + , migrate_encoding/4 + , migration_status/1 + , finalize_migration/2 + ]). -export([ start_link/0 , init/1 @@ -49,6 +55,7 @@ backends = #{} :: #{ alias() => backend() } , standalone = #{} :: #{{alias(), table()} := cf() } , default_opts = [] :: [{atom(), _}] + , migrators = #{} :: #{ table() => pid() } }). -type st() :: #st{}. @@ -79,6 +86,9 @@ | {write_table_property, tabname(), tuple()} | {remove_aliases, [alias()]} | {migrate, [tabname() | {tabname(), map()}], rpt()} + | {migrate_encoding, tabname(), any(), rpt()} + | {migration_status, tabname()} + | {finalize_migration, tabname()} | {prep_close, table()} | {close_table, table()} | {clear_table, table() | cf() }. @@ -289,6 +299,39 @@ migrate_standalone(Alias, Tabs, Rpt0) -> end, call(Alias, {migrate, Tabs, Rpt}). +%% @doc Start online encoding migration for a set table (dual-write + CF copy). +-spec migrate_encoding(alias(), tabname(), Encoding :: any()) -> + ok | {error, term()}. +migrate_encoding(Alias, Tab, Encoding) -> + migrate_encoding(Alias, Tab, Encoding, undefined). + +-spec migrate_encoding(alias(), tabname(), Encoding :: any(), rpt()) -> + ok | {error, term()}. +migrate_encoding(Alias, Tab, Encoding, Rpt0) -> + Rpt = case Rpt0 of + undefined -> undefined; + To when is_pid(To); is_atom(To) -> + #{to => To, tag => migrate_encoding}; + #{} = M -> M + end, + call(Alias, {migrate_encoding, Tab, Encoding, Rpt}). + +-spec migration_status(tabname()) -> idle | map(). +migration_status(Tab) when is_atom(Tab) -> + case get_ref(Tab, error) of + #{migration := _} = Ref -> + case maps:get(migration_meta, Ref, undefined) of + #{} = Meta -> Meta; + undefined -> #{phase => dual_write} + end; + _ -> + idle + end. + +-spec finalize_migration(alias(), tabname()) -> ok | {error, term()}. +finalize_migration(Alias, Tab) -> + call(Alias, {finalize_migration, Tab}). + -spec call(alias() | [], req()) -> no_return() | any(). call(Alias, Req) -> call(Alias, Req, infinity). @@ -451,6 +494,15 @@ handle_info({mnesia_table_event, Event}, St) -> _ -> {noreply, St} end; +handle_info({'DOWN', _MRef, process, Pid, Reason}, + #st{migrators = Ms} = St) -> + case lists:keyfind(Pid, 2, maps:to_list(Ms)) of + {Tab, Pid} -> + ?log(info, "Encoding migrator for ~p exited: ~p", [Tab, Reason]), + {noreply, St#st{migrators = maps:remove(Tab, Ms)}}; + false -> + {noreply, St} + end; handle_info(_Msg, St) -> {noreply, St}. @@ -593,16 +645,36 @@ handle_req(Alias, {migrate, Tabs0, Rpt}, Backend, St) -> {reply, Res, St1}; {error, _} = Error -> {reply, Error, St} - end. + end; +handle_req(Alias, {migrate_encoding, Tab, Encoding, Rpt}, Backend, St) -> + case start_encoding_migration(Alias, Tab, Encoding, Rpt, Backend, St) of + {ok, St1} -> + {reply, ok, St1}; + {error, _} = Error -> + {reply, Error, St} + end; +handle_req(Alias, {finalize_migration, Tab}, Backend, St) -> + case do_finalize_encoding_migration(Alias, Tab, Backend, St) of + {ok, St1} -> + {reply, ok, St1}; + {error, _} = Error -> + {reply, Error, St} + end; +handle_req(_Alias, {migration_status, Tab}, _Backend, St) -> + {reply, migration_status(Tab), St}. handle_load_table_req(Alias, Name, TRec, Backend, St) -> case create_table_from_trec(Alias, Name, TRec, Backend, St) of {ok, TRec1, St1} -> TRec2 = TRec1#{status => open}, - St2 = update_cf(Alias, Name, TRec2, St1), + Backend1 = maps:get(Alias, St1#st.backends, Backend), + {TRec3, St2} = + maybe_resume_encoding_migration( + Alias, Name, TRec2, Backend1, St1), + St3 = update_cf(Alias, Name, TRec3, St2), ?log(debug, "Table loaded ~p", [Name]), - put_pt(Name, TRec2), - {reply, {ok, TRec2}, St2}; + put_pt(Name, TRec3), + {reply, {ok, TRec3}, St3}; {error, _} = Error -> {reply, Error, St} end. @@ -1158,8 +1230,11 @@ create_table_as_cf(Alias, Name, #{db_ref := DbRef} = R, St) -> CfName = tab_to_cf_name(Name), case create_column_family(DbRef, CfName, cfopts(), R) of {ok, CfH} -> - R1 = check_version_and_encoding(R#{ cf_handle => CfH - , type => column_family }), + R1 = check_version_and_encoding( + R#{ cf_handle => CfH + , type => column_family + , cf_gen => maps:get(cf_gen, R, 0) + , cf_name => CfName }), {ok, R1, update_cf(Alias, Name, R1, St)}; {error, _} = Error -> Error @@ -1363,37 +1438,413 @@ admin_cfs({info, _} = I) -> [ {tab_to_cf_name(I), cfopts()} ]. map_cfs({ok, Ref, CfHandles}, CFs, Alias, Acc) -> ZippedCFs = lists:zip(CFs, CfHandles), - %% io:fwrite("ZippedCFs = ~p~n", [ZippedCFs]), - CfInfo = maps:from_list( - [{cf_name_to_tab(N, Alias), #{ db_ref => Ref - , cf_handle => H - , alias => Alias - , status => pre_existing - , type => column_family }} - || {{N,_}, H} <- ZippedCFs]), + %% Group versioned data CFs by logical tab and generation. cf_info keeps one + %% provisional live handle per logical name (highest gen); full map is in + %% data_cf_by_gen for migration resume (source gen may be lower than target). + {CfInfo, ByGen} = + lists:foldl( + fun({{N, _}, H}, {Map, BG}) -> + Logical = cf_name_to_tab(N, Alias), + Gen = case cf_name_to_data_gen(N) of + {ok, _, G} -> G; + error -> 0 + end, + Rec = #{ db_ref => Ref + , cf_handle => H + , alias => Alias + , status => pre_existing + , type => column_family + , cf_gen => Gen + , cf_name => N }, + BG1 = case is_atom(Logical) of + true -> + GMap0 = maps:get(Logical, BG, #{}), + BG#{Logical => GMap0#{Gen => Rec}}; + false -> + BG + end, + Map1 = case maps:find(Logical, Map) of + {ok, #{cf_gen := OldG}} + when is_integer(OldG), OldG > Gen -> + Map; + _ -> + Map#{Logical => Rec} + end, + {Map1, BG1} + end, {#{}, #{}}, ZippedCFs), {ok, Acc#{ db_ref => Ref - , cf_info => CfInfo }}. + , cf_info => CfInfo + , data_cf_by_gen => ByGen }}. -tab_to_cf_name(Tab) when is_atom(Tab) -> write_term({d, Tab}); +%% Column-family naming +%% --------------------- +%% RocksDB requires unique CF names within a DB. Logical table names (atoms) +%% are not enough once we rewrite a table into a new CF (encoding migration): +%% the old and new CFs must coexist until cutover. +%% +%% Data CFs are versioned: +%% gen 0 (legacy / first create): "{d, Tab}" +%% gen N (N >= 1): "{d, Tab, N}" +%% +%% The live generation is stored on the db_ref as `cf_gen` and durably in +%% admin info `{cf_gen, Tab}`. `cf_name_to_tab/2` maps any generation back to +%% the logical table name for open/recovery; when several gens exist, the +%% recorded live gen wins (see map_cfs / resolve_data_cf). +%% +%% Index / retainer / admin CFs are unchanged: one logical resource ↔ one CF. + +tab_to_cf_name(Tab) when is_atom(Tab) -> + data_cf_name(Tab, 0); tab_to_cf_name({admin, Alias}) -> write_term({a, Alias}); tab_to_cf_name({info, Tab}) -> write_term({n, Tab}); tab_to_cf_name({Tab, index, I}) -> write_term({i, Tab, I}); tab_to_cf_name({Tab, retainer, R}) -> write_term({r, Tab, R}). +%% Physical CF name for a data table generation. +data_cf_name(Tab, 0) when is_atom(Tab) -> + write_term({d, Tab}); +data_cf_name(Tab, Gen) when is_atom(Tab), is_integer(Gen), Gen > 0 -> + write_term({d, Tab, Gen}). + write_term(T) -> lists:flatten(io_lib:fwrite("~w", [T])). cf_name_to_tab(Cf, Alias) -> case read_term(Cf) of - {ok, {d, Table}} -> Table; - {ok, {i, Table, I}} -> {Table, index, I}; - {ok, {r, Table, R}} -> {Table, retainer, R}; - {ok, {n, Table}} -> {info, Table}; - {ok, {a, Alias}} -> {admin, Alias}; + {ok, {d, Table}} -> Table; + {ok, {d, Table, _Gen}} -> Table; %% versioned data CF → logical tab + {ok, {i, Table, I}} -> {Table, index, I}; + {ok, {r, Table, R}} -> {Table, retainer, R}; + {ok, {n, Table}} -> {info, Table}; + {ok, {a, Alias}} -> {admin, Alias}; _ -> {ext, Alias, Cf} end. +%% Parse generation from a data CF name string; non-data → error. +cf_name_to_data_gen(Cf) -> + case read_term(Cf) of + {ok, {d, Table}} -> {ok, Table, 0}; + {ok, {d, Table, Gen}} -> {ok, Table, Gen}; + _ -> + error + end. + +%% ===================================================================== +%% Online encoding migration (set tables) +%% ===================================================================== + +start_encoding_migration(Alias, Tab, Encoding0, Rpt, Backend, St) + when is_atom(Tab) -> + case find_cf(Alias, Tab, Backend, St) of + {ok, #{status := open, semantics := bag}} -> + {error, bag_not_supported}; + {ok, #{status := open, migration := _}} -> + {error, already_migrating}; + {ok, #{status := open} = OldRef} -> + case maps:get(type, OldRef, column_family) of + standalone -> + {error, standalone_not_supported}; + column_family -> + start_encoding_migration_(Alias, Tab, Encoding0, Rpt, + OldRef, Backend, St) + end; + {ok, _} -> + {error, not_open}; + error -> + {error, not_found} + end; +start_encoding_migration(_, _, _, _, _, _) -> + {error, badarg}. + +start_encoding_migration_(Alias, Tab, Encoding0, Rpt, OldRef, Backend, St) -> + As = maps:get(attributes, + maps:get(properties, OldRef, #{}), + [key, val]), + case mnesia_rocksdb_lib:check_encoding(Encoding0, As) of + {ok, NewEnc} -> + case maps:get(encoding, OldRef) of + NewEnc -> + {error, same_encoding}; + _OldEnc -> + create_and_start_encoding_mig( + Alias, Tab, NewEnc, Rpt, OldRef, Backend, St) + end; + {error, _} = Err -> + Err + end. + +create_and_start_encoding_mig(Alias, Tab, NewEnc, Rpt, OldRef, + #{db_ref := DbRef} = _Backend, St) -> + OldGen = maps:get(cf_gen, OldRef, 0), + NewGen = OldGen + 1, + CfName = data_cf_name(Tab, NewGen), + case create_column_family(DbRef, CfName, cfopts(), OldRef) of + {ok, CfH} -> + %% Target ref: same DB, new versioned CF, new encoding; no migration field. + NewRef0 = maps:without( + [migration, migration_meta, migration_epoch], + OldRef), + NewRef0b = NewRef0#{ cf_handle => CfH + , encoding => NewEnc + , cf_gen => NewGen + , cf_name => CfName + , name => Tab + , type => column_family + , status => open }, + %% Persist encoding in user_properties *before* check_version_and_encoding + %% so it is not replaced by the table default. + NewRef1 = update_user_properties( + {mrdb_encoding, NewEnc}, + check_version_and_encoding(NewRef0b)), + NewRef = NewRef1#{encoding => NewEnc, cf_gen => NewGen, cf_name => CfName}, + Meta = #{ phase => copying + , target => #{encoding => NewEnc, cf_gen => NewGen} + , source_gen => OldGen + , cursor => '$first' + , epoch => 1 + , started_at => erlang:system_time(millisecond) }, + %% Live ref: dual-write to NewRef; reads still use Old CF handle. + LiveRef = OldRef#{ migration => NewRef + , migration_meta => Meta + , migration_epoch => 1 + , cf_gen => OldGen }, + write_info(Alias, Tab, encoding_migration, Meta), + %% Keep live encoding as old for correct reads; only target has NewEnc. + %% Schema mrdb_encoding is updated on finalize. + LiveRef2 = LiveRef#{migration := NewRef}, + St1 = update_cf(Alias, Tab, LiveRef2, St), + put_pt(Tab, LiveRef2), + {Pid, _MRef} = spawn_monitor( + fun() -> + encoding_migrator(Alias, Tab, LiveRef2, + NewRef, Rpt) + end), + rpt(Rpt, "Started encoding migration ~p gen ~p -> ~p (~s)~n", + [Tab, OldGen, NewGen, CfName]), + {ok, St1#st{migrators = maps:put(Tab, Pid, St#st.migrators)}}; + {error, _} = Err -> + Err + end. + +%% Re-attach dual-write (and copier) after restart when durable meta says so. +maybe_resume_encoding_migration(Alias, Name, LiveRef, Backend, St) + when is_atom(Name) -> + case read_info(Alias, Name, encoding_migration, undefined) of + #{phase := Phase, target := #{encoding := Enc, cf_gen := NewGen}} = Meta + when Phase =:= copying; Phase =:= copy_done -> + SourceGen = maps:get(source_gen, Meta, NewGen - 1), + ByGen = maps:get(data_cf_by_gen, Backend, #{}), + TabGens = maps:get(Name, ByGen, #{}), + case {maps:find(SourceGen, TabGens), maps:find(NewGen, TabGens)} of + {{ok, SrcRec}, {ok, TgtRec}} -> + resume_encoding_migration_( + Alias, Name, LiveRef, Meta, Enc, NewGen, SourceGen, + SrcRec, TgtRec, Phase, St); + _ -> + ?log(warning, + "encoding migration meta for ~p but CFs missing " + "(source=~p target=~p gens=~p)", + [Name, SourceGen, NewGen, maps:keys(TabGens)]), + {LiveRef, St} + end; + _ -> + %% Completed migration: ensure live handle matches durable cf_gen. + maybe_align_cf_gen(Alias, Name, LiveRef, Backend, St) + end; +maybe_resume_encoding_migration(_, _, LiveRef, _, St) -> + {LiveRef, St}. + +maybe_align_cf_gen(Alias, Name, LiveRef, Backend, St) -> + case read_info(Alias, Name, cf_gen, undefined) of + Gen when is_integer(Gen) -> + ByGen = maps:get(data_cf_by_gen, Backend, #{}), + case maps:find(Gen, maps:get(Name, ByGen, #{})) of + {ok, Rec} -> + Live1 = maps:merge( + LiveRef, + maps:with([cf_handle, cf_gen, cf_name, db_ref], + Rec)), + {Live1, St}; + error -> + {LiveRef, St} + end; + _ -> + {LiveRef, St} + end. + +resume_encoding_migration_(Alias, Name, LiveRef, Meta, Enc, NewGen, SourceGen, + SrcRec, TgtRec, Phase, St) -> + %% Live reads use source gen CF + old encoding (from LiveRef / schema). + Live0 = maps:merge( + LiveRef, + maps:with([cf_handle, cf_name, db_ref], SrcRec)), + Live1 = Live0#{cf_gen => SourceGen, status => open}, + NewRef0 = maps:without( + [migration, migration_meta, migration_epoch], LiveRef), + NewRef = NewRef0#{ cf_handle => maps:get(cf_handle, TgtRec) + , cf_name => maps:get(cf_name, TgtRec) + , cf_gen => NewGen + , encoding => Enc + , db_ref => maps:get(db_ref, TgtRec) + , name => Name + , status => open }, + NewRef1 = update_user_properties({mrdb_encoding, Enc}, NewRef), + NewRef2 = NewRef1#{encoding => Enc}, + Meta1 = Meta#{phase => Phase}, + Live2 = Live1#{ migration => NewRef2 + , migration_meta => Meta1 + , migration_epoch => maps:get(epoch, Meta, 1) }, + put_pt(Name, Live2), + St1 = case Phase of + copying -> + case maps:get(Name, St#st.migrators, undefined) of + Pid when is_pid(Pid) -> + St; + _ -> + {Pid, _} = spawn_monitor( + fun() -> + encoding_migrator( + Alias, Name, Live2, NewRef2, + undefined) + end), + St#st{migrators = maps:put(Name, Pid, St#st.migrators)} + end; + copy_done -> + St + end, + {Live2, St1}. + +%% Background copier: walk old CF, install into new if missing (no clobber). +encoding_migrator(Alias, Tab, OldRef, NewRef, Rpt) -> + try + Chunk = 500, + N = encoding_copy_loop(OldRef, NewRef, '$first', 0, Chunk, Rpt), + Meta = #{ phase => copy_done + , target => #{encoding => maps:get(encoding, NewRef)} + , cursor => '$end' + , epoch => 1 + , copied => N + , finished_at => erlang:system_time(millisecond) }, + %% Update live PT meta to copy_done (keep dual-write). + case get_ref(Tab, error) of + #{migration := _} = Live -> + put_pt(Tab, Live#{migration_meta => Meta}); + _ -> + ok + end, + write_info(Alias, Tab, encoding_migration, Meta), + rpt(Rpt, "Encoding migration copy done for ~p (~p objs)~n", [Tab, N]), + ok + catch + C:R:ST -> + ?log(error, "encoding_migrator ~p failed: ~p:~p / ~p", + [Tab, C, R, ST]), + error({C, R}) + end. + +encoding_copy_loop(OldRef, NewRef, Cursor, N, Chunk, Rpt) -> + {Batch, Cont} = encoding_select_chunk(OldRef, Cursor, Chunk), + N1 = lists:foldl( + fun(Obj, Acc) -> + encoding_maybe_install(OldRef, NewRef, Obj), + Acc + 1 + end, N, Batch), + case Cont of + '$end_of_table' -> + N1; + NextCursor -> + maybe_progress(Rpt, N1), + encoding_copy_loop(OldRef, NewRef, NextCursor, N1, Chunk, Rpt) + end. + +%% Chunked select on the old CF. Continuations use mrdb:select/1 when available. +encoding_select_chunk(OldRef, '$first', Limit) -> + case mrdb:select(OldRef, [{'_', [], ['$_']}], Limit) of + {L, Cont} when is_list(L) -> + {L, {cont, Cont}}; + L when is_list(L) -> + {L, '$end_of_table'}; + '$end_of_table' -> + {[], '$end_of_table'} + end; +encoding_select_chunk(_OldRef, {cont, Cont}, _Limit) -> + case mrdb:select(Cont) of + {L, Cont1} when is_list(L) -> + {L, {cont, Cont1}}; + L when is_list(L) -> + {L, '$end_of_table'}; + '$end_of_table' -> + {[], '$end_of_table'} + end; +encoding_select_chunk(OldRef, _Other, Limit) -> + %% Fallback: restart from first (safe, may re-process; install is idempotent). + encoding_select_chunk(OldRef, '$first', Limit). + +encoding_maybe_install(OldRef, NewRef, Obj) -> + Name = maps:get(name, OldRef), + KP = mnesia_rocksdb_lib:keypos(Name), + Key = element(KP, Obj), + OldOnly = maps:without( + [migration, migration_meta, migration_epoch], OldRef), + %% Live re-check on old (source of truth for reads during migration). + case mrdb:read(OldOnly, Key) of + [] -> + ok; %% deleted after we observed Obj + [LiveObj] -> + case mrdb:read(NewRef, Key) of + [_] -> + ok; %% dual-write already installed + [] -> + mrdb:insert(NewRef, LiveObj) + end + end. + +do_finalize_encoding_migration(Alias, Tab, _Backend, St) -> + %% Prefer PT (migrator updates phase there) over gen_server state. + case get_ref(Tab, error) of + #{migration := MigRef} = LiveRef -> + Meta = maps:get(migration_meta, LiveRef, #{}), + case maps:get(phase, Meta, undefined) of + copy_done -> + finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, St); + Phase -> + {error, {not_ready, Phase}} + end; + error -> + {error, not_found}; + _ -> + {error, not_migrating} + end. + +%% Cutover: make the versioned target CF the live one and drop the old CF. +%% No second copy — the migration target *is* the new generation. +finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, St) -> + #{db_ref := DbRef, cf_handle := OldCfH} = LiveRef, + NewEnc = maps:get(encoding, MigRef), + NewGen = maps:get(cf_gen, MigRef, maps:get(cf_gen, LiveRef, 0) + 1), + LiveNew = maps:without( + [migration, migration_meta, migration_epoch], + MigRef#{ name => Tab + , status => open + , encoding => NewEnc + , cf_gen => NewGen }), + LiveNew1 = update_user_properties({mrdb_encoding, NewEnc}, LiveNew), + put_pt(Tab, LiveNew1), + St1 = update_cf(Alias, Tab, LiveNew1, St), + %% Durable live generation + schema encoding for restart. + write_info(Alias, Tab, cf_gen, NewGen), + delete_info(Alias, Tab, encoding_migration), + _ = maybe_write_user_props(LiveNew1), + %% Drop previous generation CF (by handle; name may be {d,Tab} or {d,Tab,G}). + ok = rocksdb:drop_column_family(DbRef, OldCfH), + catch rocksdb:destroy_column_family(DbRef, OldCfH), + drop_cached_cf(maps:get(cf_name, LiveRef, + data_cf_name(Tab, maps:get(cf_gen, LiveRef, 0))), + OldCfH), + St2 = St1#st{migrators = maps:remove(Tab, St1#st.migrators)}, + {ok, St2}. + read_term(Str) -> case erl_scan:string(Str) of {ok, Tokens, _} -> diff --git a/src/mrdb.erl b/src/mrdb.erl index 8148eda..fe5736b 100644 --- a/src/mrdb.erl +++ b/src/mrdb.erl @@ -811,8 +811,12 @@ insert_(#{semantics := bag} = Ref, Key, EncKey, EncVal, Obj, Opts) -> batch_if_index(Ref, insert, bag, fun insert_bag/5, Key, EncKey, EncVal, Obj, Opts); %% insert_bag(Ref, Obj, Opts); insert_(Ref, Key, EncKey, EncVal, Obj, Opts) -> - batch_if_index(Ref, insert, set, fun insert_set/5, Key, EncKey, EncVal, Obj, Opts). - %% insert_set(Ref, Obj, Opts). + %% Close over Key/Obj so dual-write can re-encode into migration target CF. + F = fun(R, EK, EV, _Ix, Os) -> + rdb_put(R, EK, EV, Os), + dual_put(R, Key, Obj, Os) + end, + batch_if_index(Ref, insert, set, F, Key, EncKey, EncVal, Obj, Opts). insert_set(Ref, EncKey, EncVal, _, Opts) -> rdb_put(Ref, EncKey, EncVal, Opts). @@ -822,6 +826,19 @@ insert_bag(Ref, EncKey, EncVal, _, Opts) -> %% #{vsn := 1} -> insert_bag_v1(Ref, EncKey, EncVal, Opts). +%% During online encoding migration, mirror set put/delete onto the target CF. +dual_put(#{migration := Mig0} = R, Key, Obj, Opts) -> + Mig = maps:merge(Mig0, maps:with([activity, snapshot], R)), + rdb_put(Mig, encode_key(Key, Mig), encode_val(Obj, Mig), Opts); +dual_put(_, _, _, _) -> + ok. + +dual_delete(#{migration := Mig0} = R, Key, Opts) -> + Mig = maps:merge(Mig0, maps:with([activity, snapshot], R)), + rdb_delete(Mig, encode_key(Key, Mig), Opts); +dual_delete(_, _, _) -> + ok. + batch_if_index(#{mode := mnesia} = Ref, _, _, F, _Key, EncKey, Data, _Obj, Opts) -> F(Ref, EncKey, Data, undefined, Opts); batch_if_index(#{name := Name, properties := #{index := [_|_] = Ixs}} = Ref, @@ -1264,7 +1281,11 @@ delete(Tab, Key, Opts) -> delete_(#{semantics := bag} = Ref, Key, EncKey, Opts) -> batch_if_index(Ref, delete, bag, fun delete_bag/5, Key, EncKey, [], [], Opts); delete_(Ref, Key, EncKey, Opts) -> - batch_if_index(Ref, delete, set, fun delete_set/5, Key, EncKey, [], [], Opts). + F = fun(R, EK, _D, _Ix, Os) -> + rdb_delete(R, EK, Os), + dual_delete(R, Key, Os) + end, + batch_if_index(Ref, delete, set, F, Key, EncKey, [], [], Opts). delete_object(Tab, Obj) -> delete_object(Tab, Obj, []). @@ -1598,15 +1619,17 @@ do_del_obj_bag_(Sz, K, Res, Obj, #{name := Name} = Ref, I, Opts) -> delete_obj_set(_, _, _, not_found, _) -> ok; -delete_obj_set(Ref, _, _, RawKey, Opts) when is_binary(RawKey) -> - rdb_delete(Ref, RawKey, Opts); +delete_obj_set(#{name := Name} = Ref, _, Obj, RawKey, Opts) when is_binary(RawKey) -> + rdb_delete(Ref, RawKey, Opts), + dual_delete(Ref, element(keypos(Name), Obj), Opts); delete_obj_set(#{name := Name} = Ref, EncKey, Obj, _, Opts) -> case rdb_get(Ref, EncKey, []) of {ok, Bin} -> Key = element(keypos(Name), Obj), case decode_val(Bin, Key, Ref) of Obj -> - rdb_delete(Ref, EncKey, Opts); + rdb_delete(Ref, EncKey, Opts), + dual_delete(Ref, Key, Opts); _ -> ok end; diff --git a/test/mnesia_rocksdb_migration_SUITE.erl b/test/mnesia_rocksdb_migration_SUITE.erl index 7e81cc2..c5cd3b2 100644 --- a/test/mnesia_rocksdb_migration_SUITE.erl +++ b/test/mnesia_rocksdb_migration_SUITE.erl @@ -16,6 +16,9 @@ manual_migration/1 , migrate_with_encoding_change/1 , auto_migration/1 + , online_encoding_migration/1 + , online_encoding_migration_restart/1 + , online_encoding_migration_interrupt/1 ]). -include_lib("common_test/include/ct.hrl"). @@ -31,7 +34,10 @@ all() -> groups() -> [ {all_tests, [sequence], [ manual_migration - , migrate_with_encoding_change ]} + , migrate_with_encoding_change + , online_encoding_migration + , online_encoding_migration_restart + , online_encoding_migration_interrupt ]} ]. init_per_suite(Config) -> @@ -125,6 +131,113 @@ migrate_with_encoding_change(_Config) -> auto_migration(_Config) -> ok. +%% Online encoding migration: term keys -> sext keys while dual-writing. +online_encoding_migration(_Config) -> + ok = create_tab(enc, [{attributes, [k, v]}]), + %% Default set encoding is {term, {value, term}} for 2-attr tables. + lists:foreach( + fun(I) -> + mrdb:insert(enc, {enc, I, I * 10}) + end, lists:seq(1, 50)), + 50 = length(mrdb:select(enc, [{'_', [], ['$_']}])), + Ref0 = mrdb:get_ref(enc), + ct:log("Before migration ref: ~p", [maps:with([encoding, type, semantics], Ref0)]), + {term, _} = maps:get(encoding, Ref0), + %% Concurrent writes during migration + ok = mnesia_rocksdb_admin:migrate_encoding( + rdb, enc, {sext, {value, term}}), + %% Dual-write should be active + #{migration := _} = mrdb:get_ref(enc), + mrdb:insert(enc, {enc, 100, 1000}), + mrdb:delete(enc, 5), + %% Wait for copy_done + ok = wait_copy_done(enc, 50), + ct:log("Status after copy: ~p", [mnesia_rocksdb_admin:migration_status(enc)]), + ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc), + Ref1 = mrdb:get_ref(enc), + ct:log("After finalize ref: ~p", [maps:with([encoding, type], Ref1)]), + {sext, _} = maps:get(encoding, Ref1), + false = maps:is_key(migration, Ref1), + Objs = lists:sort(mrdb:select(enc, [{'_', [], ['$_']}])), + %% 50 - 1 deleted + 1 inserted = 50 + 50 = length(Objs), + false = lists:keymember(5, 2, Objs), + true = lists:keymember(100, 2, Objs), + {enc, 100, 1000} = lists:keyfind(100, 2, Objs), + %% sext-encoded key is readable via rdb_get + {ok, _} = mrdb:rdb_get(Ref1, sext:encode(100), []), + ok. + +wait_copy_done(_Tab, 0) -> + {error, timeout}; +wait_copy_done(Tab, N) -> + case mnesia_rocksdb_admin:migration_status(Tab) of + #{phase := copy_done} -> + ok; + _ -> + timer:sleep(50), + wait_copy_done(Tab, N - 1) + end. + +%% Complete migration, restart mnesia, verify encoding + data survive. +online_encoding_migration_restart(_Config) -> + ok = create_tab(enc_r, [{attributes, [k, v]}]), + [mrdb:insert(enc_r, {enc_r, I, I * 3}) || I <- lists:seq(1, 30)], + ok = mnesia_rocksdb_admin:migrate_encoding( + rdb, enc_r, {sext, {value, term}}), + ok = wait_copy_done(enc_r, 100), + ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc_r), + {sext, _} = maps:get(encoding, mrdb:get_ref(enc_r)), + Before = lists:sort(mrdb:select(enc_r, [{'_', [], ['$_']}])), + 30 = length(Before), + %% Close and reopen DB (same VM: stop/start mnesia + wait for tables). + stopped = mnesia:stop(), + ok = mnesia:start(), + ok = mnesia:wait_for_tables([enc_r], 10000), + Ref = mrdb:get_ref(enc_r), + ct:log("After restart ref: ~p", + [maps:with([encoding, cf_gen, type], Ref)]), + {sext, _} = maps:get(encoding, Ref), + false = maps:is_key(migration, Ref), + After = lists:sort(mrdb:select(enc_r, [{'_', [], ['$_']}])), + Before = After, + {ok, _} = mrdb:rdb_get(Ref, sext:encode(1), []), + ok. + +%% Stop mid-migration (after dual-write is up), reopen, finish copy + finalize. +online_encoding_migration_interrupt(_Config) -> + ok = create_tab(enc_i, [{attributes, [k, v]}]), + [mrdb:insert(enc_i, {enc_i, I, I}) || I <- lists:seq(1, 100)], + ok = mnesia_rocksdb_admin:migrate_encoding( + rdb, enc_i, {sext, {value, term}}), + %% Dual-write active; do not wait for copy_done — interrupt promptly. + #{migration := _} = mrdb:get_ref(enc_i), + mrdb:insert(enc_i, {enc_i, 200, 200}), + mrdb:delete(enc_i, 10), + stopped = mnesia:stop(), + ok = mnesia:start(), + ok = mnesia:wait_for_tables([enc_i], 10000), + %% Migration should be re-armed (dual-write) or already copy_done. + case mnesia_rocksdb_admin:migration_status(enc_i) of + #{phase := Phase} when Phase =:= copying; Phase =:= copy_done -> + ok; + Other -> + ct:fail({expected_migration_after_restart, Other}) + end, + %% Writes during resumed dual-write + mrdb:insert(enc_i, {enc_i, 201, 201}), + ok = wait_copy_done(enc_i, 200), + ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc_i), + {sext, _} = maps:get(encoding, mrdb:get_ref(enc_i)), + false = maps:is_key(migration, mrdb:get_ref(enc_i)), + Objs = lists:sort(mrdb:select(enc_i, [{'_', [], ['$_']}])), + false = lists:keymember(10, 2, Objs), + true = lists:keymember(200, 2, Objs), + true = lists:keymember(201, 2, Objs), + %% 100 - 1 delete + 2 inserts = 101 + 101 = length(Objs), + ok. + ok({ok, Value}) -> Value. tr_opts() ->