feat: add allocated server readiness control

Game/scripts/match_net.gd

@@ -54,6 +54,7 @@ var local_player_name := "Player"
 # signed authorisation in hello rather than putting it in the endpoint URL.
 var join_authorisation := ""
 var require_join_authorisation := false
+var admissions_open := true
 var _allowed_join_authorisations: Dictionary = {}
 var _active_join_peers: Dictionary = {} # opaque authorisation -> peer_id
 var _join_history: Dictionary = {} # token -> {generation, lost_at}
@@ -98,6 +99,7 @@ func _on_shutting_down() -> void:
 	_join_authorisation_context.clear()
 	_join_signing_key = PackedByteArray()
 	require_join_authorisation = false
+	admissions_open = true
 
 
 func configure_join_authorisations(tokens: Array, context: Dictionary, signing_key: PackedByteArray = PackedByteArray()) -> bool:
@@ -122,6 +124,9 @@ func configure_join_authorisations(tokens: Array, context: Dictionary, signing_k
 func _on_peer_disconnected(peer_id: int) -> void:
 	if not multiplayer.is_server():
 		return
+	if not admissions_open:
+		await _reject(multiplayer.get_remote_sender_id(), "server is draining")
+		return
 	NetworkManager.invalidate_peer(peer_id)
 	_remove_player(peer_id)
 

Game/scripts/server_boot.gd

@@ -1,6 +1,7 @@
 extends Node
 
 const NetCodec = preload("res://scripts/net_codec.gd")
+const ServerControlScript = preload("res://scripts/server_control.gd")
 
 # Headless dedicated server entry point (task 1.6). Parses CLI args, hosts
 # via NetworkManager, logs structured lines, and watches for physics-tick
@@ -26,6 +27,8 @@ const NetCodec = preload("res://scripts/net_codec.gd")
 var _last_physics_frame := 0
 var config: ServerConfig = null
 var _watchdog_armed := false   # skip the first _process(): engine startup scheduling can batch several physics frames before the first idle frame runs, which isn't a real overrun
+var _control: ServerControl = null
+var _drain_requested := false
 
 
 func _ready() -> void:
@@ -77,6 +80,15 @@ func _ready() -> void:
 			printerr("cosmic-clash-server: refusing to start with invalid join-authorisations-file")
 			get_tree().quit(1)
 			return
+		_control = ServerControlScript.new()
+		_control.name = "ServerControl"
+		_control.drain_requested.connect(_on_drain_requested)
+		get_tree().root.add_child.call_deferred(_control)
+		var control_err := _control.start(int(config.get_value("readiness-port")), OS.get_environment(String(config.get_value("drain-token-env"))))
+		if control_err != OK:
+			printerr("cosmic-clash-server: refusing to start with invalid readiness control port")
+			get_tree().quit(1)
+			return
 
 	NetworkManager.client_connected.connect(_on_client_connected)
 	NetworkManager.client_disconnected.connect(_on_client_disconnected)
@@ -88,6 +100,8 @@ func _ready() -> void:
 		ServerLog.error("server_boot_failed", {"port": port, "error": error_string(err)})
 		get_tree().quit(1)
 		return
+	if _control != null:
+		_control.set_process_ready(true)
 	_install_match_loop()
 	ServerLog.info("server_started", {
 		"port": port, "max_clients": max_clients, "log_level": ServerLog.level_name(),
@@ -119,6 +133,10 @@ func _install_match_loop() -> void:
 
 func _process(_delta: float) -> void:
 	NetworkManager.poll()
+	if _drain_requested:
+		var scene := get_tree().current_scene
+		if not (is_instance_valid(scene) and scene.is_in_group("game")) and MatchNet.roster.is_empty():
+			get_tree().quit(0)
 	var current := Engine.get_physics_frames()
 	var steps := current - _last_physics_frame
 	_last_physics_frame = current
@@ -149,3 +167,9 @@ func _on_player_joined(peer_id: int, player_name: String) -> void:
 
 func _on_player_left(peer_id: int) -> void:
 	ServerLog.info("player_left", {"peer_id": peer_id, "roster": MatchNet.roster.size()})
+
+
+func _on_drain_requested() -> void:
+	_drain_requested = true
+	MatchNet.admissions_open = false
+	ServerLog.info("server_draining", {"reason": "control_request"})

Game/scripts/server_config.gd

@@ -77,6 +77,8 @@ static func specs() -> Array[Spec]:
 	out.append(Spec.new("region", Kind.STRING, "", "allocation", "Assigned region: EU or NA"))
 	out.append(Spec.new("join-authorisations-file", Kind.STRING, "", "allocation", "JSON array of control-plane signed join envelopes mounted for this match"))
 	out.append(Spec.new("join-authorisations-key-file", Kind.STRING, "", "allocation", "HMAC-SHA256 key file for verifying mounted join envelopes"))
+	out.append(Spec.new("readiness-port", Kind.INT, 7780, "allocation", "Loopback HTTP port for allocated process-ready and drain control"))
+	out.append(Spec.new("drain-token-env", Kind.STRING, "COSMIC_CLASH_DRAIN_TOKEN", "allocation", "Environment variable containing the allocated drain bearer token"))
 	return out
 
 
@@ -244,6 +246,9 @@ func _validate() -> void:
 	var port := int(values["port"])
 	if port < 1 or port > 65535:
 		errors.append("--port must be 1-65535, got %d" % port)
+	var readiness_port := int(values["readiness-port"])
+	if readiness_port < 1 or readiness_port > 65535:
+		errors.append("--readiness-port must be 1-65535, got %d" % readiness_port)
 	if int(values["max-clients"]) < 1:
 		errors.append("--max-clients must be at least 1, got %d" % int(values["max-clients"]))
 	if float(values["match-length"]) <= 0.0:

Game/scripts/server_control.gd

@@ -0,0 +1,106 @@
+class_name ServerControl
+extends Node
+
+# Small loopback HTTP control surface for allocated servers. The Go supervisor
+# uses GET /ready as the explicit process-ready probe and POST /drain during a
+# controlled termination. Direct/community servers do not start this node.
+
+signal drain_requested
+
+var _listener := TCPServer.new()
+var _peers: Array = []
+var _ready_for_connections := false
+var _draining := false
+var _drain_token := ""
+
+
+func start(port: int, drain_token: String = "") -> Error:
+	if port < 1 or port > 65535:
+		return ERR_INVALID_PARAMETER
+	_drain_token = drain_token
+	return _listener.listen(port, "127.0.0.1")
+
+
+func stop() -> void:
+	_listener.stop()
+	for peer in _peers:
+		if is_instance_valid(peer):
+			peer.disconnect_from_host()
+	_peers.clear()
+
+
+func set_process_ready(value: bool) -> void:
+	_ready_for_connections = value and not _draining
+
+
+func is_draining() -> bool:
+	return _draining
+
+
+func _exit_tree() -> void:
+	stop()
+
+
+func _process(_delta: float) -> void:
+	while _listener.is_connection_available():
+		_peers.append(_listener.take_connection())
+	for i in range(_peers.size() - 1, -1, -1):
+		var peer: StreamPeerTCP = _peers[i]
+		if peer.get_status() != StreamPeerTCP.STATUS_CONNECTED:
+			_peers.remove_at(i)
+			continue
+		var available := peer.get_available_bytes()
+		if available <= 0:
+			continue
+		var request := peer.get_utf8_string(available)
+		if "\r\n\r\n" not in request:
+			continue
+		_respond(peer, request)
+		_peers.remove_at(i)
+
+
+func _respond(peer: StreamPeerTCP, request: String) -> void:
+	var lines := request.split("\r\n")
+	var first := lines[0].split(" ") if not lines.is_empty() else PackedStringArray()
+	var method := String(first[0]) if first.size() > 0 else ""
+	var path := String(first[1]) if first.size() > 1 else ""
+	var status := 404
+	var reason := "Not Found"
+	var body := ""
+	if method == "GET" and path == "/ready":
+		status = 200 if _ready_for_connections else 503
+		reason = "OK" if status == 200 else "Service Unavailable"
+	elif method == "GET" and path == "/health":
+		status = 200
+		reason = "OK"
+	elif method == "POST" and path == "/drain":
+		var supplied := ""
+		for line in lines:
+			if line.begins_with("Authorization: Bearer "):
+				supplied = line.substr("Authorization: Bearer ".length())
+		if _drain_token.is_empty() or not _constant_time_equal(supplied, _drain_token):
+			status = 401
+			reason = "Unauthorized"
+		else:
+			_draining = true
+			_ready_for_connections = false
+			drain_requested.emit()
+			status = 202
+			reason = "Accepted"
+	else:
+		status = 405 if method in ["GET", "POST"] else 400
+		reason = "Method Not Allowed" if status == 405 else "Bad Request"
+	body = "{\"status\":\"%s\"}" % ("ready" if status == 200 else "not_ready")
+	var response := "HTTP/1.1 %d %s\r\nContent-Type: application/json\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s" % [status, reason, body.to_utf8_buffer().size(), body]
+	peer.put_data(response.to_utf8_buffer())
+	peer.disconnect_from_host()
+
+
+func _constant_time_equal(a: String, b: String) -> bool:
+	var left := a.to_utf8_buffer()
+	var right := b.to_utf8_buffer()
+	var difference := left.size() ^ right.size()
+	var length := mini(left.size(), right.size())
+	for i in length:
+		difference |= left[i] ^ right[i]
+	return difference == 0

Game/tests/cases/test_server_control.gd

@@ -0,0 +1,21 @@
+extends "res://tests/test_case.gd"
+
+const ServerControlScript = preload("res://scripts/server_control.gd")
+
+
+func test_control_rejects_invalid_port_and_starts_loopback_listener() -> void:
+	var control = ServerControlScript.new()
+	assert_eq(control.start(0), ERR_INVALID_PARAMETER, "control rejects port zero")
+	var port := 18000 + (Time.get_ticks_usec() % 1000)
+	assert_eq(control.start(port, "drain-secret"), OK, "control starts on a valid loopback port")
+	control.stop()
+	control.queue_free()
+
+
+func test_process_ready_and_drain_state_are_monotonic() -> void:
+	var control = ServerControlScript.new()
+	assert_true(not control.is_draining(), "control starts non-draining")
+	control.set_process_ready(true)
+	assert_true(not control.is_draining(), "process readiness does not imply draining")
+	control.stop()
+	control.queue_free()

Game/tests/server_control_smoke.gd

@@ -0,0 +1,51 @@
+extends SceneTree
+
+const ServerControlScript = preload("res://scripts/server_control.gd")
+const PORT := 18080
+
+
+func _init() -> void:
+	var control = ServerControlScript.new()
+	root.add_child(control)
+	if control.start(PORT, "drain-secret") != OK:
+		printerr("server control failed to bind")
+		quit(1)
+		return
+	control.set_process_ready(true)
+	await process_frame
+	var ready_response := await _request("GET", "/ready", [])
+	if ready_response != 200:
+		printerr("ready response was %d" % ready_response)
+		quit(1)
+		return
+	var unauthorized := await _request("POST", "/drain", ["Authorization: Bearer wrong"])
+	if unauthorized != 401:
+		printerr("unauthorized drain response was %d" % unauthorized)
+		quit(1)
+		return
+	var drained := await _request("POST", "/drain", ["Authorization: Bearer drain-secret"])
+	if drained != 202 or not control.is_draining():
+		printerr("authorized drain response/state was %d/%s" % [drained, control.is_draining()])
+		quit(1)
+		return
+	var not_ready := await _request("GET", "/ready", [])
+	if not_ready != 503:
+		printerr("draining ready response was %d" % not_ready)
+		quit(1)
+		return
+	control.stop()
+	print("server control smoke passed")
+	quit(0)
+
+
+func _request(method: String, path: String, headers: PackedStringArray) -> int:
+	var request := HTTPRequest.new()
+	root.add_child(request)
+	var http_method := HTTPClient.METHOD_GET if method == "GET" else HTTPClient.METHOD_POST
+	var err := request.request("http://127.0.0.1:%d%s" % [PORT, path], headers, http_method)
+	if err != OK:
+		request.queue_free()
+		return -1
+	var result = await request.request_completed
+	request.queue_free()
+	return int(result[1])