Skip to content

Commit 44ce5d6

Browse files
alcoclaude
andcommitted
Fix race in ConsumerRegistry.unregister_name/1 using atomic match_delete
The previous no-op implementation left stale PIDs in the ETS table. A naive lookup-then-delete fix would still be racey because the entry could be overwritten between the lookup and delete. Using :ets.match_delete/2 atomically deletes the entry only if the pid matches the calling (dying) process. Closes #3864 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 258f1dc commit 44ce5d6

2 files changed

Lines changed: 75 additions & 2 deletions

File tree

‎packages/sync-service/lib/electric/shapes/consumer_registry.ex‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,12 @@ defmodule Electric.Shapes.ConsumerRegistry do
2828
if register_consumer!(pid, shape_handle, ets_name(stack_id)), do: :yes, else: :no
2929
end
3030

31-
# don't unregister when the pid exits -- we have mechanisms to ensure that happens cleanly
32-
def unregister_name({_stack_id, _shape_handle}) do
31+
# Atomically remove the ETS entry only if it still belongs to the calling
32+
# process (the dying consumer). If a replacement consumer has already
33+
# registered under the same shape_handle, match_delete is a no-op because
34+
# the pid won't match.
35+
def unregister_name({stack_id, shape_handle}) do
36+
:ets.match_delete(ets_name(stack_id), {shape_handle, self()})
3337
:ok
3438
end
3539

‎packages/sync-service/test/electric/shapes/consumer_registry_test.exs‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -255,6 +255,75 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
255255
end
256256
end
257257

258+
describe "unregister_name/1" do
259+
test "removes entry when pid matches the calling process", ctx do
260+
handle = "handle-1"
261+
table = ctx.registry_state.table
262+
263+
# Register the current process
264+
:ets.insert(table, {handle, self()})
265+
assert [{^handle, pid}] = :ets.lookup(table, handle)
266+
assert pid == self()
267+
268+
ConsumerRegistry.unregister_name({ctx.stack_id, handle})
269+
270+
assert :ets.lookup(table, handle) == []
271+
end
272+
273+
test "does not remove entry belonging to a different process", ctx do
274+
handle = "handle-1"
275+
table = ctx.registry_state.table
276+
277+
other_pid = spawn(fn -> Process.sleep(:infinity) end)
278+
:ets.insert(table, {handle, other_pid})
279+
280+
ConsumerRegistry.unregister_name({ctx.stack_id, handle})
281+
282+
assert [{^handle, ^other_pid}] = :ets.lookup(table, handle)
283+
end
284+
285+
test "does not remove a replacement consumer's entry (race scenario)", ctx do
286+
handle = "handle-1"
287+
table = ctx.registry_state.table
288+
parent = self()
289+
290+
# Simulate: old consumer (a spawned process) calls unregister_name,
291+
# but a replacement has already registered under the same handle.
292+
replacement_pid = spawn(fn -> Process.sleep(:infinity) end)
293+
294+
# First, register the old process
295+
old_task =
296+
Task.async(fn ->
297+
:ets.insert(table, {handle, self()})
298+
send(parent, :old_registered)
299+
300+
# Wait for the replacement to overwrite
301+
receive do
302+
:proceed_to_unregister -> :ok
303+
end
304+
305+
# Now unregister_name should NOT delete the replacement's entry
306+
ConsumerRegistry.unregister_name({ctx.stack_id, handle})
307+
send(parent, :old_unregistered)
308+
end)
309+
310+
assert_receive :old_registered
311+
312+
# Simulate the replacement consumer registering (delete old + insert new)
313+
:ets.delete(table, handle)
314+
:ets.insert(table, {handle, replacement_pid})
315+
316+
# Let the old process proceed with unregister
317+
send(old_task.pid, :proceed_to_unregister)
318+
assert_receive :old_unregistered
319+
320+
# The replacement's entry must still be there
321+
assert [{^handle, ^replacement_pid}] = :ets.lookup(table, handle)
322+
323+
Task.await(old_task)
324+
end
325+
end
326+
258327
describe "broadcast/1" do
259328
test "sends message to all subscribers" do
260329
pid = self()

0 commit comments

Comments
 (0)