feat: Schemas

addons/netfox/encoder/diff-history-encoder.gd

@@ -3,6 +3,7 @@ class_name _DiffHistoryEncoder
 
 var _history: _PropertyHistoryBuffer
 var _property_cache: PropertyCache
+var _serializers: Dictionary
 
 var _full_snapshot := {}
 var _encoded_snapshot := {}
@@ -14,14 +15,15 @@ var _has_received := false
 
 static var _logger := NetfoxLogger._for_netfox("DiffHistoryEncoder")
 
-func _init(p_history: _PropertyHistoryBuffer, p_property_cache: PropertyCache):
+func _init(p_history: _PropertyHistoryBuffer, p_property_cache: PropertyCache, p_serializers: Dictionary) -> void:
 	_history = p_history
 	_property_cache = p_property_cache
+	_serializers = p_serializers
 
 func add_properties(properties: Array[PropertyEntry]) -> void:
 	var has_new_properties := false
 
-	for property_entry in properties:
+	for property_entry: PropertyEntry in properties:
 		var is_new := _ensure_property_idx(property_entry.to_string())
 		has_new_properties = has_new_properties or is_new
 
@@ -47,12 +49,19 @@ func encode(tick: int, reference_tick: int, properties: Array[PropertyEntry]) ->
 	var buffer := StreamPeerBuffer.new()
 	buffer.put_u8(_version)
 
-	for property in diff_snapshot.properties():
-		var property_idx := _property_indexes.get_by_value(property) as int
-		var property_value = diff_snapshot.get_value(property)
-
+	for property_path in diff_snapshot.properties():
+		var property_idx := _property_indexes.get_by_value(property_path) as int
 		buffer.put_u8(property_idx)
-		buffer.put_var(property_value)
+
+		var val = diff_snapshot.get_value(property_path)
+
+		if _serializers.has(property_path):
+			_serializers[property_path].encode(val, buffer)
+		else:
+			# Fallback
+			var data: PackedByteArray = var_to_bytes(val)
+			buffer.put_u32(data.size())
+			buffer.put_data(data)
 
 	return buffer.data_array
 
@@ -79,14 +88,24 @@ func decode(data: PackedByteArray, properties: Array[PropertyEntry]) -> _Propert
 	_has_received = true
 
 	while buffer.get_available_bytes() > 0:
+		# 1. Read Property Index
 		var property_idx := buffer.get_u8()
-		var property_value := buffer.get_var()
+
 		if not _property_indexes.has_key(property_idx):
 			_logger.warning("Received unknown property index %d, ignoring!", [property_idx])
-			continue
+			break
 
-		var property_entry := _property_indexes.get_by_key(property_idx)
-		result.set_value(property_entry, property_value)
+		var property_path := _property_indexes.get_by_key(property_idx)
+
+		var val
+		if _serializers.has(property_path):
+			val = _serializers[property_path].decode(buffer)
+		else:
+			var size: int = buffer.get_u32()
+			var bytes: Array = buffer.get_data(size)
+			val = bytes_to_var(bytes)
+
+		result.set_value(property_path, val)
 
 	return result
 

addons/netfox/encoder/redundant-history-encoder.gd

@@ -8,12 +8,18 @@ var redundancy: int = 4:
 var _history: _PropertyHistoryBuffer
 var _properties: Array[PropertyEntry]
 var _property_cache: PropertyCache
+var _serializers: Dictionary
 
 var _version := 0
 var _has_received := false
 
 var _logger := NetfoxLogger._for_netfox("RedundantHistoryEncoder")
 
+func _init(p_history: _PropertyHistoryBuffer, p_property_cache: PropertyCache, p_serializers: Dictionary):
+	_history = p_history
+	_property_cache = p_property_cache
+	_serializers = p_serializers
+
 func get_redundancy() -> int:
 	return redundancy
 
@@ -31,28 +37,48 @@ func set_properties(properties: Array[PropertyEntry]) -> void:
 		_version = (_version + 1) % 256
 		_properties = properties.duplicate()
 
-func encode(tick: int, properties: Array[PropertyEntry]) -> Array:
+func encode(tick: int, properties: Array[PropertyEntry]) -> PackedByteArray:
 	if _history.is_empty():
-		return []
-	var data := []
-
-	for i in range(mini(redundancy, _history.size())):
+		return PackedByteArray()
+
+	var buffer := StreamPeerBuffer.new()
+
+	buffer.put_u8(_version)
+
+	for i: int in range(mini(redundancy, _history.size())):
 		var offset_tick := tick - i
 		if offset_tick < _history.get_earliest_tick():
 			break
 
 		var snapshot := _history.get_snapshot(offset_tick)
