Takeshi
b84bade045
Preserve production code, regression fixtures, audio assets and provenance, test infrastructure, and the original-completion plan and progress ledger. Final game and Windows acceptance remain pending; excluded local artifacts and native experiments are not part of this snapshot.
415 lines
14 KiB
GDScript
415 lines
14 KiB
GDScript
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
|