From 08d2ad196f08eab590a5051ce545a5fdc59b55b7 Mon Sep 17 00:00:00 2001 From: ChronoFinale Date: Mon, 13 Jul 2026 14:22:44 -0500 Subject: [PATCH 1/2] fix(networking): FIFO-queue api_client responses so concurrent requests don't collide The api_client kept a single pending_callback and a single, self-clearing mqtt.on_http_response slot. Two requests in flight at once overwrote each other: the second request replaced the first's handler, so when the first response arrived it ran the wrong parser (surfacing as "Failed to parse server response") and the second response was dropped entirely. This broke any flow that issues two requests close together -- notably the queue-guard "Leave Queue & Continue" firing leave_matchmaking_queue and then join_lobby in the same tick, which made the join fail while a direct join (single request) worked. 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 send order. Replace the single slot with a FIFO queue of handlers plus a persistent router that pops the front on each event -- matching every response to its request. Convert the remaining hand-rolled single-slot handlers (account link/unlink, lobby set_metadata, enable_chat, send_chat_message) onto the same queue so nothing bypasses it. Adds dev/test_api_client_fifo.lua: two overlapping requests each receive their own response (success and error routing), plus a control transcribing the old single-slot design to prove it misroutes/drops (reproduces the bug). --- dev/test_api_client_fifo.lua | 128 ++++++++++++++++++++++++++++++ networking/api_client/account.lua | 69 +++------------- networking/api_client/client.lua | 104 ++++++++++++------------ networking/api_client/lobby.lua | 93 ++++------------------ 4 files changed, 207 insertions(+), 187 deletions(-) create mode 100644 dev/test_api_client_fifo.lua diff --git a/dev/test_api_client_fifo.lua b/dev/test_api_client_fifo.lua new file mode 100644 index 0000000..0a809ed --- /dev/null +++ b/dev/test_api_client_fifo.lua @@ -0,0 +1,128 @@ +-- 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 ──────────────────────────── +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' }, +} + +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)') + +-- ── 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..334f378 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,90 @@ 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. Idempotent +-- and re-applied on every enqueue so a reconnect that swaps self.mqtt still routes. +-- Each inbound event pops the oldest pending handler (FIFO) and dispatches to it. +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 - end + local entry = table.remove(self._queue, 1) + if entry then entry.on_response(status, body) end + end + self.mqtt.on_http_error = function(msg) + local entry = table.remove(self._queue, 1) + if entry then entry.on_error(msg) 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) From 2ff2289a115feb91e82d86ad0b98bfd8bce84624 Mon Sep 17 00:00:00 2001 From: ChronoFinale Date: Mon, 13 Jul 2026 17:32:01 -0500 Subject: [PATCH 2/2] chore(networking): document api_client FIFO invariants + empty-queue tripwire Add a warn when a response/error arrives with no pending request -- a canary for response-queue desync. It is impossible today (the worker runs requests serially so responses are FIFO; MPAPI.reconnect rebuilds a fresh client rather than swapping a live transport), but it warns loudly instead of silently misrouting if a future change breaks either invariant. Document the three invariants FIFO matching depends on. Test covers the tripwire firing. --- dev/test_api_client_fifo.lua | 17 +++++++++++++++++ networking/api_client/client.lua | 30 +++++++++++++++++++++++++----- 2 files changed, 42 insertions(+), 5 deletions(-) diff --git a/dev/test_api_client_fifo.lua b/dev/test_api_client_fifo.lua index 0a809ed..c833b0a 100644 --- a/dev/test_api_client_fifo.lua +++ b/dev/test_api_client_fifo.lua @@ -14,10 +14,12 @@ -- 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' @@ -123,6 +125,21 @@ check(old_leave == nil, 'control: leave callback NEVER fired (its handler was ov 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/client.lua b/networking/api_client/client.lua index 334f378..aacb7ea 100644 --- a/networking/api_client/client.lua +++ b/networking/api_client/client.lua @@ -41,17 +41,37 @@ function api_client:_transport_ready() return self.mqtt and self.mqtt.tx_channel end --- Install the persistent response router on the current mqtt transport. Idempotent --- and re-applied on every enqueue so a reconnect that swaps self.mqtt still routes. --- Each inbound event pops the oldest pending handler (FIFO) and dispatches to it. +-- 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) local entry = table.remove(self._queue, 1) - if entry then entry.on_response(status, body) end + 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) end + if entry then + entry.on_error(msg) + else + MPAPI.sendWarnMessage('api_client: http_error with no pending request -- response-queue desync') + end end end