-		for property in properties:
-			data.append(snapshot.get_value(property.to_string()))
-
-	data.append(_version)
-	return data
-
-func decode(data: Array, properties: Array[PropertyEntry]) -> Array[_PropertySnapshot]:
-	if data.is_empty() or properties.is_empty():
-		return []
+		for property: PropertyEntry in properties:
+			var path = property.to_string()
+			var value = snapshot.get_value(path)
+
+			if _serializers.has(path):
+				_serializers[path].encode(value, buffer)
+			else:
+				# Fallback
+				var data: PackedByteArray = var_to_bytes(value)
+				buffer.put_u32(data.size())
+				buffer.put_data(data)
+
+	return buffer.data_array
+
+func decode(data: Variant, properties: Array[PropertyEntry]) -> Array[_PropertySnapshot]:
+	var result: Array[_PropertySnapshot] = []
+
+	if data.is_empty():
+		return result
 
-	var packet_version := data.pop_back() as int
+	if data is Array:
+		return _decode_legacy_array(data, properties)
+
+	var buffer := StreamPeerBuffer.new()
+	if data is PackedByteArray:
+		buffer.data_array = data
+
+	var packet_version := buffer.get_u8()
 
 	if packet_version != _version:
 		if not _has_received:
@@ -61,22 +87,54 @@ func decode(data: Array, properties: Array[PropertyEntry]) -> Array[_PropertySna
 		else:
 			# Version mismatch, can't parse
 			_logger.warning("Version mismatch! own: %d, received: %s", [_version, packet_version])
-			return []
+			return result
+
+	_has_received = true
+
+	while buffer.get_available_bytes() > 0:
+		var snapshot = _PropertySnapshot.new()
+		var snapshot_valid = true
+
+		for property: PropertyEntry in properties:
+			# Stop if we run out of data mid-snapshot
+			if buffer.get_available_bytes() == 0:
+				snapshot_valid = false
+				break
+
+			var path: String = property.to_string()
+			var val
+
+			if _serializers.has(path):
+				val = _serializers[path].decode(buffer)
+			else:
+				var size: int = buffer.get_u32()
+				var bytes: Array = buffer.get_data(size)
+				# bytes is [error, data] from StreamPeerBuffer
+				val = bytes_to_var(bytes[1])
+
+			snapshot.set_value(path, val)
+
+		if snapshot_valid:
+			result.append(snapshot)
+		else:
+			break
+
+	return result
+
+func _decode_legacy_array(data: Array, properties: Array[PropertyEntry]) -> Array[_PropertySnapshot]:
+	if data.is_empty() or properties.is_empty(): return []
+	var packet_version = data.pop_back() as int
+	if packet_version != _version:
+		return []
 
 	var result: Array[_PropertySnapshot] = []
-	var redundancy := data.size() / properties.size()
-	result.assign(range(redundancy)
-		.map(func(__): return _PropertySnapshot.new())
-	)
+	var redundancy = data.size() / properties.size()
+	result.assign(range(redundancy).map(func(__): return _PropertySnapshot.new()))
 
 	for i in range(data.size()):
-		var offset_idx := i / properties.size()
-		var prop_idx := i % properties.size()
-
+		var offset_idx = i / properties.size()
+		var prop_idx = i % properties.size()
 		result[offset_idx].set_value(properties[prop_idx].to_string(), data[i])
-
-	_has_received = true
-
 	return result
 
 # Returns earliest new tick as int, or -1 if no new ticks applied
@@ -109,8 +167,3 @@ func apply(tick: int, snapshots: Array[_PropertySnapshot], sender: int = 0) -> i
 			earliest_new_tick = offset_tick
 
 	return earliest_new_tick
-
-
-func _init(p_history: _PropertyHistoryBuffer, p_property_cache: PropertyCache):
-	_history = p_history
-	_property_cache = p_property_cache

addons/netfox/encoder/snapshot-history-encoder.gd

@@ -4,53 +4,85 @@ class_name _SnapshotHistoryEncoder
 var _history: _PropertyHistoryBuffer
 var _property_cache: PropertyCache
 var _properties: Array[PropertyEntry]
+var _serializers: Dictionary
 
 var _version := -1
 var _has_received := false
 
 static var _logger := NetfoxLogger._for_netfox("_SnapshotHistoryEncoder")
 
-func _init(p_history: _PropertyHistoryBuffer, p_property_cache: PropertyCache):
+func _init(p_history: _PropertyHistoryBuffer, p_property_cache: PropertyCache, p_serializers: Dictionary) -> void:
 	_history = p_history
 	_property_cache = p_property_cache
+	_serializers = p_serializers
 
 func set_properties(properties: Array[PropertyEntry]) -> void:
 	if _properties != properties:
 		_version = (_version + 1) % 256
 		_properties = properties.duplicate()
 
-func encode(tick: int, properties: Array[PropertyEntry]) -> Array:
+func encode(tick: int, properties: Array[PropertyEntry]) -> PackedByteArray:
 	var snapshot := _history.get_snapshot(tick)
-	var data := []
-	data.resize(properties.size())
-
-	for i in range(properties.size()):
-		data[i] = snapshot.get_value(properties[i].to_string())
-	data.append(_version)
+	var buffer := StreamPeerBuffer.new()
+
+	buffer.put_u8(_version)
+
+	for property: PropertyEntry in properties:
+		var path: String = property.to_string()
+		var value = snapshot.get_value(path)
+
+		# Serializers are guaranteed to exist because of RollbackSynchronizer._compile_serializers
+		if _serializers.has(path):
+			_serializers[path].encode(value, buffer)
+		else:
+			# Safety fallback if something went wrong with compilation
+			var data: PackedByteArray = var_to_bytes(value)
+			buffer.put_u32(data.size())
+			buffer.put_data(data)
 
-	return data
+	return buffer.data_array
 
-func decode(data: Array, properties: Array[PropertyEntry]) -> _PropertySnapshot:
+func decode(data: Variant, properties: Array[PropertyEntry]) -> _PropertySnapshot:
 	var result := _PropertySnapshot.new()
-	var packet_version = data.pop_back()
+
+	# Handle case where data might be empty
+	if data.is_empty():
+		return result
+
+	var buffer := StreamPeerBuffer.new()
+	if data is PackedByteArray:
+		buffer.data_array = data
+	elif data is Array:
+		# Fallback
+		return _decode_legacy_array(data, properties)
+
+	# Read Version
+	var packet_version: int = buffer.get_u8()
 
 	if packet_version != _version:
 		if not _has_received:
-			# First packet, assume version is OK
 			_version = packet_version
 		else:
-			# Version mismatch, can't parse
 			_logger.warning("Version mismatch! own: %d, received: %s", [_version, packet_version])
 			return result
 
-	if properties.size() != data.size():
-		_logger.warning("Received snapshot with %d entries, with %d known - parsing as much as possible", [data.size(), properties.size()])
-
-	for i in range(0, mini(data.size(), properties.size())):
-		result.set_value(properties[i].to_string(), data[i])
-
 	_has_received = true
 
+	for property: PropertyEntry in properties:
+		if buffer.get_available_bytes() == 0:
+			break
+
+		var path: String = property.to_string()
+
+		if _serializers.has(path):
+			var val = _serializers[path].decode(buffer)
+			result.set_value(path, val)
+		else:
+			# Fallback decode
+			var size: int = buffer.get_u32()
+			var bytes: Array = buffer.get_data(size)
+			result.set_value(path, bytes_to_var(bytes))
+
 	return result
 
 func apply(tick: int, snapshot: _PropertySnapshot, sender: int = -1) -> bool:
@@ -65,3 +97,13 @@ func apply(tick: int, snapshot: _PropertySnapshot, sender: int = -1) -> bool:
 
 	_history.set_snapshot(tick, snapshot)
 	return true
+
+func _decode_legacy_array(data: Array, properties: Array[PropertyEntry]) -> _PropertySnapshot:
+	# Copied from original implementation just in case
+	var result := _PropertySnapshot.new()
+	var packet_version = data.pop_back()
+	if packet_version != _version:
+		return result
+	for i: int in range(0, mini(data.size(), properties.size())):
+		result.set_value(properties[i].to_string(), data[i])
+	return result

addons/netfox/rollback/composite/rollback-history-recorder.gd

@@ -17,7 +17,8 @@ func configure(
 		p_state_history: _PropertyHistoryBuffer, p_input_history: _PropertyHistoryBuffer,
 		p_state_property_config: _PropertyConfig, p_input_property_config: _PropertyConfig,
 		p_property_cache: PropertyCache,
-		p_skipset: _Set
+		p_skipset: _Set,
+		p_serializers: Dictionary
 	) -> void:
 	_state_history = p_state_history
 	_input_history = p_input_history