From 2a7c8ec81b2a97c64fbfd07738a6678ca876ea67 Mon Sep 17 00:00:00 2001 From: Andrew Barnes Date: Thu, 6 Aug 2026 15:29:37 -0400 Subject: [PATCH] Harden multiplayer session terminal handling --- src/link/Net.lua | 11 +- src/link/Session.lua | 119 +++++++++++++++++-- tests/engine/link_session.lua | 218 ++++++++++++++++++++++++++++++++++ 3 files changed, 330 insertions(+), 18 deletions(-) diff --git a/src/link/Net.lua b/src/link/Net.lua index 950bee1e..b8d4654c 100644 --- a/src/link/Net.lua +++ b/src/link/Net.lua @@ -207,7 +207,7 @@ function Net:send(msg) end if self.peerEnd then -- loopback: re-encode through json like the wire local decoded = Json.decode(Json.encode(msg)) - if decoded and not self.peerEnd.closed then + if decoded ~= nil and not self.peerEnd.closed then table.insert(self.peerEnd.inbox, decoded) end return @@ -256,13 +256,12 @@ end function Net:handleTCPLine(line) local msg = Json.decode(line) - if not msg then + if msg == nil then Logger.warn("link: bad relay message %q", line:sub(1, 60)) return end - if not handleGenericRelayControl(self, msg) then - table.insert(self.inbox, msg) - end + if type(msg) == "table" and handleGenericRelayControl(self, msg) then return end + table.insert(self.inbox, msg) end -- pulls every complete "\n"-terminated line out of rxBuf (leaving a @@ -352,7 +351,7 @@ function Net:update() end elseif event.type == "receive" then local msg = Json.decode(event.data) - if msg then + if msg ~= nil then table.insert(self.inbox, msg) else Logger.warn("link: bad message %q", tostring(event.data):sub(1, 60)) diff --git a/src/link/Session.lua b/src/link/Session.lua index 68bd02b7..ecb4a64b 100644 --- a/src/link/Session.lua +++ b/src/link/Session.lua @@ -48,13 +48,43 @@ function Session:_refreshStatus() self._status = self.paired and "paired" or "connecting" self.closed = false self.error = nil + return end + if #self._inbox > 0 then + self._status = "draining" + self.closed = false + self.error = nil + return + end + self._status = self._terminal.status + self.closed = true + self.error = self._terminal.status == "failed" + and (self._terminal.detail or self._terminal.reason) or nil +end + +function Session:_latchTerminal(status, reason, detail) + if self._terminal then return false end + self._terminal = { status = status, reason = reason, detail = detail } + self:_refreshStatus() + return true +end + +function Session:_closeTransport() + if self._transportCloseCalled then return true end + self._transportCloseCalled = true + local ok, detail = pcall(self._transport.close, self._transport) + return ok, ok and nil or tostring(detail) end function Session:getRole() return self._role end function Session:getKind() return self._kind end function Session:getStatus() return self._status end -function Session:getFailure() return nil, nil end +function Session:getFailure() + if not self._terminal or self._terminal.status ~= "failed" then + return nil, nil + end + return self._terminal.reason, self._terminal.detail +end function Session:hasPending() return #self._inbox > 0 end function Session:send(message) @@ -63,13 +93,62 @@ function Session:send(message) end function Session:update() - if self._terminal then return end - self._transport:update() - self:_syncMetadata() - local messages = self._transport:poll() - for _, message in ipairs(messages) do - self._inbox[#self._inbox + 1] = message + if self._terminal then + self:_refreshStatus() + return end + + local failureReason, failureDetail + local updateOk, updateDetail = pcall(self._transport.update, self._transport) + self:_syncMetadata() + if not updateOk then + failureReason, failureDetail = "transport_error", tostring(updateDetail) + elseif self._transport.error then + failureReason = "transport_error" + failureDetail = tostring(self._transport.error) + end + + local pollOk, messages = pcall(self._transport.poll, self._transport) + if not pollOk then + if not failureReason then + failureReason, failureDetail = "transport_error", tostring(messages) + end + elseif type(messages) ~= "table" then + if not failureReason then + failureReason, failureDetail = "transport_error", + "transport poll returned non-table" + end + else + for index = 1, #messages do + local message = messages[index] + if type(message) ~= "table" or type(message.type) ~= "string" then + if not failureReason then + failureReason = "protocol_error" + failureDetail = ("message %d must be a table with string type") + :format(index) + end + break + end + self._inbox[#self._inbox + 1] = message + end + end + + self:_syncMetadata() + if failureReason then + self:_latchTerminal("failed", failureReason, failureDetail) + self:_closeTransport() + elseif self._transport.closed then + local closeOk, closeDetail = self:_closeTransport() + if closeOk then + self:_latchTerminal("closed") + else + self:_latchTerminal("failed", "transport_error", closeDetail) + end + end + self:_refreshStatus() +end + +local function finishRead(self) self:_refreshStatus() end @@ -77,7 +156,9 @@ function Session:take(messageType) assert(type(messageType) == "string", "Session.take requires a message type") for index, message in ipairs(self._inbox) do if message.type == messageType then - return table.remove(self._inbox, index) + local found = table.remove(self._inbox, index) + finishRead(self) + return found end end return nil @@ -85,19 +166,33 @@ end function Session:pollOne() if #self._inbox == 0 then return nil end - return table.remove(self._inbox, 1) + local message = table.remove(self._inbox, 1) + finishRead(self) + return message end function Session:poll() local messages = self._inbox self._inbox = {} + finishRead(self) return messages end function Session:close() - if self._transportCloseCalled then return end - self._transportCloseCalled = true - self._transport:close() + if self._status == "closed" or self._status == "failed" then return end + if self._terminal then + self:_closeTransport() + self:_refreshStatus() + return + end + local ok, detail = self:_closeTransport() + if ok then + self:_latchTerminal("closed") + else + self:_latchTerminal("failed", "transport_error", detail) + end + self:_syncMetadata() + self:_refreshStatus() end return Session diff --git a/tests/engine/link_session.lua b/tests/engine/link_session.lua index 1ab3313e..9ec9d36b 100644 --- a/tests/engine/link_session.lua +++ b/tests/engine/link_session.lua @@ -2,6 +2,7 @@ package.path = "./?.lua;./?/init.lua;" .. package.path local T = require("tests.modkit") local Net = require("src.link.Net") +local Json = require("src.link.Json") local Session = require("src.link.Session") local function sessionPair() @@ -10,6 +11,37 @@ local function sessionPair() Session.new(guestNet, { role = "guest", kind = "link" }) end +local function fakeTransport(options) + options = options or {} + local transport = { + paired = options.paired ~= false, + closed = false, + error = nil, + inbox = options.inbox or {}, + closeCount = 0, + } + function transport:update() + if options.onUpdate then options.onUpdate(self) end + if options.updateError then error(options.updateError) end + end + function transport:poll() + if options.pollError then error(options.pollError) end + local messages = self.inbox + self.inbox = {} + return messages + end + function transport:send(message) + self.sent = message + return true + end + function transport:close() + self.closeCount = self.closeCount + 1 + self.closed = true + if options.closeError then error(options.closeError) end + end + return transport +end + do local host, guest = sessionPair() T.eq(host:getRole(), "host", "host role is assigned locally") @@ -81,4 +113,190 @@ do T.eq(session.code, "ROOM02", "update refreshes relay metadata") end +do + local transport = fakeTransport({ onUpdate = function(self) + self.inbox[#self.inbox + 1] = { type = "bye", final = true } + self.closed = true + end }) + local session = Session.new(transport, { role = "host", kind = "link" }) + session:update() + T.eq(session:getStatus(), "draining", "normal close drains its final packet") + T.eq(session.closed, false, "compatibility closed waits for the FIFO") + T.eq(session:take("bye").final, true, "final close packet remains observable") + T.eq(session:getStatus(), "closed", "normal drain reaches closed") + T.eq(transport.closeCount, 1, "transport cleanup runs once") +end + +do + local transport = fakeTransport({ + onUpdate = function(self) self.closed = true end, + closeError = "normal cleanup exploded", + }) + local session = Session.new(transport, { role = "host", kind = "link" }) + T.check(pcall(session.update, session), + "normal-close cleanup exception does not escape the game loop") + local reason, detail = session:getFailure() + T.eq(reason, "transport_error", + "normal-close cleanup exception becomes a transport failure") + T.check(detail:find("normal cleanup exploded", 1, true) ~= nil, + "normal-close cleanup failure keeps its diagnostic detail") + T.eq(session:getStatus(), "failed", + "normal-close cleanup exception cannot report a clean close") +end + +do + local transport = fakeTransport({ onUpdate = function(self) + self.inbox = { { type = "before", sequence = 1 } } + self.error = "socket failed" + self.closed = true + end }) + local session = Session.new(transport, { role = "guest", kind = "link" }) + session:update() + local reason, detail = session:getFailure() + T.eq(reason, "transport_error", "transport failure has a stable reason") + T.eq(detail, "socket failed", "transport failure retains original detail") + T.eq(session:getStatus(), "draining", "transport failure drains valid prefix") + T.eq(session.error, nil, "legacy error stays hidden during drain") + T.eq(session.closed, false, "legacy closed stays false during failed drain") + T.eq(session:pollOne().sequence, 1, "failed drain returns its valid prefix") + T.eq(session:getStatus(), "failed", "failed drain reaches failed") + T.eq(session.error, "socket failed", "legacy error appears at terminal failure") + transport.error = "later error" + session:update() + local _, latchedDetail = session:getFailure() + T.eq(latchedDetail, "socket failed", "first terminal failure stays latched") +end + +do + local transport = fakeTransport({ inbox = { + { type = "before", sequence = 1 }, + false, + { type = "after", sequence = 3 }, + } }) + local session = Session.new(transport, { role = "host", kind = "link" }) + session:update() + local reason = session:getFailure() + T.eq(reason, "protocol_error", "malformed packet fails as protocol_error") + T.eq(session:getStatus(), "draining", "malformed batch drains valid prefix") + local messages = session:poll() + T.eq(#messages, 1, "malformed value and untrusted tail are not exposed") + T.eq(messages[1].sequence, 1, "valid prefix survives malformed packet") + T.eq(session:getStatus(), "failed", "protocol drain reaches failed") +end + +do + local transport = fakeTransport({ inbox = { { type = 7 } } }) + local session = Session.new(transport, { role = "host", kind = "link" }) + session:update() + T.eq(session:getFailure(), "protocol_error", + "table without string type is a protocol error") +end + +do + local transport = fakeTransport({ + inbox = { { type = "future_world_packet", value = 9 } }, + }) + local session = Session.new(transport, { role = "host", kind = "link" }) + session:update() + T.eq(session:pollOne().value, 9, "unknown typed packet stays mode-owned") +end + +do + local transport = fakeTransport({ + inbox = { { type = "already_decoded", value = 4 } }, + updateError = "update exploded", + }) + local session = Session.new(transport, { role = "host", kind = "link" }) + local ok = pcall(session.update, session) + T.check(ok, "transport update exception does not escape the game loop") + T.eq(session:getStatus(), "draining", "update exception still drains prior inbox") + T.eq(session:pollOne().value, 4, "decoded packet survives update exception") + T.eq(session:getStatus(), "failed", "update exception becomes terminal failure") +end + +do + local transport = fakeTransport({ pollError = "poll exploded" }) + local session = Session.new(transport, { role = "host", kind = "link" }) + T.check(pcall(session.update, session), + "transport poll exception does not escape the game loop") + local reason, detail = session:getFailure() + T.eq(reason, "transport_error", "poll exception is a transport failure") + T.check(detail:find("poll exploded", 1, true) ~= nil, + "poll exception keeps its diagnostic detail") +end + +do + local transport = fakeTransport({ closeError = "close exploded" }) + local session = Session.new(transport, { role = "guest", kind = "link" }) + T.check(pcall(session.close, session), + "transport close exception does not escape cleanup") + T.eq(session:getFailure(), "transport_error", + "close exception is a transport failure") + session:close() + T.eq(transport.closeCount, 1, "failed close is still attempted only once") +end + +do + local transport = fakeTransport() + local session = Session.new(transport, { role = "guest", kind = "link" }) + session:close() + session:close() + session:update() + T.eq(transport.closeCount, 1, "close and post-terminal update are idempotent") + T.eq(session:getStatus(), "closed", "explicit close reaches closed") +end + +do + T.check(not pcall(Session.new, nil, { role = "host", kind = "link" }), + "constructor rejects missing transport") + local transport = fakeTransport() + T.check(not pcall(Session.new, transport, { role = "leader", kind = "link" }), + "constructor rejects unsupported role") + T.check(not pcall(Session.new, transport, { role = "host", kind = "" }), + "constructor rejects empty kind") +end + +do + local senderNet, receiverNet = Net.loopbackPair() + local receiver = Session.new(receiverNet, { role = "guest", kind = "link" }) + senderNet:send(false) + receiver:update() + T.eq(receiver:getFailure(), "protocol_error", + "loopback forwards decoded false to session validation") +end + +do + local delivered = false + local transport = Net.new() + transport.enetHost = { + service = function() + if delivered then return nil end + delivered = true + return { type = "receive", data = "false" } + end, + } + local session = Session.new(transport, { role = "guest", kind = "link" }) + session:update() + T.eq(session:getFailure(), "protocol_error", + "ENet forwards decoded false to session validation") +end + +do + local transport = Net.new() + local session = Session.new(transport, { role = "host", kind = "tournament" }) + T.check(pcall(transport.handleTCPLine, transport, "42"), + "TCP control handoff does not index a decoded scalar") + session:update() + T.eq(session:getFailure(), "protocol_error", + "decoded TCP scalar reaches session validation") +end + +do + local transport = Net.new() + transport:handleTCPLine(Json.encode({ type = "hosted", code = "ABCDEF" })) + T.eq(transport.code, "ABCDEF", "valid relay controls stay transport-owned") + transport:handleTCPLine(Json.encode({ type = "hello", name = "RED" })) + T.eq(transport:poll()[1].name, "RED", "valid application packet stays intact") +end + T.finish("link_session")