Skip to content

Commit 8090602

Browse files
committed
fix(api_call): protect node rotation and health state with a mutex
`@current_node_index` and per-node `is_healthy`/`last_access_timestamp` were unsynchronised. With multiple threads sharing one client (Puma, Sidekiq), two threads in `next_node` could compute the same incremented index from the same starting value (skipping a node), and a healthcheck write could be observed mid-update with a stale timestamp. Wraps the round-robin loop in `next_node` and the writes in `set_node_healthcheck` in a single per-instance Mutex. The mutex is acquired briefly (no I/O inside it), so contention is negligible. Adds regression specs covering concurrent multi-node rotation and concurrent single-node health updates.
1 parent f33f224 commit 8090602

2 files changed

Lines changed: 72 additions & 9 deletions

File tree

lib/typesense/api_call.rb

Lines changed: 15 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ def initialize(configuration)
2222

2323
@logger = @configuration.logger
2424

25+
@nodes_mutex = Mutex.new
2526
initialize_metadata_for_nodes
2627
@current_node_index = -1
2728
end
@@ -141,14 +142,17 @@ def next_node
141142

142143
# Fallback to nodes as usual
143144
@logger.debug "Nodes health: #{@nodes.each_with_index.map { |node, i| "Node #{i} is #{node[:is_healthy] == true ? 'Healthy' : 'Unhealthy'}" }.join(' || ')}"
144-
candidate_node = nil
145-
(0..@nodes.length).each do |_i|
146-
@current_node_index = (@current_node_index + 1) % @nodes.length
147-
candidate_node = @nodes[@current_node_index]
148-
if candidate_node[:is_healthy] == true || node_due_for_healthcheck?(candidate_node)
149-
@logger.debug "Updated current node to Node #{candidate_node[:index]}"
150-
return candidate_node
145+
candidate_node = @nodes_mutex.synchronize do
146+
node = nil
147+
(0..@nodes.length).each do |_i|
148+
@current_node_index = (@current_node_index + 1) % @nodes.length
149+
node = @nodes[@current_node_index]
150+
if node[:is_healthy] == true || node_due_for_healthcheck?(node)
151+
@logger.debug "Updated current node to Node #{node[:index]}"
152+
return node
153+
end
151154
end
155+
node
152156
end
153157

154158
# None of the nodes are marked healthy, but some of them could have become healthy since last health check.
@@ -175,8 +179,10 @@ def initialize_metadata_for_nodes
175179
end
176180

177181
def set_node_healthcheck(node, is_healthy:)
178-
node[:is_healthy] = is_healthy
179-
node[:last_access_timestamp] = Time.now.to_i
182+
@nodes_mutex.synchronize do
183+
node[:is_healthy] = is_healthy
184+
node[:last_access_timestamp] = Time.now.to_i
185+
end
180186
end
181187

182188
def custom_exception_klass_for(response)

spec/typesense/api_call_spec.rb

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -258,4 +258,61 @@
258258
it_behaves_like 'General error handling', :delete
259259
it_behaves_like 'Node selection', :delete
260260
end
261+
262+
describe 'concurrent node rotation' do
263+
it 'distributes selection evenly across nodes when called from many threads' do
264+
thread_count = 16
265+
iterations_per_thread = 90
266+
num_nodes = typesense.configuration.nodes.length
267+
268+
counts = Array.new(num_nodes, 0)
269+
counts_mutex = Mutex.new
270+
271+
threads = Array.new(thread_count) do
272+
Thread.new do
273+
local_counts = Array.new(num_nodes, 0)
274+
iterations_per_thread.times do
275+
node = api_call.send(:next_node)
276+
local_counts[node[:index]] += 1
277+
end
278+
counts_mutex.synchronize do
279+
local_counts.each_with_index { |c, i| counts[i] += c }
280+
end
281+
end
282+
end
283+
threads.each(&:join)
284+
285+
expected_per_node = (thread_count * iterations_per_thread) / num_nodes
286+
expect(counts).to all(eq(expected_per_node))
287+
end
288+
289+
context 'with a single node' do
290+
let(:typesense) do
291+
Typesense::Client.new(
292+
api_key: 'abcd',
293+
nodes: [{ host: 'node0', port: 8108, protocol: 'http' }],
294+
connection_timeout_seconds: 10,
295+
retry_interval_seconds: 0.01,
296+
log_level: Logger::ERROR
297+
)
298+
end
299+
300+
it 'returns the single node and keeps health state consistent under concurrent writes' do
301+
node = typesense.configuration.nodes[0]
302+
303+
threads = Array.new(8) do |i|
304+
Thread.new do
305+
50.times do
306+
api_call.send(:set_node_healthcheck, node, is_healthy: i.even?)
307+
api_call.send(:next_node)
308+
end
309+
end
310+
end
311+
threads.each(&:join)
312+
313+
expect(node[:is_healthy]).to be(true).or be(false)
314+
expect(node[:last_access_timestamp]).to be_a(Integer)
315+
end
316+
end
317+
end
261318
end

0 commit comments

Comments
 (0)