Harden multiplayer session terminal handling

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))

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

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")