Skip to content

Commit aafa5c1

Browse files
robacourtclaude
andcommitted
Fix never-matches warnings in commit_all_move_positions/1
It is called from terminate/2 after terminate_writer/1 has popped :writer off the state, so the value is no longer typed as %State{} and the %State{} clauses were flagged as never matching. Match a bare map and read the fields via Map.get instead. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 372b3ba commit aafa5c1

1 file changed

Lines changed: 21 additions & 18 deletions

File tree

packages/sync-service/lib/electric/shapes/consumer.ex

Lines changed: 21 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1088,29 +1088,32 @@ defmodule Electric.Shapes.Consumer do
10881088

10891089
# Commit all staged move positions unconditionally. Called from `terminate/2`
10901090
# after `terminate_writer/1` has flushed the writer, so every staged move's
1091-
# splice rows are durable by then.
1092-
defp commit_all_move_positions(%State{staged_move_positions: staged} = state)
1093-
when staged == %{},
1094-
do: state
1091+
# splice rows are durable by then. Note this runs after `terminate_writer/1`
1092+
# has popped `:writer` off the state, so we match a bare map rather than
1093+
# `%State{}`.
1094+
defp commit_all_move_positions(state) do
1095+
staged = Map.get(state, :staged_move_positions, %{})
1096+
storage = Map.get(state, :storage)
1097+
1098+
if staged == %{} or is_nil(storage) do
1099+
state
1100+
else
1101+
positions =
1102+
Enum.reduce(staged, state.move_positions, fn {handle, entries}, acc ->
1103+
{_threshold, lsn} = List.last(entries)
1104+
Map.update(acc, handle, lsn, &LogOffset.max(&1, lsn))
1105+
end)
10951106

1096-
defp commit_all_move_positions(%State{storage: storage} = state) when not is_nil(storage) do
1097-
positions =
1098-
Enum.reduce(state.staged_move_positions, state.move_positions, fn {handle, entries}, acc ->
1099-
{_threshold, lsn} = List.last(entries)
1100-
Map.update(acc, handle, lsn, &LogOffset.max(&1, lsn))
1101-
end)
1107+
try do
1108+
ShapeCache.Storage.set_move_positions!(positions, storage)
1109+
rescue
1110+
_ -> :ok
1111+
end
11021112

1103-
try do
1104-
ShapeCache.Storage.set_move_positions!(positions, storage)
1105-
rescue
1106-
_ -> :ok
1113+
%{state | move_positions: positions, staged_move_positions: %{}}
11071114
end
1108-
1109-
%{state | move_positions: positions, staged_move_positions: %{}}
11101115
end
11111116

1112-
defp commit_all_move_positions(state), do: state
1113-
11141117
defp move_pipeline_fully_drained?(%EventHandler.Subqueries.Steady{queue: queue}),
11151118
do: MoveQueue.length(queue) == 0
11161119

0 commit comments

Comments
 (0)