feat: persist regional chunks atomically
This commit is contained in:
Rijad Zuzo
@@ -0,0 +1,406 @@
|
||||
extends GutTest
|
||||
|
||||
const WORLD_ID := &"regional_world"
|
||||
const ORIGIN_LOCATION_ID := &"location_origin"
|
||||
const DESTINATION_LOCATION_ID := &"location_destination"
|
||||
const ORIGIN_SETTLEMENT_ID := &"settlement_origin"
|
||||
const DESTINATION_SETTLEMENT_ID := &"settlement_destination"
|
||||
const GROUP_ID := &"group_caravan"
|
||||
const ROUTE_ID := &"route_trade"
|
||||
const DEPARTURE_TICK := 10
|
||||
|
||||
var _test_directory := ""
|
||||
|
||||
|
||||
func before_each() -> void:
|
||||
var test_parent := (
|
||||
ProjectSettings.globalize_path("user://regional_chunked_file_store_tests").simplify_path()
|
||||
)
|
||||
_remove_tree(test_parent)
|
||||
_test_directory = (
|
||||
"user://regional_chunked_file_store_tests/run_%d_%d"
|
||||
% [Time.get_ticks_usec(), get_instance_id()]
|
||||
)
|
||||
|
||||
|
||||
func after_each() -> void:
|
||||
var absolute_path := ProjectSettings.globalize_path(_test_directory).simplify_path()
|
||||
var expected_parent := (
|
||||
ProjectSettings.globalize_path("user://regional_chunked_file_store_tests").simplify_path()
|
||||
)
|
||||
if absolute_path.begins_with(expected_parent + "/"):
|
||||
_remove_tree(absolute_path)
|
||||
|
||||
|
||||
func test_atomic_generations_round_trip_exact_archive_and_report_telemetry() -> void:
|
||||
var explicit_absolute_base := ProjectSettings.globalize_path(_test_directory)
|
||||
var store := RegionalChunkedFileStore.new(explicit_absolute_base)
|
||||
var first := _fixture_archive(false)
|
||||
var second := _fixture_archive(true)
|
||||
|
||||
assert_eq(store.get_base_directory(), explicit_absolute_base)
|
||||
assert_true(store.save(first))
|
||||
var first_generation := store.get_generation_ids()[0]
|
||||
assert_true(store.save(second))
|
||||
var generation_ids := store.get_generation_ids()
|
||||
assert_eq(generation_ids.size(), 2)
|
||||
assert_eq(generation_ids[0], first_generation)
|
||||
assert_true(
|
||||
(
|
||||
DirAccess.open(
|
||||
ProjectSettings.globalize_path(store.get_generation_directory(first_generation))
|
||||
)
|
||||
!= null
|
||||
)
|
||||
)
|
||||
var save_telemetry := store.get_last_telemetry()
|
||||
assert_true(save_telemetry["succeeded"])
|
||||
assert_gt(save_telemetry["serialized_bytes"], 0)
|
||||
assert_gt(save_telemetry["bytes_written"], 0)
|
||||
assert_gte(save_telemetry["write_usec"], 0)
|
||||
assert_gte(save_telemetry["validation_usec"], 0)
|
||||
|
||||
var latest_generation := generation_ids[-1]
|
||||
var generation_path := store.get_generation_directory(latest_generation)
|
||||
var generation_files := DirAccess.get_files_at(generation_path)
|
||||
assert_eq(generation_files.size(), second.get_chunk_ids().size() + 1)
|
||||
assert_true(RegionalChunkedFileStore.MANIFEST_FILE_NAME in generation_files)
|
||||
for file_name: String in generation_files:
|
||||
assert_false("/" in file_name)
|
||||
assert_false("\\" in file_name)
|
||||
assert_lte(file_name.length(), 64)
|
||||
var decoded: Variant = _read_variant(generation_path.path_join(file_name))
|
||||
assert_true(_contains_only_primitive_values(decoded))
|
||||
|
||||
var restored := store.load_archive()
|
||||
assert_not_null(restored)
|
||||
assert_eq(restored.checksum(), second.checksum())
|
||||
assert_eq(restored.restore_service().checksum(), second.restore_service().checksum())
|
||||
var load_telemetry := store.get_last_telemetry()
|
||||
assert_true(load_telemetry["succeeded"])
|
||||
assert_eq(load_telemetry["generation_id"], latest_generation)
|
||||
assert_false(load_telemetry["recovered_generation"])
|
||||
assert_gt(load_telemetry["bytes_read"], 0)
|
||||
assert_gte(load_telemetry["read_usec"], 0)
|
||||
assert_gte(load_telemetry["validation_usec"], 0)
|
||||
|
||||
|
||||
func test_interrupted_pointer_and_corrupt_latest_generation_recover_last_valid() -> void:
|
||||
var store := RegionalChunkedFileStore.new(_test_directory)
|
||||
var first := _fixture_archive(false)
|
||||
var second := _fixture_archive(true)
|
||||
assert_true(store.save(first))
|
||||
var first_generation := store.get_generation_ids()[0]
|
||||
|
||||
var current_pointer := ProjectSettings.globalize_path(store.get_current_pointer_path())
|
||||
var previous_pointer := ProjectSettings.globalize_path(
|
||||
_test_directory.path_join(RegionalChunkedFileStore.PREVIOUS_POINTER_FILE_NAME)
|
||||
)
|
||||
assert_eq(DirAccess.rename_absolute(current_pointer, previous_pointer), OK)
|
||||
var recovered_missing_pointer := store.load_archive()
|
||||
assert_not_null(recovered_missing_pointer)
|
||||
assert_eq(recovered_missing_pointer.checksum(), first.checksum())
|
||||
assert_true(store.get_last_telemetry()["recovered_generation"])
|
||||
|
||||
assert_true(store.save(second))
|
||||
var latest_generation := store.get_generation_ids()[-1]
|
||||
var latest_path := store.get_generation_directory(latest_generation)
|
||||
var manifest: Dictionary = _read_variant(
|
||||
latest_path.path_join(RegionalChunkedFileStore.MANIFEST_FILE_NAME)
|
||||
)
|
||||
var first_chunk_file := String(manifest["chunk_files"][0]["file_name"])
|
||||
_write_variant(latest_path.path_join(first_chunk_file), {"corrupt": true})
|
||||
var interrupted_path := ProjectSettings.globalize_path(
|
||||
store.get_generations_directory().path_join(
|
||||
"temporary_generation_00000000000000000003_interrupted"
|
||||
)
|
||||
)
|
||||
assert_eq(DirAccess.make_dir_recursive_absolute(interrupted_path), OK)
|
||||
_write_variant(interrupted_path.path_join("partial.bin"), {"partial": true})
|
||||
|
||||
var recovered := store.load_archive()
|
||||
assert_not_null(recovered)
|
||||
assert_eq(recovered.checksum(), first.checksum())
|
||||
assert_eq(store.get_last_telemetry()["generation_id"], first_generation)
|
||||
assert_true(store.get_last_telemetry()["recovered_generation"])
|
||||
assert_gte(store.get_last_telemetry()["candidates_validated"], 2)
|
||||
assert_true(FileAccess.file_exists(latest_path.path_join(first_chunk_file)))
|
||||
|
||||
|
||||
func test_partial_location_load_reads_only_selected_chunks_and_keeps_unloaded_authority() -> void:
|
||||
var store := RegionalChunkedFileStore.new(_test_directory)
|
||||
var archive := _fixture_archive(false)
|
||||
assert_true(store.save(archive))
|
||||
var saved_telemetry := store.get_last_telemetry()
|
||||
var generation_id := store.get_generation_ids()[0]
|
||||
var generation_path := store.get_generation_directory(generation_id)
|
||||
var manifest: Dictionary = _read_variant(
|
||||
generation_path.path_join(RegionalChunkedFileStore.MANIFEST_FILE_NAME)
|
||||
)
|
||||
var scheduler_entry := _find_chunk_file_entry(
|
||||
manifest, RegionalChunkedPersistence.SCHEDULER_CHUNK_ID
|
||||
)
|
||||
var scheduler_path := generation_path.path_join(String(scheduler_entry["file_name"]))
|
||||
_write_variant(scheduler_path, {"corrupt_unloaded_scheduler": true})
|
||||
|
||||
var partial := store.load_active_location_metadata(ORIGIN_LOCATION_ID)
|
||||
assert_false(partial.is_empty())
|
||||
assert_eq(partial["generation_id"], generation_id)
|
||||
assert_eq(
|
||||
partial["loaded_chunk_ids"],
|
||||
[
|
||||
RegionalChunkedPersistence.GLOBAL_CHUNK_ID,
|
||||
"location:%s" % ORIGIN_LOCATION_ID,
|
||||
]
|
||||
)
|
||||
assert_true(RegionalChunkedPersistence.SCHEDULER_CHUNK_ID in partial["unloaded_chunk_ids"])
|
||||
assert_eq(partial["mobile_group_refs"].size(), 1)
|
||||
assert_false(partial["full_generation_validated"])
|
||||
assert_true(FileAccess.file_exists(scheduler_path))
|
||||
var partial_telemetry := store.get_last_telemetry()
|
||||
assert_true(partial_telemetry["partial_load"])
|
||||
assert_eq(partial_telemetry["files_read"], 4)
|
||||
assert_lt(partial_telemetry["bytes_read"], saved_telemetry["bytes_written"])
|
||||
assert_null(store.load_archive())
|
||||
|
||||
|
||||
func test_unsafe_base_and_traversal_pointer_cannot_escape_generation_root() -> void:
|
||||
var archive := _fixture_archive(false)
|
||||
var unsafe_relative := RegionalChunkedFileStore.new("../regional_escape")
|
||||
assert_false(unsafe_relative.save(archive))
|
||||
assert_true("unsafe" in unsafe_relative.get_last_error())
|
||||
assert_false(RegionalChunkedFileStore.new("user://").save(archive))
|
||||
assert_false(RegionalChunkedFileStore.new("res://").save(archive))
|
||||
|
||||
var store := RegionalChunkedFileStore.new(_test_directory)
|
||||
assert_true(store.save(archive))
|
||||
assert_eq(store.get_generation_directory("../outside"), "")
|
||||
var malicious_pointer := {
|
||||
"schema_version": RegionalChunkedFileStore.SCHEMA_VERSION,
|
||||
"format_id": RegionalChunkedFileStore.FORMAT_ID,
|
||||
"generation_id": "../outside",
|
||||
"manifest_checksum": "0".repeat(64),
|
||||
"archive_checksum": "0".repeat(64),
|
||||
}
|
||||
_write_variant(store.get_current_pointer_path(), malicious_pointer)
|
||||
var recovered := store.load_archive()
|
||||
assert_not_null(recovered)
|
||||
assert_eq(recovered.checksum(), archive.checksum())
|
||||
assert_true(store.get_last_telemetry()["recovered_generation"])
|
||||
|
||||
|
||||
func test_explicit_conservative_prune_keeps_current_and_two_newest_valid_generations() -> void:
|
||||
var store := RegionalChunkedFileStore.new(_test_directory)
|
||||
var first := _fixture_archive(false)
|
||||
var latest := _fixture_archive(true)
|
||||
assert_true(store.save(first))
|
||||
assert_true(store.save(latest))
|
||||
assert_true(store.save(latest))
|
||||
var before := store.get_generation_ids()
|
||||
assert_eq(before.size(), 3)
|
||||
|
||||
assert_eq(store.prune_old_generations(2), 1)
|
||||
assert_eq(store.get_generation_ids(), [before[1], before[2]])
|
||||
assert_eq(store.get_last_telemetry()["operation"], "prune_old_generations")
|
||||
assert_eq(store.get_last_telemetry()["generations_pruned"], 1)
|
||||
assert_true(store.get_last_telemetry()["succeeded"])
|
||||
assert_gt(store.get_last_telemetry()["files_deleted"], 0)
|
||||
assert_eq(store.load_archive().checksum(), latest.checksum())
|
||||
|
||||
var retained := store.get_generation_ids()
|
||||
assert_eq(store.prune_old_generations(1), 0)
|
||||
assert_eq(store.get_generation_ids(), retained)
|
||||
assert_true("at least two" in store.get_last_error())
|
||||
|
||||
|
||||
func test_partial_load_does_not_resurrect_a_location_removed_by_newer_authority() -> void:
|
||||
var store := RegionalChunkedFileStore.new(_test_directory)
|
||||
var older_with_destination := _fixture_archive(false)
|
||||
var latest_without_destination := _single_location_archive()
|
||||
assert_true(store.save(older_with_destination))
|
||||
assert_true(store.save(latest_without_destination))
|
||||
|
||||
assert_true(store.load_active_location_metadata(DESTINATION_LOCATION_ID).is_empty())
|
||||
assert_true("does not contain" in store.get_last_error())
|
||||
assert_eq(store.load_archive().checksum(), latest_without_destination.checksum())
|
||||
|
||||
|
||||
func _fixture_archive(departed: bool) -> RegionalChunkedPersistence:
|
||||
var origin_location := LocationStateRecord.create(
|
||||
ORIGIN_LOCATION_ID, WORLD_ID, &"settlement", "Origin", _address(ORIGIN_LOCATION_ID)
|
||||
)
|
||||
var destination_location := LocationStateRecord.create(
|
||||
DESTINATION_LOCATION_ID,
|
||||
WORLD_ID,
|
||||
&"settlement",
|
||||
"Destination",
|
||||
_address(DESTINATION_LOCATION_ID)
|
||||
)
|
||||
var origin := SettlementStateRecord.create(
|
||||
ORIGIN_SETTLEMENT_ID,
|
||||
ORIGIN_LOCATION_ID,
|
||||
"Origin",
|
||||
&"polity_origin",
|
||||
1,
|
||||
&"founders_origin",
|
||||
0,
|
||||
{&"food": 20.0}
|
||||
)
|
||||
var destination := SettlementStateRecord.create(
|
||||
DESTINATION_SETTLEMENT_ID,
|
||||
DESTINATION_LOCATION_ID,
|
||||
"Destination",
|
||||
&"polity_destination",
|
||||
1,
|
||||
&"founders_destination",
|
||||
0,
|
||||
{&"food": 4.0}
|
||||
)
|
||||
var route := RouteStateRecord.create(ROUTE_ID, ORIGIN_LOCATION_ID, DESTINATION_LOCATION_ID, 5)
|
||||
var group := MobileGroupStateRecord.create_at_location(
|
||||
GROUP_ID,
|
||||
&"caravan",
|
||||
"Caravan",
|
||||
&"polity_origin",
|
||||
_address(ORIGIN_LOCATION_ID),
|
||||
[],
|
||||
{},
|
||||
{},
|
||||
10.0
|
||||
)
|
||||
var origin_polity := PolityStateRecord.create(
|
||||
&"polity_origin", "Origin polity", ORIGIN_LOCATION_ID, [ORIGIN_SETTLEMENT_ID]
|
||||
)
|
||||
var destination_polity := PolityStateRecord.create(
|
||||
&"polity_destination",
|
||||
"Destination polity",
|
||||
DESTINATION_LOCATION_ID,
|
||||
[DESTINATION_SETTLEMENT_ID]
|
||||
)
|
||||
var locations: Array[LocationStateRecord] = [destination_location, origin_location]
|
||||
var settlements: Array[SettlementStateRecord] = [destination, origin]
|
||||
var routes: Array[RouteStateRecord] = [route]
|
||||
var groups: Array[MobileGroupStateRecord] = [group]
|
||||
var persons: Array[PersonStateRecord] = []
|
||||
var cohorts: Array[PopulationCohortRecord] = []
|
||||
var polities: Array[PolityStateRecord] = [destination_polity, origin_polity]
|
||||
var relations: Array[DiplomaticRelationRecord] = []
|
||||
var world_state := RegionalWorldState.create(
|
||||
WORLD_ID, locations, settlements, routes, groups, persons, cohorts, polities, relations
|
||||
)
|
||||
var service := RegionalCaravanService.create(
|
||||
world_state, RegionalJobScheduler.new(), WorldEventStore.new()
|
||||
)
|
||||
if departed:
|
||||
assert_true(
|
||||
service.depart(
|
||||
GROUP_ID,
|
||||
ORIGIN_SETTLEMENT_ID,
|
||||
DESTINATION_SETTLEMENT_ID,
|
||||
ROUTE_ID,
|
||||
{&"food": 6.0},
|
||||
DEPARTURE_TICK
|
||||
)
|
||||
)
|
||||
return RegionalChunkedPersistence.capture(service, 2)
|
||||
|
||||
|
||||
func _single_location_archive() -> RegionalChunkedPersistence:
|
||||
var location := LocationStateRecord.create(
|
||||
ORIGIN_LOCATION_ID, WORLD_ID, &"settlement", "Origin", _address(ORIGIN_LOCATION_ID)
|
||||
)
|
||||
var settlement := SettlementStateRecord.create(
|
||||
ORIGIN_SETTLEMENT_ID,
|
||||
ORIGIN_LOCATION_ID,
|
||||
"Origin",
|
||||
&"polity_origin",
|
||||
1,
|
||||
&"founders_origin",
|
||||
0,
|
||||
{&"food": 20.0}
|
||||
)
|
||||
var polity := PolityStateRecord.create(
|
||||
&"polity_origin", "Origin polity", ORIGIN_LOCATION_ID, [ORIGIN_SETTLEMENT_ID]
|
||||
)
|
||||
var locations: Array[LocationStateRecord] = [location]
|
||||
var settlements: Array[SettlementStateRecord] = [settlement]
|
||||
var routes: Array[RouteStateRecord] = []
|
||||
var groups: Array[MobileGroupStateRecord] = []
|
||||
var persons: Array[PersonStateRecord] = []
|
||||
var cohorts: Array[PopulationCohortRecord] = []
|
||||
var polities: Array[PolityStateRecord] = [polity]
|
||||
var relations: Array[DiplomaticRelationRecord] = []
|
||||
var world_state := RegionalWorldState.create(
|
||||
WORLD_ID, locations, settlements, routes, groups, persons, cohorts, polities, relations
|
||||
)
|
||||
var service := RegionalCaravanService.create(
|
||||
world_state, RegionalJobScheduler.new(), WorldEventStore.new()
|
||||
)
|
||||
return RegionalChunkedPersistence.capture(service, 2)
|
||||
|
||||
|
||||
func _address(location_id: StringName) -> SpatialAddress:
|
||||
return SpatialAddress.create(WORLD_ID, location_id, Vector3.ZERO)
|
||||
|
||||
|
||||
func _find_chunk_file_entry(manifest: Dictionary, chunk_id: String) -> Dictionary:
|
||||
for raw_entry: Variant in manifest["chunk_files"]:
|
||||
if raw_entry is Dictionary and String(raw_entry.get("chunk_id", "")) == chunk_id:
|
||||
return raw_entry
|
||||
return {}
|
||||
|
||||
|
||||
func _read_variant(path: String) -> Variant:
|
||||
var file := FileAccess.open(path, FileAccess.READ)
|
||||
if file == null:
|
||||
return null
|
||||
var bytes := file.get_buffer(file.get_length())
|
||||
file.close()
|
||||
return bytes_to_var(bytes) if not bytes.is_empty() else null
|
||||
|
||||
|
||||
func _write_variant(path: String, value: Variant) -> void:
|
||||
var file := FileAccess.open(path, FileAccess.WRITE)
|
||||
assert_not_null(file)
|
||||
if file == null:
|
||||
return
|
||||
file.store_buffer(var_to_bytes(value))
|
||||
file.flush()
|
||||
file.close()
|
||||
|
||||
|
||||
func _contains_only_primitive_values(value: Variant) -> bool:
|
||||
match typeof(value):
|
||||
TYPE_NIL, TYPE_BOOL, TYPE_INT, TYPE_STRING:
|
||||
return true
|
||||
TYPE_FLOAT:
|
||||
return is_finite(float(value))
|
||||
TYPE_ARRAY:
|
||||
for item: Variant in value:
|
||||
if not _contains_only_primitive_values(item):
|
||||
return false
|
||||
return true
|
||||
TYPE_DICTIONARY:
|
||||
for key: Variant in value:
|
||||
if not key is String or not _contains_only_primitive_values(value[key]):
|
||||
return false
|
||||
return true
|
||||
return false
|
||||
|
||||
|
||||
func _remove_tree(absolute_path: String) -> void:
|
||||
var directory := DirAccess.open(absolute_path)
|
||||
if directory == null:
|
||||
return
|
||||
directory.include_hidden = true
|
||||
directory.list_dir_begin()
|
||||
var entry_name := directory.get_next()
|
||||
while not entry_name.is_empty():
|
||||
var entry_path := absolute_path.path_join(entry_name)
|
||||
if directory.current_is_dir():
|
||||
_remove_tree(entry_path)
|
||||
else:
|
||||
DirAccess.remove_absolute(entry_path)
|
||||
entry_name = directory.get_next()
|
||||
directory.list_dir_end()
|
||||
DirAccess.remove_absolute(absolute_path)
|
||||
@@ -0,0 +1 @@
|
||||
uid://cpesubt4wseva
|
||||
Reference in New Issue
Block a user