class_name CommandQueue extends RefCounted const SNAPSHOT_VERSION := 2 const LEGACY_SNAPSHOT_VERSION := 1 var _pending: Array[SimulationCommand] = [] var _in_flight: Dictionary = {} var _known_command_ids: Dictionary = {} var _idempotency_to_command_id: Dictionary = {} var _processed_results: Dictionary = {} var _accepted_metadata: Dictionary = {} func enqueue(command: SimulationCommand) -> CommandResult: if command == null: return CommandResult.failed( CommandResult.Code.INVALID_COMMAND, 0, "Command is null." ) var validation_error := command.validation_error() if not validation_error.is_empty(): return CommandResult.failed( CommandResult.Code.INVALID_COMMAND, command.command_id, validation_error ) if _known_command_ids.has(command.command_id): return CommandResult.failed( CommandResult.Code.DUPLICATE_COMMAND_ID, command.command_id, "command_id has already been accepted.", {"original_command_id": command.command_id} ) if not command.idempotency_key.is_empty(): var key := command.idempotency_key if _idempotency_to_command_id.has(key): var original_id: int = _idempotency_to_command_id[key] var duplicate_details := { "idempotency_key": String(key), "original_command_id": original_id, "processed": _processed_results.has(original_id), } if _processed_results.has(original_id): var prior: CommandResult = _processed_results[original_id] duplicate_details["original_result"] = prior.to_dictionary() return CommandResult.failed( CommandResult.Code.DUPLICATE_IDEMPOTENCY_KEY, command.command_id, "idempotency_key has already been accepted.", duplicate_details ) var queued_command := command.copy() _pending.append(queued_command) _known_command_ids[queued_command.command_id] = true _accepted_metadata[queued_command.command_id] = { "command_id": queued_command.command_id, "idempotency_key": String(queued_command.idempotency_key), } if not queued_command.idempotency_key.is_empty(): _idempotency_to_command_id[queued_command.idempotency_key] = queued_command.command_id _pending.sort_custom(_comes_before) return CommandResult.succeeded(queued_command.command_id) func drain_ready(current_tick: int) -> Array[SimulationCommand]: var ready: Array[SimulationCommand] = [] while not _pending.is_empty(): var command := _pending[0] if command.execute_tick > current_tick: break _pending.pop_front() _in_flight[command.command_id] = command ready.append(command.copy()) return ready func execute_ready(current_tick: int, executor: Callable) -> Array[CommandResult]: var results: Array[CommandResult] = [] for command in drain_ready(current_tick): var accepted_command_id := command.command_id var result: CommandResult if not executor.is_valid(): result = CommandResult.failed( CommandResult.Code.EXECUTOR_FAILED, command.command_id, "Executor is not callable." ) else: var returned: Variant = executor.call(command) if returned is CommandResult: result = returned elif returned is bool and returned: result = CommandResult.succeeded(command.command_id) else: result = CommandResult.failed( CommandResult.Code.EXECUTOR_FAILED, command.command_id, "Executor did not return a successful CommandResult." ) var completed_result := complete(accepted_command_id, result) results.append(completed_result) return results func complete(command_id: int, result: CommandResult) -> CommandResult: if not _in_flight.has(command_id): return CommandResult.failed( CommandResult.Code.NOT_FOUND, command_id, "Command is not in flight." ) if result == null: return CommandResult.failed( CommandResult.Code.INVALID_COMMAND, command_id, "Result is null." ) var stored_result := result.copy() stored_result.command_id = command_id _processed_results[command_id] = stored_result _in_flight.erase(command_id) return stored_result.copy() func get_processed_result(command_id: int) -> CommandResult: if not _processed_results.has(command_id): return null var result: CommandResult = _processed_results[command_id] return result.copy() func get_result_by_idempotency_key(key: StringName) -> CommandResult: if not _idempotency_to_command_id.has(key): return null var command_id: int = _idempotency_to_command_id[key] if not _processed_results.has(command_id): return null var result: CommandResult = _processed_results[command_id] return result.replay_copy() func has_idempotency_key(key: StringName) -> bool: return _idempotency_to_command_id.has(key) func pending_count() -> int: return _pending.size() func in_flight_count() -> int: return _in_flight.size() func processed_count() -> int: return _processed_results.size() ## Read only pending commands, without serializing the processed history. func get_pending_commands() -> Array[SimulationCommand]: var result: Array[SimulationCommand] = [] for command: SimulationCommand in _pending: result.append(command.copy()) return result func get_accepted_command_metadata() -> Array[Dictionary]: var result: Array[Dictionary] = [] var command_ids: Array = _accepted_metadata.keys() command_ids.sort() for command_id: Variant in command_ids: result.append((_accepted_metadata[command_id] as Dictionary).duplicate(true)) return result func next_execute_tick() -> int: if _pending.is_empty(): return -1 return _pending[0].execute_tick func create_snapshot() -> Dictionary: var pending_data: Array[Dictionary] = [] for command in _pending: pending_data.append(command.to_dictionary()) var in_flight_ids: Array = _in_flight.keys() in_flight_ids.sort() var in_flight_data: Array[Dictionary] = [] for command_id: Variant in in_flight_ids: var command: SimulationCommand = _in_flight[command_id] in_flight_data.append(command.to_dictionary()) var processed_ids: Array = _processed_results.keys() processed_ids.sort() var processed_data: Array[Dictionary] = [] for command_id: Variant in processed_ids: var result: CommandResult = _processed_results[command_id] processed_data.append(result.to_dictionary()) var idempotency_data := {} var idempotency_keys: Array = _idempotency_to_command_id.keys() idempotency_keys.sort_custom(func(a: Variant, b: Variant) -> bool: return String(a) < String(b)) for key: Variant in idempotency_keys: idempotency_data[String(key)] = int(_idempotency_to_command_id[key]) return { "version": SNAPSHOT_VERSION, "pending": pending_data, "in_flight": in_flight_data, "processed": processed_data, "idempotency": idempotency_data, "accepted": get_accepted_command_metadata(), } func restore_snapshot(snapshot: Dictionary) -> bool: var version := LEGACY_SNAPSHOT_VERSION if snapshot.has("version"): if typeof(snapshot["version"]) != TYPE_INT: return false version = int(snapshot["version"]) if version not in [LEGACY_SNAPSHOT_VERSION, SNAPSHOT_VERSION]: return false if version == SNAPSHOT_VERSION and not _has_exact_keys( snapshot, ["version", "pending", "in_flight", "processed", "idempotency", "accepted"] ): return false for required_key in ["pending", "in_flight", "processed", "idempotency"]: if not snapshot.has(required_key): return false if not snapshot["pending"] is Array \ or not snapshot["in_flight"] is Array \ or not snapshot["processed"] is Array \ or not snapshot["idempotency"] is Dictionary \ or (version == SNAPSHOT_VERSION and not snapshot.get("accepted", null) is Array): return false var restored_pending: Array[SimulationCommand] = [] var restored_in_flight := {} var restored_known_ids := {} var restored_processed := {} var restored_idempotency := {} var command_keys_by_id := {} for command_data: Variant in snapshot["pending"]: if not command_data is Dictionary: return false if version == SNAPSHOT_VERSION and not _strict_command_dictionary(command_data): return false var command := SimulationCommand.from_dictionary(command_data) if not command.is_valid() or restored_known_ids.has(command.command_id): return false restored_pending.append(command) restored_known_ids[command.command_id] = true command_keys_by_id[command.command_id] = command.idempotency_key for command_data: Variant in snapshot["in_flight"]: if not command_data is Dictionary: return false if version == SNAPSHOT_VERSION and not _strict_command_dictionary(command_data): return false var command := SimulationCommand.from_dictionary(command_data) if not command.is_valid() or restored_known_ids.has(command.command_id): return false restored_in_flight[command.command_id] = command restored_known_ids[command.command_id] = true command_keys_by_id[command.command_id] = command.idempotency_key for result_data: Variant in snapshot["processed"]: if not result_data is Dictionary: return false if version == SNAPSHOT_VERSION and not _strict_result_dictionary(result_data): return false var result := CommandResult.from_dictionary(result_data) if result.command_id <= 0 or restored_known_ids.has(result.command_id): return false restored_processed[result.command_id] = result restored_known_ids[result.command_id] = true var mapped_key_by_id := {} for key: Variant in snapshot["idempotency"]: if typeof(key) not in [TYPE_STRING, TYPE_STRING_NAME] \ or StringName(key).is_empty() \ or typeof(snapshot["idempotency"][key]) != TYPE_INT: return false var mapped_command_id := int(snapshot["idempotency"][key]) if not restored_known_ids.has(mapped_command_id) \ or mapped_key_by_id.has(mapped_command_id): return false restored_idempotency[StringName(key)] = mapped_command_id mapped_key_by_id[mapped_command_id] = StringName(key) for command_id: Variant in command_keys_by_id: var command_key := StringName(command_keys_by_id[command_id]) if command_key.is_empty(): if mapped_key_by_id.has(command_id): return false elif mapped_key_by_id.get(command_id, &"") != command_key: return false var restored_metadata := {} if version == SNAPSHOT_VERSION: var previous_command_id := 0 for metadata_value: Variant in snapshot["accepted"]: if not metadata_value is Dictionary or not _has_exact_keys( metadata_value, ["command_id", "idempotency_key"] ): return false var command_id_value: Variant = metadata_value.get("command_id", null) var idempotency_value: Variant = metadata_value.get("idempotency_key", null) if typeof(command_id_value) != TYPE_INT \ or int(command_id_value) <= previous_command_id \ or typeof(idempotency_value) not in [TYPE_STRING, TYPE_STRING_NAME]: return false var command_id := int(command_id_value) var idempotency_key := StringName(idempotency_value) if not restored_known_ids.has(command_id) or restored_metadata.has(command_id): return false if command_keys_by_id.has(command_id) \ and StringName(command_keys_by_id[command_id]) != idempotency_key: return false if idempotency_key.is_empty(): if mapped_key_by_id.has(command_id): return false elif mapped_key_by_id.get(command_id, &"") != idempotency_key: return false restored_metadata[command_id] = { "command_id": command_id, "idempotency_key": String(idempotency_key), } previous_command_id = command_id if restored_metadata.size() != restored_known_ids.size(): return false else: # Legacy snapshots did not retain processed Command input. Their reverse # idempotency map still lets migration preserve the exactly-once key. var legacy_ids: Array = restored_known_ids.keys() legacy_ids.sort() for command_id: Variant in legacy_ids: var idempotency_key := StringName(command_keys_by_id.get( command_id, mapped_key_by_id.get(command_id, &"") )) restored_metadata[command_id] = { "command_id": int(command_id), "idempotency_key": String(idempotency_key), } restored_pending.sort_custom(_comes_before) _pending = restored_pending _in_flight = restored_in_flight _known_command_ids = restored_known_ids _processed_results = restored_processed _idempotency_to_command_id = restored_idempotency _accepted_metadata = restored_metadata return true func _strict_command_dictionary(data: Dictionary) -> bool: if not _has_exact_keys(data, [ "command_id", "issued_tick", "execute_tick", "priority", "command_type", "actor_id", "target_id", "payload", "idempotency_key", ]): return false for key: String in ["command_id", "issued_tick", "execute_tick", "priority"]: if typeof(data.get(key, null)) != TYPE_INT: return false for key: String in ["command_type", "actor_id", "target_id", "idempotency_key"]: if typeof(data.get(key, null)) not in [TYPE_STRING, TYPE_STRING_NAME]: return false return data.get("payload", null) is Dictionary func _strict_result_dictionary(data: Dictionary) -> bool: if not _has_exact_keys(data, [ "code", "command_id", "target_id", "message", "resolution_hints", "details", "is_replay", ]): return false var code_value: Variant = data.get("code", null) if typeof(code_value) not in [TYPE_STRING, TYPE_STRING_NAME] \ or CommandResult.Code.keys().find(String(code_value)) < 0 \ or typeof(data.get("command_id", null)) != TYPE_INT \ or typeof(data.get("target_id", null)) not in [TYPE_STRING, TYPE_STRING_NAME] \ or typeof(data.get("message", null)) != TYPE_STRING \ or not data.get("resolution_hints", null) is Array \ or not data.get("details", null) is Dictionary \ or typeof(data.get("is_replay", null)) != TYPE_BOOL: return false for hint: Variant in data["resolution_hints"]: if typeof(hint) != TYPE_STRING: return false return true func _has_exact_keys(source: Dictionary, expected: Array) -> bool: if source.size() != expected.size(): return false var seen: Dictionary = {} for raw_key: Variant in source: if typeof(raw_key) not in [TYPE_STRING, TYPE_STRING_NAME]: return false var key := String(raw_key) if key not in expected or seen.has(key): return false seen[key] = true return seen.size() == expected.size() func _comes_before(left: SimulationCommand, right: SimulationCommand) -> bool: if left.execute_tick != right.execute_tick: return left.execute_tick < right.execute_tick if left.priority != right.priority: return left.priority > right.priority return left.command_id < right.command_id