this repo has no description
Fix BeginBuffer liveness check not taking teams into account
+29
-16
+29
-16
lib/servers/begin_buffer.ex
+29
-16
lib/servers/begin_buffer.ex
···
138
138
end
139
139
140
140
defp maybe_complete_flush(%State{} = state) do
141
-
# We use the replication_factor for the current generation to do the liveness check
142
-
%TLogGeneration{replication_factor: replication_factor} = hd(state.cluster.tlog_generations)
141
+
%{
142
+
cluster: cluster,
143
+
check_locked_reply_ids: check_locked_reply_ids,
144
+
} = state
143
145
144
-
# TODO: handle teams
145
-
case length(state.check_locked_reply_ids) >= replication_factor do
146
-
true ->
147
-
read_version = state.batch_read_version
146
+
# Retrieve TLog teams for the current generation
147
+
%TLogGeneration{} = gen = hd(cluster.tlog_generations)
148
+
tlog_teams = Enum.chunk_every(gen.tlog_ids, gen.replication_factor, gen.replication_factor, :discard)
148
149
149
-
Enum.each(state.batch_buffer, fn from ->
150
-
SimServer.reply(from, {:ok, read_version})
151
-
end)
150
+
# Liveness is confirmed when all TLogs from any team confirm they are not locked,
151
+
# as a recovery must lock at least one TLog from every team to proceed
152
+
liveness_confirmed? = Enum.any?(tlog_teams, fn ids ->
153
+
Enum.all?(ids, fn id -> id in check_locked_reply_ids end)
154
+
end)
152
155
153
-
%{state |
154
-
batch_buffer: nil,
155
-
batch_read_version: nil,
156
-
check_locked_nonce: nil,
157
-
check_locked_reply_ids: nil,
158
-
}
159
-
156
+
case liveness_confirmed? do
157
+
true -> do_complete_flush(state)
160
158
false -> state
161
159
end
160
+
end
161
+
162
+
defp do_complete_flush(%State{} = state) do
163
+
read_version = state.batch_read_version
164
+
165
+
Enum.each(state.batch_buffer, fn from ->
166
+
SimServer.reply(from, {:ok, read_version})
167
+
end)
168
+
169
+
%{state |
170
+
batch_buffer: nil,
171
+
batch_read_version: nil,
172
+
check_locked_nonce: nil,
173
+
check_locked_reply_ids: nil,
174
+
}
162
175
end
163
176
end