From d5eb2b81f9d671ce808f81b14c7292178e85e2aa Mon Sep 17 00:00:00 2001 From: Rijad Zuzo Date: Wed, 12 Aug 2026 20:50:44 +0200 Subject: [PATCH] feat: add deterministic regional scheduler --- .../regional/AnalyticalRegionalUpdates.gd | 59 +++ .../regional/AnalyticalRegionalUpdates.gd.uid | 1 + simulation/regional/KeyedRandom.gd | 69 ++++ simulation/regional/KeyedRandom.gd.uid | 1 + simulation/regional/RegionalJobScheduler.gd | 349 ++++++++++++++++++ .../regional/RegionalJobScheduler.gd.uid | 1 + simulation/state/ScheduledJobRecord.gd | 216 +++++++++++ simulation/state/ScheduledJobRecord.gd.uid | 1 + tests/unit/test_regional_scheduler.gd | 213 +++++++++++ tests/unit/test_regional_scheduler.gd.uid | 1 + 10 files changed, 911 insertions(+) create mode 100644 simulation/regional/AnalyticalRegionalUpdates.gd create mode 100644 simulation/regional/AnalyticalRegionalUpdates.gd.uid create mode 100644 simulation/regional/KeyedRandom.gd create mode 100644 simulation/regional/KeyedRandom.gd.uid create mode 100644 simulation/regional/RegionalJobScheduler.gd create mode 100644 simulation/regional/RegionalJobScheduler.gd.uid create mode 100644 simulation/state/ScheduledJobRecord.gd create mode 100644 simulation/state/ScheduledJobRecord.gd.uid create mode 100644 tests/unit/test_regional_scheduler.gd create mode 100644 tests/unit/test_regional_scheduler.gd.uid diff --git a/simulation/regional/AnalyticalRegionalUpdates.gd b/simulation/regional/AnalyticalRegionalUpdates.gd new file mode 100644 index 0000000..72ae15c --- /dev/null +++ b/simulation/regional/AnalyticalRegionalUpdates.gd @@ -0,0 +1,59 @@ +class_name AnalyticalRegionalUpdates +extends RefCounted + + +static func linear_value( + current_value: float, + rate_per_tick: float, + elapsed_ticks: int, + minimum: float = -INF, + maximum: float = INF +) -> float: + if elapsed_ticks <= 0 or not is_finite(current_value) or not is_finite(rate_per_tick): + return clampf(current_value, minimum, maximum) + return clampf(current_value + rate_per_tick * elapsed_ticks, minimum, maximum) + + +static func exponential_value( + current_value: float, + multiplier_per_tick: float, + elapsed_ticks: int, + minimum: float = -INF, + maximum: float = INF +) -> float: + if ( + elapsed_ticks <= 0 + or not is_finite(current_value) + or not is_finite(multiplier_per_tick) + or multiplier_per_tick < 0.0 + ): + return clampf(current_value, minimum, maximum) + return clampf(current_value * pow(multiplier_per_tick, elapsed_ticks), minimum, maximum) + + +static func periodic_occurrences( + first_due_tick: int, interval_ticks: int, through_tick: int +) -> int: + if first_due_tick < 0 or interval_ticks <= 0 or through_tick < first_due_tick: + return 0 + return (through_tick - first_due_tick) / interval_ticks + 1 + + +static func transfer_amount( + available: float, + demand: float, + rate_per_tick: float, + elapsed_ticks: int, + loss_fraction: float = 0.0 +) -> Dictionary: + var sent := minf( + maxf(available, 0.0), + minf(maxf(demand, 0.0), maxf(rate_per_tick, 0.0) * maxi(elapsed_ticks, 0)) + ) + var received := sent * (1.0 - clampf(loss_fraction, 0.0, 1.0)) + return { + "sent": sent, + "received": received, + "remaining_available": maxf(available - sent, 0.0), + "remaining_demand": maxf(demand - received, 0.0), + } diff --git a/simulation/regional/AnalyticalRegionalUpdates.gd.uid b/simulation/regional/AnalyticalRegionalUpdates.gd.uid new file mode 100644 index 0000000..8e02a05 --- /dev/null +++ b/simulation/regional/AnalyticalRegionalUpdates.gd.uid @@ -0,0 +1 @@ +uid://bw56tsrd8qtgr diff --git a/simulation/regional/KeyedRandom.gd b/simulation/regional/KeyedRandom.gd new file mode 100644 index 0000000..4e0d541 --- /dev/null +++ b/simulation/regional/KeyedRandom.gd @@ -0,0 +1,69 @@ +class_name KeyedRandom +extends RefCounted + +const MASK_32 := 0xFFFFFFFF +const UINT32_RANGE := 4294967296.0 + + +static func value_u32( + world_seed: int, system_id: StringName, entity_id: StringName, occurrence: int +) -> int: + var hash_value := int(world_seed) & MASK_32 + hash_value = _mix_u32(hash_value ^ 0x9E3779B9) + for byte in String(system_id).to_utf8_buffer(): + hash_value = _mix_u32(hash_value ^ int(byte)) + hash_value = _mix_u32(hash_value ^ 0x85EBCA6B) + for byte in String(entity_id).to_utf8_buffer(): + hash_value = _mix_u32(hash_value ^ int(byte)) + hash_value = _mix_u32(hash_value ^ (occurrence & MASK_32)) + return hash_value + + +static func value_float( + world_seed: int, system_id: StringName, entity_id: StringName, occurrence: int +) -> float: + return float(value_u32(world_seed, system_id, entity_id, occurrence)) / UINT32_RANGE + + +static func range_int( + world_seed: int, + system_id: StringName, + entity_id: StringName, + occurrence: int, + minimum: int, + maximum: int +) -> int: + if maximum < minimum: + return minimum + var width := maximum - minimum + 1 + return minimum + value_u32(world_seed, system_id, entity_id, occurrence) % width + + +static func range_float( + world_seed: int, + system_id: StringName, + entity_id: StringName, + occurrence: int, + minimum: float, + maximum: float +) -> float: + if maximum <= minimum: + return minimum + return lerpf(minimum, maximum, value_float(world_seed, system_id, entity_id, occurrence)) + + +static func chance( + world_seed: int, + system_id: StringName, + entity_id: StringName, + occurrence: int, + probability: float +) -> bool: + return value_float(world_seed, system_id, entity_id, occurrence) < clampf(probability, 0.0, 1.0) + + +static func _mix_u32(value: int) -> int: + var mixed := value & MASK_32 + mixed = ((mixed ^ (mixed >> 16)) * 0x7FEB352D) & MASK_32 + mixed = ((mixed ^ (mixed >> 15)) * 0x846CA68B) & MASK_32 + return (mixed ^ (mixed >> 16)) & MASK_32 diff --git a/simulation/regional/KeyedRandom.gd.uid b/simulation/regional/KeyedRandom.gd.uid new file mode 100644 index 0000000..cd12740 --- /dev/null +++ b/simulation/regional/KeyedRandom.gd.uid @@ -0,0 +1 @@ +uid://c00kbf23ey8na diff --git a/simulation/regional/RegionalJobScheduler.gd b/simulation/regional/RegionalJobScheduler.gd new file mode 100644 index 0000000..59a085e --- /dev/null +++ b/simulation/regional/RegionalJobScheduler.gd @@ -0,0 +1,349 @@ +class_name RegionalJobScheduler +extends RefCounted + +const SCHEMA_VERSION := 1 +const UNLIMITED_BUDGET := -1 + +var _jobs_by_id: Dictionary = {} +var _job_heap: Array[ScheduledJobRecord] = [] +var _heap_index_by_id: Dictionary = {} +var _dedupe_job_ids: Dictionary = {} +var _sequence_job_ids: Dictionary = {} +var _next_stable_sequence := 0 +var _last_drain_tick := -1 +var _executed_count := 0 +var _last_executed_job: ScheduledJobRecord + + +static func from_dictionary(record_data: Dictionary) -> RegionalJobScheduler: + var scheduler := RegionalJobScheduler.new() + return scheduler if scheduler.restore_from_dictionary(record_data) else null + + +func size() -> int: + return _jobs_by_id.size() + + +func is_empty() -> bool: + return _jobs_by_id.is_empty() + + +func has_job(job_id: StringName) -> bool: + return _jobs_by_id.has(job_id) + + +func get_by_id(job_id: StringName) -> ScheduledJobRecord: + return _jobs_by_id.get(job_id) as ScheduledJobRecord + + +func schedule(job: ScheduledJobRecord) -> bool: + if ( + job == null + or not job.is_valid() + or has_job(job.get_job_id()) + or _sequence_job_ids.has(job.get_stable_sequence()) + or (_last_executed_job != null and not ScheduledJobRecord.compare(_last_executed_job, job)) + ): + return false + var dedupe_key := job.get_dedupe_key() + if not dedupe_key.is_empty() and _dedupe_job_ids.has(dedupe_key): + return false + var stored := ScheduledJobRecord.from_dictionary(job.to_dictionary()) + if stored == null: + return false + _jobs_by_id[stored.get_job_id()] = stored + _sequence_job_ids[stored.get_stable_sequence()] = stored.get_job_id() + if not dedupe_key.is_empty(): + _dedupe_job_ids[dedupe_key] = stored.get_job_id() + _heap_push(stored) + _next_stable_sequence = maxi(_next_stable_sequence, stored.get_stable_sequence() + 1) + return true + + +func schedule_new( + job_id: StringName, + job_type: StringName, + due_tick: int, + phase: int, + entity_id: StringName, + payload: Dictionary = {}, + dedupe_key: StringName = &"", + repeat_interval: int = ScheduledJobRecord.NO_REPEAT +) -> ScheduledJobRecord: + var job := ScheduledJobRecord.create( + job_id, + job_type, + due_tick, + phase, + entity_id, + _next_stable_sequence, + payload, + dedupe_key, + repeat_interval + ) + return job if schedule(job) else null + + +func cancel(job_id: StringName) -> ScheduledJobRecord: + var job := get_by_id(job_id) + if job == null: + return null + var remove_index := int(_heap_index_by_id[job_id]) + var last_index := _job_heap.size() - 1 + if remove_index != last_index: + _heap_swap(remove_index, last_index) + _job_heap.pop_back() + _heap_index_by_id.erase(job_id) + _jobs_by_id.erase(job_id) + _sequence_job_ids.erase(job.get_stable_sequence()) + var dedupe_key := job.get_dedupe_key() + if not dedupe_key.is_empty(): + _dedupe_job_ids.erase(dedupe_key) + if remove_index < _job_heap.size(): + var parent := (remove_index - 1) / 2 + if ( + remove_index > 0 + and ScheduledJobRecord.compare(_job_heap[remove_index], _job_heap[parent]) + ): + _heap_sift_up(remove_index) + else: + _heap_sift_down(remove_index) + return job + + +func peek_next() -> ScheduledJobRecord: + return _job_heap[0] if not _job_heap.is_empty() else null + + +func get_all_sorted() -> Array[ScheduledJobRecord]: + var jobs := _job_heap.duplicate() + jobs.sort_custom(ScheduledJobRecord.compare) + return jobs + + +func get_due_count(current_tick: int) -> int: + var count := 0 + for job in _jobs_by_id.values(): + if (job as ScheduledJobRecord).get_due_tick() <= current_tick: + count += 1 + return count + + +func take_due( + current_tick: int, execution_budget: int = UNLIMITED_BUDGET +) -> Array[ScheduledJobRecord]: + var due: Array[ScheduledJobRecord] = [] + if current_tick < 0 or current_tick < _last_drain_tick or execution_budget < UNLIMITED_BUDGET: + return due + _last_drain_tick = current_tick + if execution_budget == 0: + return due + var remaining := execution_budget + while not _job_heap.is_empty(): + var next := _job_heap[0] + if next.get_due_tick() > current_tick: + break + if remaining == 0: + break + var job := cancel(next.get_job_id()) + due.append(job) + _executed_count += 1 + _last_executed_job = job + if job.is_repeating(): + var next_occurrence := job.create_next_occurrence(_next_stable_sequence) + if not schedule(next_occurrence): + push_error("Could not schedule next occurrence for '%s'" % job.get_job_id()) + if remaining > 0: + remaining -= 1 + return due + + +func execute_due( + current_tick: int, execution_budget: int, executor: Callable +) -> Array[ScheduledJobRecord]: + var executed: Array[ScheduledJobRecord] = [] + if not executor.is_valid() or execution_budget < UNLIMITED_BUDGET: + return executed + var remaining := execution_budget + while remaining != 0: + var batch := take_due(current_tick, 1) + if batch.is_empty(): + break + var job := batch[0] + executed.append(job) + executor.call(job) + if remaining > 0: + remaining -= 1 + return executed + + +func restore(records: Array[ScheduledJobRecord], cursor: Dictionary = {}) -> bool: + var candidate := RegionalJobScheduler.new() + for record in records: + if not candidate.schedule(record): + return false + var next_sequence := int(cursor.get("next_stable_sequence", candidate._next_stable_sequence)) + var last_drain_tick := int(cursor.get("last_drain_tick", -1)) + var executed_count := int(cursor.get("executed_count", 0)) + var saved_last_order: Variant = cursor.get("last_executed_order") + var last_executed_job := _parse_cursor_order(saved_last_order) + if ( + next_sequence < candidate._next_stable_sequence + or last_drain_tick < -1 + or executed_count < 0 + or (executed_count == 0 and last_executed_job != null) + or (executed_count > 0 and last_executed_job == null) + or (saved_last_order != null and last_executed_job == null) + or ( + last_executed_job != null + and ( + last_executed_job.get_due_tick() > last_drain_tick + or next_sequence <= last_executed_job.get_stable_sequence() + ) + ) + ): + return false + if ( + last_executed_job != null + and not candidate._job_heap.is_empty() + and not ScheduledJobRecord.compare(last_executed_job, candidate._job_heap[0]) + ): + return false + _adopt(candidate) + _next_stable_sequence = next_sequence + _last_drain_tick = last_drain_tick + _executed_count = executed_count + _last_executed_job = last_executed_job + return true + + +func restore_from_dictionary(record_data: Dictionary) -> bool: + if int(record_data.get("schema_version", -1)) != SCHEMA_VERSION: + return false + if not record_data.has_all(["jobs", "cursor"]) or not record_data["jobs"] is Array: + return false + if not record_data["cursor"] is Dictionary: + return false + if not ( + record_data["cursor"] + . has_all( + [ + "next_stable_sequence", + "last_drain_tick", + "executed_count", + "last_executed_order", + ] + ) + ): + return false + var records: Array[ScheduledJobRecord] = [] + for saved_job in record_data["jobs"]: + if not saved_job is Dictionary: + return false + var job := ScheduledJobRecord.from_dictionary(saved_job) + if job == null: + return false + records.append(job) + return restore(records, record_data["cursor"]) + + +func get_cursor() -> Dictionary: + return { + "next_stable_sequence": _next_stable_sequence, + "last_drain_tick": _last_drain_tick, + "executed_count": _executed_count, + "last_executed_order": _serialize_cursor_order(_last_executed_job), + } + + +func to_dictionary() -> Dictionary: + var jobs: Array[Dictionary] = [] + for job in get_all_sorted(): + jobs.append(job.to_dictionary()) + return { + "schema_version": SCHEMA_VERSION, + "jobs": jobs, + "cursor": get_cursor(), + } + + +func _heap_push(job: ScheduledJobRecord) -> void: + var index := _job_heap.size() + _job_heap.append(job) + _heap_index_by_id[job.get_job_id()] = index + _heap_sift_up(index) + + +func _heap_sift_up(start_index: int) -> void: + var index := start_index + while index > 0: + var parent := (index - 1) / 2 + if not ScheduledJobRecord.compare(_job_heap[index], _job_heap[parent]): + break + _heap_swap(index, parent) + index = parent + + +func _heap_sift_down(start_index: int) -> void: + var index := start_index + while true: + var left := index * 2 + 1 + if left >= _job_heap.size(): + return + var right := left + 1 + var smallest := left + if ( + right < _job_heap.size() + and ScheduledJobRecord.compare(_job_heap[right], _job_heap[left]) + ): + smallest = right + if not ScheduledJobRecord.compare(_job_heap[smallest], _job_heap[index]): + return + _heap_swap(index, smallest) + index = smallest + + +func _heap_swap(first_index: int, second_index: int) -> void: + var first := _job_heap[first_index] + _job_heap[first_index] = _job_heap[second_index] + _job_heap[second_index] = first + _heap_index_by_id[_job_heap[first_index].get_job_id()] = first_index + _heap_index_by_id[_job_heap[second_index].get_job_id()] = second_index + + +func _adopt(source: RegionalJobScheduler) -> void: + _jobs_by_id = source._jobs_by_id + _job_heap = source._job_heap + _heap_index_by_id = source._heap_index_by_id + _dedupe_job_ids = source._dedupe_job_ids + _sequence_job_ids = source._sequence_job_ids + _next_stable_sequence = source._next_stable_sequence + + +static func _serialize_cursor_order(job: ScheduledJobRecord) -> Variant: + if job == null: + return null + return { + "due_tick": job.get_due_tick(), + "phase": job.get_phase(), + "entity_id": String(job.get_entity_id()), + "stable_sequence": job.get_stable_sequence(), + } + + +static func _parse_cursor_order(value: Variant) -> ScheduledJobRecord: + if value == null: + return null + if ( + not value is Dictionary + or not value.has_all(["due_tick", "phase", "entity_id", "stable_sequence"]) + ): + return null + return ScheduledJobRecord.create( + &"__cursor__", + &"__cursor__", + int(value["due_tick"]), + int(value["phase"]), + StringName(value["entity_id"]), + int(value["stable_sequence"]) + ) diff --git a/simulation/regional/RegionalJobScheduler.gd.uid b/simulation/regional/RegionalJobScheduler.gd.uid new file mode 100644 index 0000000..ffbf9bb --- /dev/null +++ b/simulation/regional/RegionalJobScheduler.gd.uid @@ -0,0 +1 @@ +uid://d03xv4blyvbr6 diff --git a/simulation/state/ScheduledJobRecord.gd b/simulation/state/ScheduledJobRecord.gd new file mode 100644 index 0000000..910c732 --- /dev/null +++ b/simulation/state/ScheduledJobRecord.gd @@ -0,0 +1,216 @@ +class_name ScheduledJobRecord +extends RefCounted + +const SCHEMA_VERSION := 1 +const NO_REPEAT := 0 + +var data: Dictionary + + +func _init(record_data: Dictionary = {}) -> void: + data = record_data.duplicate(true) + + +static func create( + job_id: StringName, + job_type: StringName, + due_tick: int, + phase: int, + entity_id: StringName, + stable_sequence: int, + payload: Dictionary = {}, + dedupe_key: StringName = &"", + repeat_interval: int = NO_REPEAT, + occurrence: int = 0 +) -> ScheduledJobRecord: + var payload_status := [true] + var normalized_payload: Variant = _normalize_serializable(payload, payload_status) + if not payload_status[0] or not normalized_payload is Dictionary: + return null + var record := ( + ScheduledJobRecord + . new( + { + "schema_version": SCHEMA_VERSION, + "job_id": String(job_id), + "job_type": String(job_type), + "due_tick": due_tick, + "phase": phase, + "entity_id": String(entity_id), + "stable_sequence": stable_sequence, + "dedupe_key": String(dedupe_key), + "repeat_interval": repeat_interval, + "occurrence": occurrence, + "payload": normalized_payload, + } + ) + ) + return record if record.is_valid() else null + + +static func from_dictionary(record_data: Dictionary) -> ScheduledJobRecord: + if int(record_data.get("schema_version", -1)) != SCHEMA_VERSION: + return null + if not ( + record_data + . has_all( + [ + "job_id", + "job_type", + "due_tick", + "phase", + "entity_id", + "stable_sequence", + "dedupe_key", + "repeat_interval", + "occurrence", + "payload", + ] + ) + ): + return null + if not record_data["payload"] is Dictionary: + return null + return create( + StringName(record_data["job_id"]), + StringName(record_data["job_type"]), + int(record_data["due_tick"]), + int(record_data["phase"]), + StringName(record_data["entity_id"]), + int(record_data["stable_sequence"]), + record_data["payload"], + StringName(record_data["dedupe_key"]), + int(record_data["repeat_interval"]), + int(record_data["occurrence"]) + ) + + +func is_valid() -> bool: + return ( + not get_job_id().is_empty() + and not get_job_type().is_empty() + and get_due_tick() >= 0 + and get_phase() >= 0 + and not get_entity_id().is_empty() + and get_stable_sequence() >= 0 + and get_repeat_interval() >= NO_REPEAT + and get_occurrence() >= 0 + and data["payload"] is Dictionary + ) + + +func get_job_id() -> StringName: + return StringName(data["job_id"]) + + +func get_job_type() -> StringName: + return StringName(data["job_type"]) + + +func get_due_tick() -> int: + return int(data["due_tick"]) + + +func get_phase() -> int: + return int(data["phase"]) + + +func get_entity_id() -> StringName: + return StringName(data["entity_id"]) + + +func get_stable_sequence() -> int: + return int(data["stable_sequence"]) + + +func get_dedupe_key() -> StringName: + return StringName(data["dedupe_key"]) + + +func get_repeat_interval() -> int: + return int(data["repeat_interval"]) + + +func get_occurrence() -> int: + return int(data["occurrence"]) + + +func get_payload() -> Dictionary: + return data["payload"].duplicate(true) + + +func is_repeating() -> bool: + return get_repeat_interval() > NO_REPEAT + + +func create_next_occurrence(next_stable_sequence: int = -1) -> ScheduledJobRecord: + if not is_repeating(): + return null + var sequence := get_stable_sequence() if next_stable_sequence < 0 else next_stable_sequence + var next_occurrence := get_occurrence() + 1 + return create( + &"%s#%d" % [get_job_id().get_slice("#", 0), next_occurrence], + get_job_type(), + get_due_tick() + get_repeat_interval(), + get_phase(), + get_entity_id(), + sequence, + get_payload(), + get_dedupe_key(), + get_repeat_interval(), + next_occurrence + ) + + +func to_dictionary() -> Dictionary: + return data.duplicate(true) + + +static func compare(first: ScheduledJobRecord, second: ScheduledJobRecord) -> bool: + if first.get_due_tick() != second.get_due_tick(): + return first.get_due_tick() < second.get_due_tick() + if first.get_phase() != second.get_phase(): + return first.get_phase() < second.get_phase() + if first.get_entity_id() != second.get_entity_id(): + return String(first.get_entity_id()) < String(second.get_entity_id()) + return first.get_stable_sequence() < second.get_stable_sequence() + + +static func _normalize_serializable(value: Variant, status: Array) -> Variant: + match typeof(value): + TYPE_NIL, TYPE_BOOL, TYPE_INT, TYPE_STRING: + return value + TYPE_STRING_NAME: + return String(value) + TYPE_FLOAT: + if not is_finite(float(value)): + status[0] = false + return value + TYPE_ARRAY: + var normalized_array: Array = [] + for item in value: + normalized_array.append(_normalize_serializable(item, status)) + return normalized_array + TYPE_DICTIONARY: + return _normalize_dictionary(value, status) + status[0] = false + return null + + +static func _normalize_dictionary(value: Dictionary, status: Array) -> Dictionary: + var source_keys: Dictionary = {} + for raw_key in value: + if not raw_key is String and not raw_key is StringName: + status[0] = false + return {} + var key := String(raw_key) + if source_keys.has(key): + status[0] = false + return {} + source_keys[key] = raw_key + var keys: Array = source_keys.keys() + keys.sort() + var normalized: Dictionary = {} + for key in keys: + normalized[key] = _normalize_serializable(value[source_keys[key]], status) + return normalized diff --git a/simulation/state/ScheduledJobRecord.gd.uid b/simulation/state/ScheduledJobRecord.gd.uid new file mode 100644 index 0000000..751c63a --- /dev/null +++ b/simulation/state/ScheduledJobRecord.gd.uid @@ -0,0 +1 @@ +uid://dnvfgloigjgmy diff --git a/tests/unit/test_regional_scheduler.gd b/tests/unit/test_regional_scheduler.gd new file mode 100644 index 0000000..37d17c1 --- /dev/null +++ b/tests/unit/test_regional_scheduler.gd @@ -0,0 +1,213 @@ +extends GutTest + + +func test_explicit_job_order_is_due_phase_entity_sequence() -> void: + var scheduler := RegionalJobScheduler.new() + assert_true(scheduler.schedule(ScheduledJobRecord.create(&"z", &"work", 8, 0, &"b", 2))) + assert_true(scheduler.schedule(ScheduledJobRecord.create(&"d", &"work", 7, 2, &"a", 3))) + assert_true(scheduler.schedule(ScheduledJobRecord.create(&"c", &"work", 7, 1, &"z", 1))) + assert_true(scheduler.schedule(ScheduledJobRecord.create(&"b", &"work", 7, 1, &"a", 9))) + assert_true(scheduler.schedule(ScheduledJobRecord.create(&"a", &"work", 7, 1, &"a", 4))) + + assert_eq(_job_ids(scheduler.get_all_sorted()), [&"a", &"b", &"c", &"d", &"z"]) + assert_eq(scheduler.get_due_count(6), 0) + assert_eq(scheduler.get_due_count(7), 4) + assert_eq(_job_ids(scheduler.take_due(7, 2)), [&"a", &"b"]) + assert_eq(_job_ids(scheduler.take_due(7)), [&"c", &"d"]) + assert_eq(_job_ids(scheduler.take_due(8)), [&"z"]) + assert_true(scheduler.is_empty()) + + +func test_budgets_one_eight_and_unlimited_preserve_order_and_checksum() -> void: + var budget_one := _run_fixture(1) + var budget_eight := _run_fixture(8) + var unlimited := _run_fixture(RegionalJobScheduler.UNLIMITED_BUDGET) + + assert_eq(budget_one["order"], budget_eight["order"]) + assert_eq(budget_one["order"], unlimited["order"]) + assert_eq(budget_one["checksum"], budget_eight["checksum"]) + assert_eq(budget_one["checksum"], unlimited["checksum"]) + assert_eq(budget_one["order"].size(), 37) + assert_eq(budget_one["remaining"], 0) + assert_eq(budget_eight["remaining"], 0) + assert_eq(unlimited["remaining"], 0) + + +func test_dynamic_jobs_interleave_identically_across_execution_budgets() -> void: + var budget_one := _run_dynamic_fixture(1) + var budget_eight := _run_dynamic_fixture(8) + var unlimited := _run_dynamic_fixture(RegionalJobScheduler.UNLIMITED_BUDGET) + + assert_eq(budget_one, [&"parent", &"child", &"tail"]) + assert_eq(budget_eight, budget_one) + assert_eq(unlimited, budget_one) + + +func test_restore_mid_due_queue_continues_without_drop_or_reorder() -> void: + var uninterrupted := _fixture_scheduler() + var expected: Array[StringName] = [] + while not uninterrupted.is_empty(): + expected.append_array(_job_ids(uninterrupted.take_due(100, 3))) + + var interrupted := _fixture_scheduler() + var actual := _job_ids(interrupted.take_due(100, 5)) + var saved := interrupted.to_dictionary() + var restored := RegionalJobScheduler.from_dictionary(saved) + + assert_not_null(restored) + assert_eq(restored.to_dictionary(), saved) + while not restored.is_empty(): + actual.append_array(_job_ids(restored.take_due(100, 2))) + assert_eq(actual, expected) + assert_eq(actual.size(), 37) + assert_eq(actual.duplicate().reduce(_unique_id_count, {} as Dictionary).size(), 37) + assert_eq(restored.get_cursor()["executed_count"], 37) + + +func test_schedule_cancel_dedupe_and_repeating_jobs_preserve_queue_contract() -> void: + var scheduler := RegionalJobScheduler.new() + var repeating := scheduler.schedule_new( + &"weather#0", &"weather", 2, 0, &"jajce", {}, &"weather:jajce", 3 + ) + assert_not_null(repeating) + assert_null( + scheduler.schedule_new(&"duplicate", &"weather", 2, 0, &"jajce", {}, &"weather:jajce", 3) + ) + assert_eq(_job_ids(scheduler.take_due(2, 1)), [&"weather#0"]) + assert_true(scheduler.has_job(&"weather#1")) + assert_eq(scheduler.get_by_id(&"weather#1").get_due_tick(), 5) + assert_eq(scheduler.get_by_id(&"weather#1").get_occurrence(), 1) + assert_eq(_job_ids(scheduler.take_due(100, 2)), [&"weather#1", &"weather#2"]) + assert_true(scheduler.has_job(&"weather#3")) + assert_not_null(scheduler.cancel(&"weather#3")) + assert_true(scheduler.is_empty()) + assert_null(scheduler.cancel(&"missing")) + var duplicate_sequence := ScheduledJobRecord.create(&"one", &"work", 9, 0, &"a", 40) + assert_true(scheduler.schedule(duplicate_sequence)) + assert_false(scheduler.schedule(ScheduledJobRecord.create(&"two", &"work", 10, 0, &"b", 40))) + + +func test_keyed_random_is_order_independent_and_key_sensitive() -> void: + var first := KeyedRandom.value_u32(1337, &"caravan", &"caravan_a", 4) + var unrelated := KeyedRandom.value_u32(1337, &"weather", &"jajce", 99) + var repeated := KeyedRandom.value_u32(1337, &"caravan", &"caravan_a", 4) + + assert_eq(first, repeated) + assert_ne(first, unrelated) + assert_ne(first, KeyedRandom.value_u32(1338, &"caravan", &"caravan_a", 4)) + assert_ne(first, KeyedRandom.value_u32(1337, &"caravan", &"caravan_b", 4)) + assert_ne(first, KeyedRandom.value_u32(1337, &"caravan", &"caravan_a", 5)) + assert_true(KeyedRandom.value_float(1337, &"caravan", &"a", 1) >= 0.0) + assert_true(KeyedRandom.value_float(1337, &"caravan", &"a", 1) < 1.0) + assert_true(KeyedRandom.range_int(1337, &"caravan", &"a", 1, 3, 7) in range(3, 8)) + assert_true(KeyedRandom.chance(1337, &"caravan", &"a", 1, 1.0)) + assert_false(KeyedRandom.chance(1337, &"caravan", &"a", 1, 0.0)) + + +func test_analytical_elapsed_tick_updates_match_stepwise_results() -> void: + var stepwise_linear := 4.0 + for _tick in range(17): + stepwise_linear = clampf(stepwise_linear + 0.75, 0.0, 20.0) + assert_almost_eq( + AnalyticalRegionalUpdates.linear_value(4.0, 0.75, 17, 0.0, 20.0), stepwise_linear, 0.000001 + ) + + var stepwise_exponential := 100.0 + for _tick in range(24): + stepwise_exponential *= 0.98 + assert_almost_eq( + AnalyticalRegionalUpdates.exponential_value(100.0, 0.98, 24), stepwise_exponential, 0.000001 + ) + assert_eq(AnalyticalRegionalUpdates.periodic_occurrences(5, 3, 4), 0) + assert_eq(AnalyticalRegionalUpdates.periodic_occurrences(5, 3, 14), 4) + var transfer := AnalyticalRegionalUpdates.transfer_amount(12.0, 9.0, 2.0, 3, 0.25) + assert_eq(transfer["sent"], 6.0) + assert_eq(transfer["received"], 4.5) + assert_eq(transfer["remaining_available"], 6.0) + assert_eq(transfer["remaining_demand"], 4.5) + + +func test_record_and_scheduler_parsers_reject_invalid_state_transactionally() -> void: + var valid := ScheduledJobRecord.create( + &"job", &"work", 3, 0, &"entity", 1, {&"z": &"value", &"a": {&"b": 2, &"a": 1}} + ) + assert_eq( + ScheduledJobRecord.from_dictionary(valid.to_dictionary()).to_dictionary(), + valid.to_dictionary() + ) + assert_eq(valid.get_payload(), {"a": {"a": 1, "b": 2}, "z": "value"}) + var invalid := valid.to_dictionary() + invalid["due_tick"] = -1 + assert_null(ScheduledJobRecord.from_dictionary(invalid)) + + var scheduler := RegionalJobScheduler.new() + assert_true(scheduler.schedule(valid)) + var before := scheduler.to_dictionary() + var malformed := before.duplicate(true) + malformed["jobs"].append(before["jobs"][0].duplicate(true)) + assert_false(scheduler.restore_from_dictionary(malformed)) + assert_eq(scheduler.to_dictionary(), before) + assert_eq(_job_ids(scheduler.take_due(3, 1)), [&"job"]) + assert_false(scheduler.schedule(ScheduledJobRecord.create(&"past", &"work", 2, 0, &"a", 2))) + + +func _run_fixture(budget: int) -> Dictionary: + var scheduler := _fixture_scheduler() + var order: Array[StringName] = [] + var checksum := 146959810 + while not scheduler.is_empty(): + var batch := scheduler.take_due(100, budget) + assert_false(batch.is_empty()) + for job in batch: + order.append(job.get_job_id()) + checksum = int((checksum * 16777619 + job.get_stable_sequence()) & 0x7FFFFFFF) + return {"order": order, "checksum": checksum, "remaining": scheduler.size()} + + +func _run_dynamic_fixture(budget: int) -> Array[StringName]: + var scheduler := RegionalJobScheduler.new() + assert_not_null(scheduler.schedule_new(&"parent", &"work", 5, 0, &"a")) + assert_not_null(scheduler.schedule_new(&"tail", &"work", 5, 2, &"z")) + var order: Array[StringName] = [] + while not scheduler.is_empty(): + var executed := scheduler.execute_due( + 5, + budget, + func(job: ScheduledJobRecord) -> void: + order.append(job.get_job_id()) + if job.get_job_id() == &"parent": + assert_not_null(scheduler.schedule_new(&"child", &"work", 5, 1, &"b")) + ) + assert_false(executed.is_empty()) + return order + + +func _fixture_scheduler() -> RegionalJobScheduler: + var scheduler := RegionalJobScheduler.new() + for index in range(37): + var due_tick := 5 + (index * 7) % 11 + var phase := (index * 5) % 3 + var entity_id := StringName("entity_%02d" % ((index * 13) % 9)) + var job := ScheduledJobRecord.create( + StringName("job_%02d" % index), + &"fixture", + due_tick, + phase, + entity_id, + index, + {"delta": index + 1} + ) + assert_true(scheduler.schedule(job)) + return scheduler + + +func _job_ids(jobs: Array[ScheduledJobRecord]) -> Array[StringName]: + var result: Array[StringName] = [] + for job in jobs: + result.append(job.get_job_id()) + return result + + +func _unique_id_count(accumulator: Dictionary, job_id: StringName) -> Dictionary: + accumulator[job_id] = true + return accumulator diff --git a/tests/unit/test_regional_scheduler.gd.uid b/tests/unit/test_regional_scheduler.gd.uid new file mode 100644 index 0000000..35ad9bb --- /dev/null +++ b/tests/unit/test_regional_scheduler.gd.uid @@ -0,0 +1 @@ +uid://dm8ovvj2dw025