diff --git a/dev/test_api_client_fifo.lua b/dev/test_api_client_fifo.lua new file mode 100644 index 0000000..c833b0a --- /dev/null +++ b/dev/test_api_client_fifo.lua @@ -0,0 +1,145 @@ +-- Standalone regression test for networking/api_client: concurrent (overlapping) +-- HTTP requests must each receive their OWN response. +-- +-- Bug: the client kept a single pending_callback + single on_http_response slot. +-- Two requests in flight at once (e.g. "Leave Queue & Continue" firing +-- leave_matchmaking_queue then join_lobby in one tick) clobbered each other: the +-- second overwrote the first's handler, so the first response ran the wrong +-- parser ("Failed to parse server response") and the second response was dropped. +-- +-- Fix: a FIFO queue of handlers + a persistent router. The worker thread runs +-- requests sequentially and returns responses in order, so popping the front +-- matches each response to its request. +-- +-- Run: luajit dev/test_api_client_fifo.lua + +-- ── Stubs to load the real client + method files ──────────────────────────── +local warns = {} +MPAPI = { + networking = {}, + make_error = function(kind, message) return { kind = kind, message = message } end, + ErrorKind = { SERVER = 'SERVER', TRANSPORT = 'TRANSPORT', AUTH_FAILED = 'AUTH_FAILED', NOT_CONNECTED = 'NOT_CONNECTED' }, + sendWarnMessage = function(msg) warns[#warns + 1] = msg end, +} + +local LEAVE_BODY = 'LEAVE_RESPONSE' +local JOIN_BODY = 'JOIN_RESPONSE' +json = { + encode = function(_) return '{}' end, + decode = function(s) + if s == LEAVE_BODY then return { left = true } end + if s == JOIN_BODY then return { token = 'jwt-xyz', lobby = { code = 'ABC' } } end + return nil + end, +} + +dofile('networking/api_client/client.lua') -- defines MPAPI.networking.api_client +dofile('networking/api_client/matchmaking.lua') -- adds leave_matchmaking_queue +dofile('networking/api_client/lobby.lua') -- adds join_lobby +local AC = MPAPI.networking.api_client + +local function make_fake_mqtt() + local m = { tx_channel = true, sent = {} } + local function rec(url) m.sent[#m.sent + 1] = url end + m.http_post_auth = function(_self, url) rec(url) end + m.http_delete_with_body_auth = function(_self, url) rec(url) end + m.http_get_auth = function(_self, url) rec(url) end + m.http_put_auth = function(_self, url) rec(url) end + m.http_delete_auth = function(_self, url) rec(url) end + return m +end + +-- ── Harness ───────────────────────────────────────────────────────────────── +local failures = 0 +local function check(cond, msg) + if cond then print('PASS: ' .. msg) else failures = failures + 1; print('FAIL: ' .. msg) end +end + +-- ── FIXED: two overlapping requests, responses delivered in order ─────────── +print() +print('-- fixed: overlapping leave + join each get their own response --') +local client = AC.new(make_fake_mqtt(), 'http://x') +local leave_res, join_res +client:leave_matchmaking_queue('tok', {}, function(err, data) leave_res = { err = err, data = data } end) +client:join_lobby('tok', 'ABC', function(err, data) join_res = { err = err, data = data } end) + +check(#client._queue == 2, 'both requests enqueued (neither clobbered the other)') + +-- Worker returns responses in request order: leave first, then join. +client.mqtt.on_http_response(200, LEAVE_BODY) +client.mqtt.on_http_response(200, JOIN_BODY) + +check(leave_res ~= nil and leave_res.err == nil and leave_res.data and leave_res.data.left == true, + 'leave callback received the LEAVE response') +check(join_res ~= nil and join_res.err == nil and join_res.data and join_res.data.token == 'jwt-xyz', + 'join callback received the JOIN response') +check(#client._queue == 0, 'queue drained after both responses') + +-- ── FIXED: an error event routes to the right (front) request ─────────────── +print() +print('-- fixed: an http error pops the front request only --') +local client2 = AC.new(make_fake_mqtt(), 'http://x') +local a_res, b_res +client2:leave_matchmaking_queue('tok', {}, function(err) a_res = err end) +client2:join_lobby('tok', 'ABC', function(err, data) b_res = { err = err, data = data } end) +client2.mqtt.on_http_error('boom') -- first request fails +client2.mqtt.on_http_response(200, JOIN_BODY) -- second still succeeds +check(a_res ~= nil and a_res.kind == 'TRANSPORT', 'first request got the transport error') +check(b_res ~= nil and b_res.err == nil and b_res.data.token == 'jwt-xyz', 'second request still got its own success') + +-- ── RED control: single-slot design (transcribed) drops/misroutes ─────────── +-- Reproduces the pre-fix behaviour: one pending_callback + one self-clearing +-- on_http_response. The second request overwrites the first's handler. +print() +print('-- control: single-slot design misroutes overlapping requests --') +local old = { mqtt = make_fake_mqtt() } +local function old_setup_json(cb) + old.pending = cb + old.mqtt.on_http_response = function(status, body) + old.mqtt.on_http_response = nil + local c = old.pending; old.pending = nil + if not c then return end + local ok, data = pcall(json.decode, body) + if not ok or not data then c({ kind = 'TRANSPORT' }, nil); return end + c(nil, data) + end +end +local function old_setup_http(cb) -- token-required (join) + old.pending = cb + old.mqtt.on_http_response = function(status, body) + old.mqtt.on_http_response = nil + local c = old.pending; old.pending = nil + if not c then return end + local ok, data = pcall(json.decode, body) + if not ok or not data then c({ kind = 'TRANSPORT' }, nil); return end + if not data.token then c({ kind = 'AUTH_FAILED' }, nil); return end + c(nil, data) + end +end +local old_leave, old_join +old_setup_json(function(err, data) old_leave = { err = err, data = data } end) -- request 1 +old_setup_http(function(err, data) old_join = { err = err, data = data } end) -- request 2 overwrites handler +old.mqtt.on_http_response(200, LEAVE_BODY) -- leave's response arrives first... +if old.mqtt.on_http_response then old.mqtt.on_http_response(200, JOIN_BODY) end -- ...join's response dropped (handler nil) +check(old_leave == nil, 'control: leave callback NEVER fired (its handler was overwritten)') +check(old_join ~= nil and old_join.err and old_join.err.kind == 'AUTH_FAILED', + 'control: join callback got LEAVE body (no token) -> spurious error (reproduces the bug)') + +-- ── Tripwire: a response with an empty queue warns (desync canary) ────────── +-- Should be impossible given the serial worker + no-in-place-swap invariants, so +-- if it ever happens the router warns instead of silently misrouting. +print() +print('-- tripwire: response with no pending request warns --') +local client3 = AC.new(make_fake_mqtt(), 'http://x') +warns = {} +client3:leave_matchmaking_queue('tok', {}, function() end) +client3.mqtt.on_http_response(200, LEAVE_BODY) -- normal: pops the one entry +check(#warns == 0, 'no warn while a request was pending') +client3.mqtt.on_http_response(200, LEAVE_BODY) -- extra response, queue now empty +check(#warns == 1, 'warned on the response that had nothing pending') +client3.mqtt.on_http_error('late') -- error with empty queue also warns +check(#warns == 2, 'warned on an error with nothing pending too') + +-- ── Summary ───────────────────────────────────────────────────────────────── +print() +if failures == 0 then print('ALL TESTS PASSED'); os.exit(0) else print(failures .. ' TEST(S) FAILED'); os.exit(1) end diff --git a/networking/api_client/account.lua b/networking/api_client/account.lua index 235bcb5..14b7c0b 100644 --- a/networking/api_client/account.lua +++ b/networking/api_client/account.lua @@ -6,45 +6,27 @@ function api_client:get_discord_link_url(jwt_token, callback) return end - self.pending_callback = callback - - self.mqtt.on_http_response = function(status, body) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if not cb then - return - end - + self:_enqueue(function(status, body) if status ~= 200 then - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, 'Server returned status ' .. tostring(status) .. ': ' .. body), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.SERVER, 'Server returned status ' .. tostring(status) .. ': ' .. body), nil) return end local ok, data = pcall(api_client.json_decode, body) if not ok or not data then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) return end if not data.url then - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, data.error or 'Server response missing URL'), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.SERVER, data.error or 'Server response missing URL'), nil) return end - cb(nil, data) - end - - self.mqtt.on_http_error = function(msg) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if cb then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) - end - end + callback(nil, data) + end, function(msg) + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) + end) self.mqtt:http_post_auth(self.base_url .. '/api/auth/link/discord', '{}', jwt_token) end @@ -55,40 +37,7 @@ function api_client:unlink_discord(jwt_token, callback) return end - self.pending_callback = callback - - self.mqtt.on_http_response = function(status, body) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if not cb then - return - end - - if status ~= 200 then - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, 'Server returned status ' .. tostring(status) .. ': ' .. body), nil) - return - end - - local ok, data = pcall(api_client.json_decode, body) - if not ok or not data then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) - return - end - - cb(nil, data) - end - - self.mqtt.on_http_error = function(msg) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if cb then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) - end - end + self:_setup_json_callback(callback) self.mqtt:http_post_auth(self.base_url .. '/api/auth/unlink/discord', '{}', jwt_token) end diff --git a/networking/api_client/client.lua b/networking/api_client/client.lua index 1f038e3..aacb7ea 100644 --- a/networking/api_client/client.lua +++ b/networking/api_client/client.lua @@ -4,7 +4,16 @@ function api_client.new(mqtt_client, base_url) local self = { mqtt = mqtt_client, base_url = base_url, - pending_callback = nil, + -- FIFO of pending response handlers, one per in-flight request. The worker + -- thread (mqtt_thread.lua) drains tx_channel in order and runs each HTTP + -- request synchronously, so http_response/http_error events come back in the + -- same order the requests were sent -- popping the front here matches each + -- response to its request. A single shared slot (the old design) meant two + -- overlapping requests clobbered each other: the second overwrote the first's + -- handler, so the first response ran the wrong parser ("Failed to parse + -- server response") and the second response was dropped. That surfaced as e.g. + -- "Leave Queue & Continue" firing leave + join in one tick and the join failing. + _queue = {}, } setmetatable(self, { __index = api_client }) return self @@ -32,95 +41,110 @@ function api_client:_transport_ready() return self.mqtt and self.mqtt.tx_channel end --- Set up HTTP response/error handlers that parse JSON and invoke callback(err, data), --- requiring a `token` field in the body (used by the auth endpoints). -function api_client:_setup_http_callback(callback) - self.pending_callback = callback - +-- Install the persistent response router on the current mqtt transport. Each +-- inbound event pops the oldest pending handler (FIFO) and dispatches to it. +-- +-- Positional FIFO matching is correct because of three invariants, all true today: +-- * ordering -- the worker (mqtt_thread.lua) runs requests serially and blocks on +-- each, so responses come back in send order; the front entry is the answer. +-- * one-event-per-request -- every handle_http_* pushes exactly one response OR +-- error (request_with_retry is bounded), so each request pops exactly one entry. +-- * no in-place transport swap -- self.mqtt is set once in new(); MPAPI.reconnect +-- rebuilds a fresh api_client with a new empty _queue rather than re-pointing an +-- existing client's transport, so the queue never outlives its transport. +-- If any of those is ever broken (an in-place reconnect, a concurrent/pipelined +-- worker, a send that skips _enqueue), the queue desyncs and every later response +-- misroutes. The empty-queue branch below is a tripwire: a response with nothing +-- pending should be impossible, so warn loudly instead of failing silently. +function api_client:_install_router() self.mqtt.on_http_response = function(status, body) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if not cb then - return + local entry = table.remove(self._queue, 1) + if entry then + entry.on_response(status, body) + else + MPAPI.sendWarnMessage('api_client: http_response with no pending request -- response-queue desync') + end + end + self.mqtt.on_http_error = function(msg) + local entry = table.remove(self._queue, 1) + if entry then + entry.on_error(msg) + else + MPAPI.sendWarnMessage('api_client: http_error with no pending request -- response-queue desync') end + end +end + +-- Enqueue a pending request's handlers. Call this immediately before sending the +-- request on the transport, so queue order matches send order. +function api_client:_enqueue(on_response, on_error) + self:_install_router() + self._queue[#self._queue + 1] = { on_response = on_response, on_error = on_error } +end +-- Response/error handlers that parse JSON and invoke callback(err, data), requiring +-- a `token` field in the body (used by the auth endpoints). +function api_client:_setup_http_callback(callback) + self:_enqueue(function(status, body) + if not callback then return end if status < 200 or status >= 300 then - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, 'Server returned status ' .. tostring(status) .. ': ' .. body), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.SERVER, 'Server returned status ' .. tostring(status) .. ': ' .. body), nil) return end local ok, data = pcall(api_client.json_decode, body) - if not ok or not data then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) return end if not data.token then - cb(MPAPI.make_error(MPAPI.ErrorKind.AUTH_FAILED, data.error or 'Server response missing token'), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.AUTH_FAILED, data.error or 'Server response missing token'), nil) return end - cb(nil, data) - end - - self.mqtt.on_http_error = function(msg) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if cb then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) + callback(nil, data) + end, function(msg) + if callback then + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) end - end + end) end -- Generic JSON-only response handler (no token field required) function api_client:_setup_json_callback(callback) - self.pending_callback = callback - - self.mqtt.on_http_response = function(status, body) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if not cb then return end + self:_enqueue(function(status, body) + if not callback then return end if status == 204 then - cb(nil, nil) + callback(nil, nil) return end if status < 200 or status >= 300 then local ok, data = pcall(api_client.json_decode, body) local errMsg = (ok and data and data.error) or ('Server returned status ' .. tostring(status)) - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, errMsg), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.SERVER, errMsg), nil) return end if body == '' or body == nil then - cb(nil, nil) + callback(nil, nil) return end local ok, data = pcall(api_client.json_decode, body) if not ok or not data then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) return end - cb(nil, data) - end - - self.mqtt.on_http_error = function(msg) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if cb then cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) end - end + callback(nil, data) + end, function(msg) + if callback then + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) + end + end) end MPAPI.networking.api_client = api_client diff --git a/networking/api_client/lobby.lua b/networking/api_client/lobby.lua index 9d31c71..1392438 100644 --- a/networking/api_client/lobby.lua +++ b/networking/api_client/lobby.lua @@ -40,40 +40,7 @@ function api_client:set_lobby_metadata(token, code, metadata, callback) return end - self.pending_callback = callback - - self.mqtt.on_http_response = function(status, body) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if not cb then - return - end - - if status < 200 or status >= 300 then - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, 'Server returned status ' .. tostring(status) .. ': ' .. body), nil) - return - end - - local ok, data = pcall(api_client.json_decode, body) - if not ok or not data then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) - return - end - - cb(nil, data) - end - - self.mqtt.on_http_error = function(msg) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if cb then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) - end - end + self:_setup_json_callback(callback) local body = api_client.json_encode({ metadata = metadata }) self.mqtt:http_put_auth(self.base_url .. '/api/lobbies/' .. code .. '/metadata', body, token) @@ -85,41 +52,27 @@ function api_client:enable_chat(jwt_token, callback) return end - self.pending_callback = callback - - self.mqtt.on_http_response = function(status, body) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if not cb then return end - + self:_enqueue(function(status, body) if status < 200 or status >= 300 then - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, 'Server returned status ' .. tostring(status) .. ': ' .. body), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.SERVER, 'Server returned status ' .. tostring(status) .. ': ' .. body), nil) return end local ok, data = pcall(api_client.json_decode, body) if not ok or not data then - cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'Failed to parse server response'), nil) return end if data.error then - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, data.error), nil) + callback(MPAPI.make_error(MPAPI.ErrorKind.SERVER, data.error), nil) return end - cb(nil, data) - end - - self.mqtt.on_http_error = function(msg) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if cb then cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) end - end + callback(nil, data) + end, function(msg) + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) + end) self.mqtt:http_post_auth(self.base_url .. '/api/auth/chat/enable', '{}', jwt_token) end @@ -130,32 +83,18 @@ function api_client:send_chat_message(jwt_token, code, message, callback) return end - self.pending_callback = callback - - self.mqtt.on_http_response = function(status, body) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if not cb then return end - + self:_enqueue(function(status, body) if status < 200 or status >= 300 then local ok, data = pcall(api_client.json_decode, body) - local msg = (ok and data and data.error) or ('Server returned status ' .. tostring(status)) - cb(MPAPI.make_error(MPAPI.ErrorKind.SERVER, msg), nil) + local emsg = (ok and data and data.error) or ('Server returned status ' .. tostring(status)) + callback(MPAPI.make_error(MPAPI.ErrorKind.SERVER, emsg), nil) return end - cb(nil, { ok = true }) - end - - self.mqtt.on_http_error = function(msg) - self.mqtt.on_http_response = nil - self.mqtt.on_http_error = nil - local cb = self.pending_callback - self.pending_callback = nil - if cb then cb(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) end - end + callback(nil, { ok = true }) + end, function(msg) + callback(MPAPI.make_error(MPAPI.ErrorKind.TRANSPORT, 'HTTP request failed: ' .. tostring(msg)), nil) + end) local body = api_client.json_encode({ message = message }) self.mqtt:http_post_auth(self.base_url .. '/api/lobbies/' .. code .. '/chat', body, jwt_token)