fix(runtime): delete owned ETS tables before exit signals

CHANGELOG.md

@@ -2,6 +2,16 @@
 
 All notable milestones for the nera project are documented in this file.
 
+## Process exits release ETS tables before exit signals
+
+- A dead process's ETS tables are deleted before linked processes receive
+  `EXIT` or monitoring processes receive `DOWN`, as on BEAM. This also covers
+  processes killed by a linked process's exit.
+- A child restarted by its supervisor can immediately create the same named
+  table without racing the previous owner's cleanup.
+- The `ets_freed_before_signal` regression checks five exit paths and repeated
+  supervised restarts, unchanged natively and on BEAM.
+
 ## Stored bitstrings keep their bytes and exact bit length
 
 - ETS and `persistent_term` now own copies of non-byte-aligned bitstrings,

crates/nera-runtime/src/ets.rs

@@ -3701,9 +3701,9 @@ mod tests {
     }
 
     // Regression: a process exit releases the names of the named tables it
-    // owned, and only those. The Erlang fixture ets_named_table_return cannot
-    // check this deterministically: natively the monitor message is sent
-    // before the exiting process's tables are deleted.
+    // owned, and only those. The scheduler's exit_deletes_ets_before_any_signal
+    // test owns the ordering relative to exit signals; this test owns isolation
+    // between named and unnamed tables belonging to different processes.
     #[test]
     fn owner_exit_releases_only_its_named_table() {
         let name = global_atom_table().intern("test_owner_exit_named");

crates/nera-runtime/src/scheduler.rs

@@ -636,16 +636,17 @@ impl Scheduler {
 // EXIT / DOWN signal propagation
 // ---------------------------------------------------------------------------
 
-/// Free a dead process's registered name, then propagate EXIT signals to
+/// Free a dead process's registered name and ETS tables, then propagate EXIT signals to
 /// linked processes and DOWN signals to monitoring processes. Called from the
 /// exit points in `worker_loop` AFTER dropping the dying process's MutexGuard
 /// but BEFORE removing it from the process table (so its heap stays alive for
 /// deep-copy).
 ///
-/// The name is freed first, as on BEAM: a process woken by the 'EXIT' or
-/// 'DOWN' on another worker, such as a supervisor restarting a named child,
-/// must find the name free. A process killed by the link cascade below frees
-/// its own name the same way before its signals go out.
+/// The name and tables are freed first, as on BEAM: a process woken by the
+/// 'EXIT' or 'DOWN' on another worker, such as a supervisor restarting a child,
+/// must be able to reuse its registered name and named tables. A process killed
+/// by the link cascade below releases its own resources the same way before
+/// its signals go out.
 ///
 /// Deadlock safety: the dying process's lock is NOT held. Each linked/monitoring
 /// process is locked individually, and the lock is dropped before any recursive
@@ -675,6 +676,7 @@ fn propagate_exit_signals(
     };
 
     unregister_dead_process(dead_pid, registered_names);
+    crate::ets::delete_tables_owned_by(dead_pid);
 
     let is_kill = exit_reason.as_atom_id() == Some(ATOM_KILL);
     let is_normal = exit_reason.as_atom_id() == Some(ATOM_NORMAL);
@@ -798,7 +800,6 @@ fn propagate_exit_signals(
             #[cfg(mio_supported)]
             reactor.cancel_for_pid(their_pid);
             crate::socket::remove_all_for_pid(their_pid);
-            crate::ets::delete_tables_owned_by(their_pid);
             alive_count.fetch_sub(1, Ordering::AcqRel);
             process_table.remove(&their_pid);
         }
@@ -922,7 +923,6 @@ fn cleanup_dead_process(
     #[cfg(mio_supported)]
     reactor.cancel_for_pid(pid);
     crate::socket::remove_all_for_pid(pid);
-    crate::ets::delete_tables_owned_by(pid);
     alive_count.fetch_sub(1, Ordering::AcqRel);
     process_table.remove(&pid);
 }
@@ -1636,7 +1636,6 @@ fn worker_loop(
                     #[cfg(mio_supported)]
                     reactor.cancel_for_pid(pid);
                     crate::socket::remove_all_for_pid(pid);
-                    crate::ets::delete_tables_owned_by(pid);
                     alive_count.fetch_sub(1, Ordering::AcqRel);
                     process_table.remove(&pid);
                 } else {
@@ -2066,6 +2065,105 @@ mod tests {
         assert_eq!(owner.inbox.len(), 1, "the owner then gets the linked process's 'DOWN'");
     }
 
+    // A supervisor may recreate a named table as soon as EXIT/DOWN arrives.
+    // Hold the observer's lock so propagation cannot deliver its first DOWN.
+    // Both direct exit and a linked-exit cascade must already have deleted
+    // their named and unnamed tables, without touching another owner's table.
+    #[test]
+    fn exit_deletes_ets_before_any_signal() {
+        let mut outcomes = Vec::new();
+        for cascade in [false, true] {
+            let scheduler = Scheduler::new(1);
+            // Dedicated PIDs avoid the process-global ETS fixtures in other tests.
+            let dead_pid = ProcessId::new(9_400_000 + u64::from(cascade) * 10);
+            let linked_pid = ProcessId::new(dead_pid.get() + 1);
+            scheduler.spawn_process(Process::new(dead_pid));
+            scheduler.process_table.get(&dead_pid).unwrap().lock_for_runtime().transition(ProcessState::Exited);
+            let observer_pid = scheduler.spawn_process(Process::new(ProcessId::new(dead_pid.get() + 2)));
+            let observer = scheduler.process_table.get(&observer_pid).unwrap().clone();
+            let monitored_pid = if cascade { linked_pid } else { dead_pid };
+            let monitor = 1_u64;
+            observer.lock_for_runtime().add_outgoing_monitor(monitor, monitored_pid);
+            let monitors = [(monitor, observer_pid)].into_iter().collect();
+            let (links, monitors) = if cascade {
+                scheduler.spawn_process(Process::new(linked_pid));
+                let linked = scheduler.process_table.get(&linked_pid).unwrap().clone();
+                let mut proc = linked.lock_for_runtime();
+                proc.add_link(dead_pid);
+                proc.otp.monitors_incoming = monitors;
+                ([linked_pid].into_iter().collect(), HashMap::new())
+            } else {
+                (HashSet::new(), monitors)
+            };
+            let atoms = crate::atom_table::global_atom_table();
+            let named = [("named_table".to_string(), ErlTerm::nil())];
+            let mut owned_tables = Vec::new();
+            for pid in [dead_pid, monitored_pid].into_iter().collect::<HashSet<_>>() {
+                let name = atoms.intern(&format!("exit_order_ets_{}", pid.get()));
+                owned_tables.push(crate::ets::ets_new_handle(name, &named, pid).unwrap());
+                owned_tables.push(crate::ets::ets_new_handle(name, &[], pid).unwrap());
+            }
+            let survivor_name = atoms.intern(&format!("exit_order_survivor_{}", dead_pid.get()));
+            let survivor = crate::ets::ets_new_handle(survivor_name, &named, observer_pid).unwrap();
+            #[cfg(mio_supported)]
+            let (reactor, join) = crate::reactor::spawn_reactor();
+
+            let guard = observer.lock_for_runtime();
+            let mut heap = crate::heap::ProcessHeap::new();
+            let undefined = ErlTerm::atom(crate::atom_table::ATOM_UNDEFINED);
+            let removed_before_down = thread::scope(|scope| {
+                let exiting = scope.spawn(|| {
+                    cleanup_dead_process(
+                        dead_pid,
+                        ErlTerm::atom(crate::atom_table::ATOM_BADARG),
+                        links,
+                        monitors,
+                        &scheduler.process_table,
+                        &scheduler.injector,
+                        &scheduler.runnable_count,
+                        &scheduler.alive_count,
+                        &scheduler.any_crash,
+                        &scheduler.initial_pid,
+                        &scheduler.registered_names,
+                        &scheduler.timer_wheel,
+                        &scheduler.io_pool,
+                        #[cfg(mio_supported)]
+                        &reactor,
+                    );
+                });
+                let deadline = Instant::now() + Duration::from_secs(5);
+                let removed = loop {
+                    if owned_tables.iter().all(|&table| {
+                        crate::ets::ets_info(table, Some("owner"), &mut heap) == undefined
+                    }) {
+                        break true;
+                    }
+                    if Instant::now() >= deadline { break false; }
+                    thread::yield_now();
+                };
+                let undelivered = observer.inbox.is_empty();
+                drop(guard);
+                exiting.join().unwrap();
+                removed && undelivered
+            });
+            #[cfg(mio_supported)]
+            { reactor.shutdown(); join.join().unwrap(); }
+
+            assert_eq!(observer.inbox.len(), 1, "the observer then gets DOWN");
+            assert_eq!(crate::ets::ets_info(survivor, Some("owner"), &mut heap), ErlTerm::pid(observer_pid));
+            assert!(crate::ets::ets_delete_table(survivor));
+            // Recreate each name for the observer, not the exited owner.
+            for table in owned_tables.into_iter().filter(|table| table.is_atom()) {
+                let name = table.as_atom_id().unwrap();
+                assert_eq!(crate::ets::ets_new_handle(name, &named, observer_pid), Some(table));
+                assert!(crate::ets::ets_delete_table(table));
+            }
+            outcomes.push((cascade, removed_before_down));
+        }
+        assert!(outcomes.iter().all(|(_, removed)| *removed),
+            "direct and cascading exits delete ETS before DOWN: {outcomes:?}");
+    }
+
     // supervisor_shutdown sometimes found a stopped child still alive: the
     // supervisor got the child's 'DOWN' and exited, and the test got the
     // supervisor's 'DOWN' and called is_process_alive/1 on the child, all

tests/README.md

@@ -461,6 +461,7 @@ These are automatically skipped by the runner and show up as SKIP in output.
 |------------------|---------------------------------------|------------|
 | `ets_basic.erl`  | ETS new/insert/lookup/delete/tab2list | ~15        |
 | `ets_keypos.erl` | ETS keypos and named tables           | ~8         |
+| `ets_freed_before_signal.erl` | A dead owner's named table can be recreated at the first EXIT or DOWN: normal exit, own reason, trapped link, kill and link cascade, 500 rounds each; a supervised gen_server creates it on each of 100 restarts; also run on BEAM | 5 exit paths + supervisor |
 | `ets_stored_fun.erl` | A fun stored in ETS (at the top of a row, nested in a list, tuple and map, after `update_element/3`) or `persistent_term` by a process that then exits or collects garbage keeps its captured list, binaries, tuple, map, float, bignum and inner fun when looked up and called; a fun read by `tab2list/1` or `persistent_term:get/1` still works after the table or term is deleted; completion marker; also run on BEAM | ~16 |
 | `ets_stored_bitstring.erl` | Non-byte-aligned heap/refc bitstrings, sub-byte values and extracted tails stored in ETS or `persistent_term` survive writer exit/GC; boxed `update_element/3` replacements and copying reads remain intact after table deletion or term erasure; completion marker; same fixture on BEAM | 4 lifetime scenarios |
 | `ets_bitstring_keys.erl` | Partial-bitstring bag duplicate suppression; copied direct/list/tuple keys in hash-backed tables for lookup, membership, insert_new, boxed updates, counters, deletion and forward/backward traversal; copied values as persistent_term keys; significant bits and exact lengths remain distinct; completion marker; same fixture on BEAM | Table-driven |

tests/ets_freed_before_signal.erl

@@ -0,0 +1,126 @@
+%% EXPECT: ets_freed_before_signal_ok
+%% A dead owner's tables must be gone before its first EXIT or DOWN, as on
+%% BEAM. Recreate the named table immediately, including in a supervised
+%% gen_server's init. Extra monitors expose cleanup delayed until after signals.
+-module(ets_freed_before_signal).
+-export([test/0, init/1, handle_call/3, handle_cast/2]).
+
+-define(TABLE, freed_ets_table).
+-define(CHILD, freed_ets_child).
+-define(ROUNDS, 500).
+-define(MONITORS, 32).
+-define(RESTARTS, 100).
+
+test() ->
+    process_flag(trap_exit, true),
+    lists:foreach(fun(Scenario) -> rounds(Scenario, ?ROUNDS) end,
+                  [monitor_normal, monitor_crash, link_exit, link_killed,
+                   link_cascade]),
+    ok = supervisor_restarts(),
+    io:format("ets_freed_before_signal_ok~n"),
+    ok.
+
+rounds(_Scenario, 0) -> ok;
+rounds(Scenario, N) ->
+    {Pid, Start, Signal, Reason} = setup(Scenario),
+    Refs = [monitor(process, Pid) || _ <- lists:seq(1, ?MONITORS)],
+    Start(),
+    receive_signal(Signal, Pid, Reason),
+    %% This must succeed on the first attempt, without waiting for cleanup.
+    ?TABLE = ets:new(?TABLE, [named_table, public]),
+    Self = self(),
+    Self = ets:info(?TABLE, owner),
+    [] = ets:lookup(?TABLE, old_owner),
+    true = ets:insert(?TABLE, {replacement, N}),
+    [{replacement, N}] = ets:lookup(?TABLE, replacement),
+    true = ets:delete(?TABLE),
+    Pending = case Signal of down -> length(Refs) - 1; exit -> length(Refs) end,
+    flush_downs(Pid, Pending),
+    rounds(Scenario, N - 1).
+
+setup(monitor_normal) ->
+    Pid = owner(spawn, fun() -> ok end),
+    {Pid, fun() -> Pid ! go end, down, normal};
+setup(monitor_crash) ->
+    Pid = owner(spawn, fun() -> exit(crash) end),
+    {Pid, fun() -> Pid ! go end, down, crash};
+setup(link_exit) ->
+    Pid = owner(spawn_link, fun() -> exit(crash) end),
+    {Pid, fun() -> Pid ! go end, exit, crash};
+setup(link_killed) ->
+    Pid = owner(spawn_link, fun() -> ok end),
+    {Pid, fun() -> exit(Pid, kill) end, exit, killed};
+setup(link_cascade) ->
+    Parent = self(),
+    Leader = spawn(fun() -> receive go -> exit(crash) end end),
+    Pid = spawn(fun() ->
+                        link(Leader),
+                        create_table(),
+                        Parent ! {ready, self()},
+                        receive never -> ok end
+                end),
+    receive {ready, Pid} -> ok end,
+    {Pid, fun() -> Leader ! go end, down, crash}.
+
+owner(Spawn, Exit) ->
+    Parent = self(),
+    Body = fun() ->
+                   create_table(),
+                   Parent ! {ready, self()},
+                   receive go -> Exit() end
+           end,
+    Pid = case Spawn of spawn -> spawn(Body); spawn_link -> spawn_link(Body) end,
+    receive {ready, Pid} -> ok end,
+    Pid.
+
+create_table() ->
+    ?TABLE = ets:new(?TABLE, [named_table, public]),
+    true = ets:insert(?TABLE, {old_owner, self()}),
+    ok.
+
+receive_signal(down, Pid, Reason) ->
+    receive {'DOWN', _, process, Pid, Reason} -> ok end;
+receive_signal(exit, Pid, Reason) ->
+    receive {'EXIT', Pid, Reason} -> ok end.
+
+flush_downs(_Pid, 0) -> ok;
+flush_downs(Pid, N) ->
+    receive {'DOWN', _, process, Pid, _} -> flush_downs(Pid, N - 1) end.
+
+%% The restart limit permits exactly the requested restarts. A failed init
+%% (badarg from ets:new) consumes another attempt and stops the supervisor.
+supervisor_restarts() ->
+    {ok, Sup} = supervisor:start_link(?MODULE, {sup, self()}),
+    Child0 = receive {started, Initial} -> Initial end,
+    ok = kill_and_restart(Child0, ?RESTARTS),
+    Child = whereis(?CHILD),
+    Ref = monitor(process, Child),
+    exit(Sup, kill),
+    receive {'EXIT', Sup, killed} -> ok end,
+    receive {'DOWN', Ref, process, Child, killed} -> ok end,
+    undefined = ets:info(?TABLE, owner),
+    ok.
+
+kill_and_restart(_Child, 0) -> ok;
+kill_and_restart(Child, N) ->
+    exit(Child, kill),
+    New = receive
+              {started, Next} when Next =/= Child -> Next;
+              {'EXIT', Sup, Reason} -> exit({supervisor_stopped, Sup, Reason})
+          end,
+    New = ets:info(?TABLE, owner),
+    ok = gen_server:call(New, ready),
+    kill_and_restart(New, N - 1).
+
+init({sup, Parent}) ->
+    {ok, {{one_for_one, ?RESTARTS, 3600},
+          [{?CHILD, {gen_server, start_link,
+                     [{local, ?CHILD}, ?MODULE, {child, Parent}, []]},
+            permanent, brutal_kill, worker, [?MODULE]}]}};
+init({child, Parent}) ->
+    create_table(),
+    Parent ! {started, self()},
+    {ok, child}.
+
+handle_call(_Request, _From, State) -> {reply, ok, State}.
+handle_cast(_Request, State) -> {noreply, State}.