4 changed files with 1639 additions and 41 deletions
@@ -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 |
@@ -371,6 +371,10 @@ retry_activity(F, Alias, #{activity := #{ type := Type
            return_abort(Type, error, retry_limit)
    end.

+%% Ctxt maps carry rocksdb opaque handles (tx | batch). Dialyzer reports
+%% Wopaque_union on try_f/2 when retry Ctxt (tx handle) is unified with the
+%% batch-activity Ctxt used on the first attempt via do_activity/3.
+-dialyzer({no_opaque, retry_activity_/4}).
 retry_activity_(inner, F, Alias, Ctxt) ->
    mrdb_stats:incr(Alias, inner_retries, 1),
    try_f(F, Ctxt);
@@ -811,8 +815,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) ->
+                insert_set(R, EK, EV, Ix, 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 +830,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 +1285,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) ->
+                delete_set(R, EK, D, Ix, 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 +1623,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;
@@ -16,6 +16,13 @@
         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
+       , change_table_type_sync/1
+       , change_table_type_async/1
+       , change_table_type_schema_only/1
+       , change_table_type_no_change/1
        ]).

 -include_lib("common_test/include/ct.hrl").
@@ -31,7 +38,14 @@ 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
+                             , change_table_type_sync
+                             , change_table_type_async
+                             , change_table_type_schema_only
+                             , change_table_type_no_change ]}
    ].

 init_per_suite(Config) ->
