+4 -3

lib/servers/manager.ex

reviewed

··· 331 331 # Once complete, any future recoveries will have to start from these TLogs 332 332 :ok = Coordinator.write_generation(state.primary_coordinator, first_tlog_generation.generation, first_tlog_generation.start_version, first_tlog_generation.tlog_ids) 333 333 334 334 - state = put_in(state.cluster.tlog_generations, [first_tlog_generation]) 335 335 - 336 334 put_in(state.cluster.status, :normal) 337 335 end 338 336 ··· 389 387 new_tlog_generation = %TLogGeneration{generation: state.cluster.generation, start_version: last_generation_end_version + 1, tlog_ids: ids.tlog} 390 388 assert new_tlog_generation.start_version > prev_tlog_generation.start_version 391 389 392 392 - state = update_in(state.cluster.tlog_generations, &[new_tlog_generation | &1]) 390 390 + state = update_in(state.cluster.tlog_generations, fn [%TLogGeneration{} = last_gen | rest] -> 391 391 + last_gen = %{last_gen | end_version: last_generation_end_version} 392 392 + [new_tlog_generation, last_gen | rest] 393 393 + end) 393 394 394 395 # Recruit new TLogs 395 396 state = recruit_tlogs(state, ids.tlog, supervisors_tlogs, last_generation_end_version, meta_pairs)

+11 -19

lib/servers/storage.ex

reviewed

··· 462 462 defp peek_logs(%State{} = state) do 463 463 assert state.cluster.status == :normal 464 464 465 465 - # Get the preferred tlog for this storage server 466 466 - %TLogGeneration{} = generation = current_tlog_generation(state) 467 467 - 468 465 nonce = make_ref() 469 466 tag = state.id 470 467 start_version = state.data_version + 1 471 468 469 469 + %TLogGeneration{} = generation = tlog_generation_for_version(state.cluster.tlog_generations, start_version) 470 470 + end_version = generation.end_version 471 471 + 472 472 + if end_version != nil do 473 473 + assert is_integer(end_version) 474 474 + assert end_version >= start_version 475 475 + end 476 476 + 472 477 buddy_server = Map.get(state.cluster.servers, buddy_tlog_id(generation.tlog_ids, state.id)) 473 478 474 479 case state.buddy_tlog_failed or is_nil(buddy_server) do ··· 484 489 [pid] 485 490 end 486 491 |> Enum.each(fn pid -> 487 487 - TLog.peek(pid, nonce, tag, start_version) 492 492 + TLog.peek(pid, nonce, tag, start_version, end_version) 488 493 end) 489 494 490 495 SimServer.send_after self(), {:peek_timeout, nonce}, 300 ··· 494 499 defp on_peek_result(%State{} = state, %PeekResult{} = result) do 495 500 assert result.nonce == state.peek_nonce 496 501 497 497 - %TLogGeneration{} = generation = current_tlog_generation(state) 502 502 + # TODO: maybe better to store start_version on peek instead of getting it back? 503 503 + %TLogGeneration{} = generation = tlog_generation_for_version(state.cluster.tlog_generations, result.start_version) 498 504 assert result.tlog_id in generation.tlog_ids 499 505 500 506 case result.tlog_id == buddy_tlog_id(generation.tlog_ids, state.id) do ··· 524 530 defp apply_merged_peek_results(%State{} = state) do 525 531 results = state.peek_results |> Map.values() |> Enum.sort_by(&(&1.tlog_id)) 526 532 527 527 - assert length(results) >= (length(current_tlog_generation(state).tlog_ids) - 2) 528 528 - 529 533 Enum.each(results, fn %PeekResult{} = result -> 530 534 assert result.start_version == (state.data_version + 1) 531 535 assert result.end_version >= state.data_version ··· 542 546 state = apply_batches(state, batches) 543 547 SimServer.send_after self(), :peek_retry, 1 544 548 %{state | data_version: min_end_version, peek_nonce: nil} 545 545 - end 546 546 - 547 547 - defp current_tlog_generation(%State{} = state) do 548 548 - %TLogGeneration{} = generation = 549 549 - Enum.find(state.cluster.tlog_generations, fn %TLogGeneration{} = tlg -> 550 550 - tlg.start_version <= state.data_version 551 551 - end) 552 552 - 553 553 - assert generation.start_version <= state.data_version 554 554 - assert generation.end_version >= state.data_version 555 555 - 556 556 - generation 557 549 end 558 550 559 551 defp apply_batches(%State{} = state, batches) when is_list(batches) do

+11 -1

lib/utils.ex

reviewed

··· 3 3 Hobbes utils. 4 4 """ 5 5 6 6 - alias Hobbes.Structs.{Cluster, Server} 6 6 + alias Hobbes.Structs.{Cluster, TLogGeneration, Server} 7 7 alias Hobbes.Construct.SimServer 8 8 alias Hobbes.Encoding.Keyset 9 9 ··· 165 165 def buddy_tlog_id(tlog_ids, server_id) when is_list(tlog_ids) and is_integer(server_id) do 166 166 index = rem(server_id, length(tlog_ids)) 167 167 Enum.at(tlog_ids, index) 168 168 + end 169 169 + 170 170 + @spec tlog_generation_for_version([TLogGeneration.t], non_neg_integer) :: TLogGeneration.t 171 171 + def tlog_generation_for_version(generations, version) when is_list(generations) and is_integer(version) do 172 172 + case Enum.find(generations, fn %TLogGeneration{} = tlg -> tlg.start_version <= version end) do 173 173 + %TLogGeneration{} = generation -> 174 174 + generation 175 175 + 176 176 + nil -> raise "Could not find a generation for version #{inspect(version)} in #{inspect(generations)}" 177 177 + end 168 178 end 169 179 170 180 @doc """