@@ -125,6 +139,222 @@ 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.
+
+%% =====================================================================
+%% change_table_type/3 and /4
+%% =====================================================================
+
+%% change_table_type/3 defaults to wait=true: copy + finalize in one call.
+%% set (term keys) -> ordered_set implies default sext key encoding migration.
+change_table_type_sync(_Config) ->
+    ok = create_tab(ctt_s, [{attributes, [k, v]}]),
+    set = mnesia:table_info(ctt_s, type),
+    [mrdb:insert(ctt_s, {ctt_s, I, I * 2}) || I <- lists:seq(1, 40)],
+    Ref0 = mrdb:get_ref(ctt_s),
+    {term, _} = maps:get(encoding, Ref0),
+    set = maps:get(type, maps:get(properties, Ref0)),
+    ok = mnesia_rocksdb_admin:change_table_type(
+           rdb, ctt_s, #{type => ordered_set, report => false}),
+    idle = mnesia_rocksdb_admin:migration_status(ctt_s),
+    Ref1 = mrdb:get_ref(ctt_s),
+    false = maps:is_key(migration, Ref1),
+    {sext, _} = maps:get(encoding, Ref1),
+    ordered_set = maps:get(type, maps:get(properties, Ref1)),
+    ordered_set = maps:get(semantics, Ref1),
+    ordered_set = mnesia:table_info(ctt_s, type),
+    assert_mrdb_encoding_up(ctt_s, {sext, {value, term}}),
+    Objs = lists:sort(mrdb:select(ctt_s, [{'_', [], ['$_']}])),
+    40 = length(Objs),
+    {ok, _} = mrdb:rdb_get(Ref1, sext:encode(1), []),
+    %% Writes after change use ordered_set / sext
+    ok = mrdb:insert(ctt_s, {ctt_s, 100, 200}),
+    {ctt_s, 100, 200} = lists:keyfind(100, 2, mrdb:select(ctt_s, [{'_', [], ['$_']}])),
+    {ok, _} = mrdb:rdb_get(mrdb:get_ref(ctt_s), sext:encode(100), []),
+    ok.
+
+%% change_table_type/4 defaults to wait=false: start only; finalize manually.
+change_table_type_async(_Config) ->
+    ok = create_tab(ctt_a, [{attributes, [k, v]}]),
+    [mrdb:insert(ctt_a, {ctt_a, I, I}) || I <- lists:seq(1, 25)],
+    Self = self(),
+    ok = mnesia_rocksdb_admin:change_table_type(
+           rdb, ctt_a, #{type => ordered_set, encoding => {sext, {value, term}}},
+           Self),
+    %% Dual-write should be on immediately; type not cut over yet.
+    #{migration := _} = mrdb:get_ref(ctt_a),
+    set = mnesia:table_info(ctt_a, type),
+    {term, _} = maps:get(encoding, mrdb:get_ref(ctt_a)),
+    mrdb:insert(ctt_a, {ctt_a, 50, 50}),
+    mrdb:delete(ctt_a, 3),
+    ok = wait_copy_done(ctt_a, 100),
+    ok = mnesia_rocksdb_admin:finalize_migration(rdb, ctt_a),
+    idle = mnesia_rocksdb_admin:migration_status(ctt_a),
+    Ref1 = mrdb:get_ref(ctt_a),
+    false = maps:is_key(migration, Ref1),
+    {sext, _} = maps:get(encoding, Ref1),
+    ordered_set = mnesia:table_info(ctt_a, type),
+    ordered_set = maps:get(type, maps:get(properties, Ref1)),
+    Objs = lists:sort(mrdb:select(ctt_a, [{'_', [], ['$_']}])),
+    false = lists:keymember(3, 2, Objs),
+    true = lists:keymember(50, 2, Objs),
+    %% 25 - 1 delete + 1 insert = 25
+    25 = length(Objs),
+    {ok, _} = mrdb:rdb_get(Ref1, sext:encode(50), []),
+    ok.
+
+%% Type change only when encoding already matches the default for the new type
+%% (set with sext -> ordered_set with same sext): schema/metadata, no CF copy.
+change_table_type_schema_only(_Config) ->
+    ok = create_tab(ctt_so,
+                    [{attributes, [k, v]},
+                     {user_properties,
+                      [{mrdb_encoding, {sext, {value, term}}}]}]),
+    set = mnesia:table_info(ctt_so, type),
+    Ref0 = mrdb:get_ref(ctt_so),
+    {sext, _} = maps:get(encoding, Ref0),
+    CfGen0 = maps:get(cf_gen, Ref0, 0),
+    [mrdb:insert(ctt_so, {ctt_so, I, I}) || I <- lists:seq(1, 10)],
+    ok = mnesia_rocksdb_admin:change_table_type(
+           rdb, ctt_so, #{type => ordered_set, report => false}),
+    idle = mnesia_rocksdb_admin:migration_status(ctt_so),
+    Ref1 = mrdb:get_ref(ctt_so),
+    false = maps:is_key(migration, Ref1),
+    {sext, _} = maps:get(encoding, Ref1),
+    CfGen0 = maps:get(cf_gen, Ref1, 0),
+    ordered_set = mnesia:table_info(ctt_so, type),
+    ordered_set = maps:get(type, maps:get(properties, Ref1)),
+    ordered_set = maps:get(semantics, Ref1),
+    10 = length(mrdb:select(ctt_so, [{'_', [], ['$_']}])),
+    {ok, _} = mrdb:rdb_get(Ref1, sext:encode(1), []),
+    ok.
+
+%% Idempotent / no-op errors.
+change_table_type_no_change(_Config) ->
+    ok = create_tab(ctt_nc, [{attributes, [k, v]}]),
+    set = mnesia:table_info(ctt_nc, type),
+    {error, no_change} =
+        mnesia_rocksdb_admin:change_table_type(
+          rdb, ctt_nc, #{type => set, report => false}),
+    %% Same encoding as live ref
+    {term, ValEnc} = maps:get(encoding, mrdb:get_ref(ctt_nc)),
+    {error, no_change} =
+        mnesia_rocksdb_admin:change_table_type(
+          rdb, ctt_nc, #{encoding => {term, ValEnc}, report => false}),
+    ok.
+
+assert_mrdb_encoding_up(Tab, Enc) ->
+    UPs = mnesia:table_info(Tab, user_properties),
+    {mrdb_encoding, Enc} = lists:keyfind(mrdb_encoding, 1, UPs),
+    #{properties := #{user_properties := UPMap}} = mrdb:get_ref(Tab),
+    {mrdb_encoding, Enc} = maps:get(mrdb_encoding, UPMap),
+    ok.
+
 ok({ok, Value}) -> Value.

 tr_opts() ->